cmd (ticdc): add pulsar consumer (#9730)

ref #9413

Makefile

@@ -163,6 +163,9 @@ kafka_consumer:
 storage_consumer:
 	$(GOBUILD) -ldflags '$(LDFLAGS)' -o bin/cdc_storage_consumer ./cmd/storage-consumer/main.go
 
+pulsar_consumer:
+	$(GOBUILD) -ldflags '$(LDFLAGS)' -o bin/cdc_pulsar_consumer ./cmd/pulsar-consumer/main.go
+
 kafka_consumer_image: 
 	@which docker || (echo "docker not found in ${PATH}"; exit 1)
 	DOCKER_BUILDKIT=1 docker build -f ./deployments/ticdc/docker/kafka-consumer.Dockerfile . -t ticdc:kafka-consumer  --platform linux/amd64
@@ -171,6 +174,10 @@ storage_consumer_image:
 	@which docker || (echo "docker not found in ${PATH}"; exit 1)
 	DOCKER_BUILDKIT=1 docker build -f ./deployments/ticdc/docker/storage-consumer.Dockerfile . -t ticdc:storage-consumer  --platform linux/amd64
 
+pulsar_consumer_image: 
+	@which docker || (echo "docker not found in ${PATH}"; exit 1)
+	DOCKER_BUILDKIT=1 docker build -f ./deployments/ticdc/docker/pulsar-consumer.Dockerfile . -t ticdc:pulsar-consumer  --platform linux/amd64
+
 filter_helper:
 	$(GOBUILD) -ldflags '$(LDFLAGS)' -o bin/cdc_filter_helper ./cmd/filter-helper/main.go
 
@@ -215,7 +222,7 @@ check_third_party_binary:
 	@which bin/minio
 	@which bin/bin/schema-registry-start
 
-integration_test_build: check_failpoint_ctl storage_consumer kafka_consumer
+integration_test_build: check_failpoint_ctl storage_consumer kafka_consumer pulsar_consumer
 	$(FAILPOINT_ENABLE)
 	$(GOTEST) -ldflags '$(LDFLAGS)' -c -cover -covermode=atomic \
 		-coverpkg=github.com/pingcap/tiflow/... \

cdc/sink/ddlsink/mq/ddlproducer/pulsar_ddl_producer.go

@@ -78,6 +78,12 @@ func (p *pulsarProducers) SyncSendMessage(ctx context.Context, topic string,
 		return err
 	}
 
+	if message.Type == model.MessageTypeDDL {
+		log.Info("pulsarProducers SyncSendMessage success",
+			zap.Any("mID", mID), zap.String("topic", topic),
+			zap.String("ddl", string(message.Value)))
+	}
+
 	log.Debug("pulsarProducers SyncSendMessage success",
 		zap.Any("mID", mID), zap.String("topic", topic))
 

cdc/sink/ddlsink/mq/mq_ddl_sink.go

@@ -95,7 +95,8 @@ func (k *DDLSink) WriteDDLEvent(ctx context.Context, ddl *model.DDLEvent) error
 
 	topic := k.eventRouter.GetTopicForDDL(ddl)
 	partitionRule := k.eventRouter.GetDLLDispatchRuleByProtocol(k.protocol)
-	log.Debug("Emit ddl event",
+	log.Info("Emit ddl event",
+		zap.String("topic", topic),
 		zap.Uint64("commitTs", ddl.CommitTs),
 		zap.String("query", ddl.Query),
 		zap.String("namespace", k.id.Namespace),

cdc/sink/dmlsink/mq/dispatcher/event_router.go

@@ -22,6 +22,7 @@ import (
 	"github.com/pingcap/tiflow/cdc/sink/dmlsink/mq/dispatcher/partition"
 	"github.com/pingcap/tiflow/cdc/sink/dmlsink/mq/dispatcher/topic"
 	"github.com/pingcap/tiflow/pkg/config"
+	"github.com/pingcap/tiflow/pkg/errors"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
 	"github.com/pingcap/tiflow/pkg/sink"
 	"github.com/pingcap/tiflow/pkg/util"
@@ -267,29 +268,30 @@ func getTopicDispatcher(
 	if sink.IsPulsarScheme(schema) {
 		err := topicExpr.PulsarValidate()
 		if err != nil {
-			return nil, err
-		}
-	}
-
-	// validate the topic expression for kafka sink
-	var p config.Protocol
-	var err error
-	if protocol != "" {
-		p, err = config.ParseSinkProtocolFromString(protocol)
-		if err != nil {
-			return nil, cerror.WrapError(cerror.ErrKafkaInvalidConfig, err)
-		}
-	}
-	if p == config.ProtocolAvro {
-		err = topicExpr.ValidateForAvro()
-		if err != nil {
-			return nil, err
+			return nil, errors.Trace(err)
 		}
 	} else {
-		err = topicExpr.Validate()
-		if err != nil {
-			return nil, err
+		// validate the topic expression for kafka sink
+		var p config.Protocol
+		var err error
+		if protocol != "" {
+			p, err = config.ParseSinkProtocolFromString(protocol)
+			if err != nil {
+				return nil, cerror.WrapError(cerror.ErrKafkaInvalidConfig, err)
+			}
+		}
+		if p == config.ProtocolAvro {
+			err = topicExpr.ValidateForAvro()
+			if err != nil {
+				return nil, err
+			}
+		} else {
+			err = topicExpr.Validate()
+			if err != nil {
+				return nil, err
+			}
 		}
 	}
+
 	return topic.NewDynamicTopicDispatcher(topicExpr), nil
 }

cdc/sink/dmlsink/mq/pulsar_dml_sink.go

@@ -59,6 +59,11 @@ func NewPulsarDMLSink(
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
+	if !util.IsPulsarSupportedProtocols(protocol) {
+		return nil, cerror.ErrSinkURIInvalid.
+			GenWithStackByArgs("unsupported protocol, " +
+				"pulsar sink currently only support these protocols: [canal-json, canal, maxwell]")
+	}
 
 	pConfig, err := pulsarConfig.NewPulsarConfig(sinkURI, replicaConfig.Sink.PulsarConfig)
 	if err != nil {

cdc/sink/util/helper.go

@@ -106,3 +106,8 @@ func GetTopicManagerAndTryCreateTopic(
 
 	return topicManager, nil
 }
+
+// IsPulsarSupportedProtocols returns whether the protocol is supported by pulsar.
+func IsPulsarSupportedProtocols(p config.Protocol) bool {
+	return p == config.ProtocolCanalJSON
+}