sink: add debezium-avro protocol (#5475)

.golangci.yml

@@ -0,0 +1,167 @@
+# golangci-lint v2 configuration for TiCDC.
+# Docs: https://golangci-lint.run/usage/configuration/
+version: "2"
+
+linters:
+  # standard enables: errcheck, govet, ineffassign, staticcheck, unused.
+  default: standard
+
+  enable:
+    # bodyclose: checks HTTP response bodies are closed.
+    - bodyclose
+    # copyloopvar: detects redundant "x := x" in Go 1.22+ loop variables.
+    - copyloopvar
+    # depguard: forbids direct imports that bypass repository wrappers.
+    - depguard
+    # durationcheck: checks for suspicious duration multiplication.
+    - durationcheck
+    # errname: checks sentinel error naming (Err prefix, Error suffix).
+    - errname
+    # errorlint: checks Go 1.13 error wrapping conventions.
+    - errorlint
+    # gosec: inspects source code for security problems.
+    - gosec
+    # importas: enforces consistent import aliases.
+    - importas
+    # misspell: finds commonly misspelled English words.
+    - misspell
+    # nilerr: finds code that returns nil even if err is not nil.
+    - nilerr
+    # noctx: finds http requests without context.Context.
+    - noctx
+    # nosprintfhostport: finds misuse of Sprintf for host:port construction.
+    - nosprintfhostport
+    # prealloc: finds slices that can be preallocated.
+    - prealloc
+    # revive: drop-in replacement for golint with many rules.
+    - revive
+    # unconvert: finds unnecessary type conversions.
+    - unconvert
+    - modernize
+
+  settings:
+    depguard:
+      rules:
+        no-direct-errors:
+          files:
+            - $all
+            - "!pkg/errors/**"
+          deny:
+            - pkg: errors$
+              desc: use github.com/pingcap/ticdc/pkg/errors outside pkg/errors
+            - pkg: github.com/pingcap/errors$
+              desc: use github.com/pingcap/ticdc/pkg/errors outside pkg/errors
+
+    importas:
+      alias:
+        - pkg: github.com/pingcap/ticdc/pkg/errors
+          alias: ""
+
+    revive:
+      rules:
+        - name: blank-imports
+        - name: context-as-argument
+        - name: context-keys-type
+        - name: dot-imports
+        - name: empty-block
+        - name: error-naming
+        - name: error-return
+        - name: error-strings
+        - name: errorf
+        - name: exported
+        - name: if-return
+        - name: increment-decrement
+        - name: indent-error-flow
+        - name: range
+        - name: receiver-naming
+        - name: redefines-builtin-id
+        - name: superfluous-else
+        - name: time-naming
+        - name: unexported-return
+        - name: unreachable-code
+        - name: unused-parameter
+        - name: var-declaration
+        - name: var-naming
+
+    # G104: Audit errors not checked (duplicates errcheck).
+    # G115: integer overflow conversions are too noisy for this codebase.
+    gosec:
+      excludes:
+        - G104
+        - G115
+
+    # ST1000: don't require package comments.
+    # ST1003: don't enforce naming conventions on legacy identifiers.
+    staticcheck:
+      checks:
+        - all
+        - -ST1000
+        - -ST1003
+
+  exclusions:
+    # Strict: exclude all known generated file patterns.
+    generated: strict
+
+    # Standard presets reduce noise from common false positives.
+    # Docs: https://golangci-lint.run/usage/false-positives/
+    presets:
+      - comments
+      - common-false-positives
+
+    rules:
+      # Relax certain linters for test files. Test helpers that are
+      # intentionally unused / unchecked / low-context are expected.
+      - path: _test\.go
+        linters:
+          - errcheck
+          - gosec
+          - noctx
+          - prealloc
+          - unused
+          - revive
+          - ineffassign
+
+      # Exclude the integration test directory entirely.
+      # Replaces golangci-lint v1's --exclude-dirs "^tests/".
+      - path: tests/
+        linters:
+          - all
+
+      # Exclude generated protobuf files.
+      - path: (.*\.)?(pb|pb\.gw)\.go
+        linters:
+          - all
+
+      # Exclude generated mocks.
+      - path: _mock\.go
+        linters:
+          - all
+
+      # Exclude msgp generated files.
+      - path: _gen\.go
+        linters:
+          - all
+
+formatters:
+  enable:
+    - gofmt
+
+  settings:
+    gofmt:
+      simplify: true
+
+run:
+  timeout: 10m
+  tests: true
+  concurrency: 4
+
+output:
+  formats:
+    text:
+      path: stdout
+      print-linter-name: true
+      colors: true
+  sort-order:
+    - linter
+    - severity
+    - file

api/v2/changefeed.go

@@ -1679,33 +1679,43 @@ func getVerifiedTables(
 	if err != nil {
 		return nil, nil, nil, err
 	}
-	if !config.IsMQScheme(scheme) {
-		return ineligibleTables, eligibleTables, allTables, nil
-	}
 
-	eventRouter, err := eventrouter.NewEventRouter(replicaConfig.Sink, topic, config.IsPulsarScheme(scheme), protocol == config.ProtocolAvro)
-	if err != nil {
+	if err := verifyTable4MQ(replicaConfig, scheme, topic, protocol, tableInfos); err != nil {
 		return nil, nil, nil, err
 	}
-	err = eventRouter.VerifyTables(tableInfos)
-	if err != nil {
-		return nil, nil, nil, err
+
+	if ctx.Err() != nil {
+		return nil, nil, nil, errors.Trace(ctx.Err())
 	}
 
-	selectors, err := columnselector.New(replicaConfig.Sink)
-	if err != nil {
-		return nil, nil, nil, err
+	return ineligibleTables, eligibleTables, allTables, nil
+}
+
+func verifyTable4MQ(
+	replicaConfig *config.ReplicaConfig,
+	scheme string,
+	topic string,
+	protocol config.Protocol,
+	tableInfos []*common.TableInfo,
+) error {
+	if !config.IsMQScheme(scheme) {
+		return nil
 	}
-	err = selectors.VerifyTables(tableInfos, eventRouter)
+
+	isAvroLike := protocol == config.ProtocolAvro || protocol == config.ProtocolDebeziumAvro
+	eventRouter, err := eventrouter.NewEventRouter(replicaConfig.Sink, topic, config.IsPulsarScheme(scheme), isAvroLike)
 	if err != nil {
-		return nil, nil, nil, err
+		return err
 	}
-
-	if ctx.Err() != nil {
-		return nil, nil, nil, errors.Trace(ctx.Err())
+	if err = eventRouter.VerifyTables(tableInfos); err != nil {
+		return err
 	}
 
-	return ineligibleTables, eligibleTables, allTables, nil
+	selectors, err := columnselector.New(replicaConfig.Sink)
+	if err != nil {
+		return err
+	}
+	return selectors.VerifyTables(tableInfos, eventRouter)
 }
 
 func GetKeyspaceValueWithDefault(c *gin.Context) string {

cmd/kafka-consumer/consumer.go

@@ -42,28 +42,48 @@ func getPartitionNum(o *option) (int32, error) {
 	}
 	defer admin.Close()
 
+	topics := strings.Split(o.topic, ",")
+	maxPartitionNum := int32(0)
 	timeout := 3000
-	for i := 0; i <= 30; i++ {
-		resp, err := admin.GetMetadata(&o.topic, false, timeout)
-		if err != nil {
-			if err.(kafka.Error).Code() == kafka.ErrTransport {
-				log.Info("retry get partition number", zap.Int("retryTime", i), zap.Int("timeout", timeout))
-				timeout += 100
-				continue
+	for _, topic := range topics {
+		topic = strings.TrimSpace(topic)
+		if topic == "" {
+			continue
+		}
+		found := false
+		for i := 0; i <= 30; i++ {
+			resp, err := admin.GetMetadata(&topic, false, timeout)
+			if err != nil {
+				var kafkaErr kafka.Error
+				if errors.As(err, &kafkaErr) && kafkaErr.Code() == kafka.ErrTransport {
+					log.Info("retry get partition number", zap.String("topic", topic), zap.Int("retryTime", i), zap.Int("timeout", timeout))
+					timeout += 100
+					continue
+				}
+				return 0, errors.Trace(err)
+			}
+			if topicDetail, ok := resp.Topics[topic]; ok {
+				numPartitions := int32(len(topicDetail.Partitions))
+				log.Info("get partition number of topic",
+					zap.String("topic", topic),
+					zap.Int32("partitionNum", numPartitions))
+				if numPartitions > maxPartitionNum {
+					maxPartitionNum = numPartitions
+				}
+				found = true
+				break
 			}
-			return 0, errors.Trace(err)
+			log.Info("retry get partition number", zap.String("topic", topic))
+			time.Sleep(1 * time.Second)
 		}
-		if topicDetail, ok := resp.Topics[o.topic]; ok {
-			numPartitions := int32(len(topicDetail.Partitions))
-			log.Info("get partition number of topic",
-				zap.String("topic", o.topic),
-				zap.Int32("partitionNum", numPartitions))
-			return numPartitions, nil
+		if !found {
+			return 0, errors.Errorf("get partition number(%s) timeout", topic)
 		}
-		log.Info("retry get partition number", zap.String("topic", o.topic))
-		time.Sleep(1 * time.Second)
 	}
-	return 0, errors.Errorf("get partition number(%s) timeout", o.topic)
+	if maxPartitionNum == 0 {
+		return 0, errors.Errorf("get partition number(%s) timeout", o.topic)
+	}
+	return maxPartitionNum, nil
 }
 
 type consumer struct {

cmd/kafka-consumer/option.go

@@ -122,11 +122,11 @@ func (o *option) Adjust(upstreamURIStr string, configFile string) {
 		}
 		o.partitionNum = int32(c)
 	}
-	partitionNum, err := getPartitionNum(o)
-	if err != nil {
-		log.Panic("cannot get the partition number", zap.String("topic", o.topic), zap.Error(err))
-	}
 	if o.partitionNum == 0 {
+		partitionNum, err := getPartitionNum(o)
+		if err != nil {
+			log.Panic("cannot get the partition number", zap.String("topic", o.topic), zap.Error(err))
+		}
 		o.partitionNum = partitionNum
 	}
 

cmd/kafka-consumer/writer.go

@@ -121,7 +121,8 @@ func newWriter(ctx context.Context, o *option) *writer {
 		w.progresses[i] = newPartitionProgress(int32(i), decoder)
 	}
 
-	eventRouter, err := eventrouter.NewEventRouter(o.sinkConfig, o.topic, false, o.protocol == config.ProtocolAvro)
+	isAvroLike := o.protocol == config.ProtocolAvro || o.protocol == config.ProtocolDebeziumAvro
+	eventRouter, err := eventrouter.NewEventRouter(o.sinkConfig, o.topic, false, isAvroLike)
 	if err != nil {
 		log.Panic("initialize the event router failed",
 			zap.Any("protocol", o.protocol), zap.Any("topic", o.topic),
@@ -538,7 +539,8 @@ func (w *writer) Write(ctx context.Context, messageType common.MessageType) bool
 
 func (w *writer) onDDL(ddl *commonEvent.DDLEvent) {
 	switch w.protocol {
-	case config.ProtocolCanalJSON, config.ProtocolOpen, config.ProtocolAvro:
+	case config.ProtocolCanalJSON, config.ProtocolOpen, config.ProtocolAvro, config.ProtocolSimple,
+		config.ProtocolDebezium, config.ProtocolDebeziumAvro:
 	default:
 		return
 	}