codec(ticdc): canal-json message do not end by terminator if using kafka sink (#11708)

close #11707

Author: 3AceShowHand

cdc/sink/codec/canal/canal_json_encoder.go

@@ -349,7 +349,8 @@ func (c *JSONBatchEncoder) AppendRowChangedEvent(
 	if err != nil {
 		return errors.Trace(err)
 	}
-	if len(c.config.Terminator) > 0 {
+
+	if c.config.IsStorageScheme && len(c.config.Terminator) > 0 {
 		value = append(value, c.config.Terminator...)
 	}
 

cdc/sink/codec/common/config.go

@@ -21,6 +21,7 @@ import (
 	"github.com/pingcap/log"
 	"github.com/pingcap/tiflow/pkg/config"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
+	"github.com/pingcap/tiflow/pkg/sink"
 	"go.uber.org/zap"
 )
 
@@ -47,6 +48,7 @@ type Config struct {
 	AvroBigintUnsignedHandlingMode string
 
 	// for sinking to cloud storage
+	IsStorageScheme      bool
 	Delimiter            string
 	Quote                string
 	NullString           string
@@ -162,6 +164,7 @@ func (c *Config) Apply(sinkURI *url.URL, config *config.ReplicaConfig) error {
 
 	}
 
+	c.IsStorageScheme = sink.IsStorageScheme(sinkURI.Scheme)
 	c.DeleteOnlyHandleKeyColumns = !config.EnableOldValue
 	return nil
 }