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

@@ -1768,7 +1768,8 @@ func verifyTable4MQ(
 		return nil
 	}
 
-	eventRouter, err := eventrouter.NewEventRouter(replicaConfig.Sink, topic, config.IsPulsarScheme(scheme), protocol == config.ProtocolAvro)
+	isAvroLike := protocol == config.ProtocolAvro || protocol == config.ProtocolDebeziumAvro
+	eventRouter, err := eventrouter.NewEventRouter(replicaConfig.Sink, topic, config.IsPulsarScheme(scheme), isAvroLike)
 	if err != nil {
 		return err
 	}

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),
@@ -541,7 +542,8 @@ func (w *writer) onDDL(ddl *event.DDLEvent) {
 		return
 	}
 	switch w.protocol {
-	case config.ProtocolCanalJSON, config.ProtocolOpen, config.ProtocolAvro, config.ProtocolSimple, config.ProtocolDebezium:
+	case config.ProtocolCanalJSON, config.ProtocolOpen, config.ProtocolAvro, config.ProtocolSimple,
+		config.ProtocolDebezium, config.ProtocolDebeziumAvro:
 	default:
 		return
 	}
@@ -643,13 +645,55 @@ func (w *writer) appendRow2Group(dml *event.DMLEvent, progress *partitionProgres
 		group.Append(dml, true)
 		return
 	}
+<<<<<<< HEAD
 	group.Append(dml, false)
 	log.Info("DML event append to the group",
 		zap.Int32("partition", group.Partition), zap.Any("offset", offset),
 		zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark),
 		zap.Uint64("appliedWatermark", group.AppliedWatermark),
 		zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID),
 		zap.Stringer("eventType", dml.RowTypes[0]))
+=======
+	switch w.protocol {
+	case config.ProtocolSimple:
+		// simple protocol set the table id for all row message, it can be known which table the row message belongs to,
+		// also consider the table partition.
+		// open protocol set the partition table id if the table is partitioned.
+		// for normal table, the table id is generated by the fake table id generator by using schema and table name.
+		// so one event group for one normal table or one table partition, replayed messages can be ignored.
+		log.Warn("DML event fallback row, since less than the group high watermark, ignore it",
+			zap.Int32("partition", progress.partition), zap.Any("offset", offset),
+			zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark),
+			zap.Any("partitionWatermark", progress.watermark), zap.Any("watermarkOffset", progress.watermarkOffset),
+			zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID),
+			zap.Stringer("eventType", dml.RowTypes[0]),
+			// zap.Any("columns", row.Columns), zap.Any("preColumns", row.PreColumns),
+			zap.Any("protocol", w.protocol), zap.Bool("IsPartition", dml.TableInfo.TableName.IsPartition))
+	case config.ProtocolCanalJSON, config.ProtocolOpen, config.ProtocolAvro,
+		config.ProtocolDebezium, config.ProtocolDebeziumAvro:
+		// for partition table, these protocols cannot assign physical table id to each dml message,
+		// we cannot distinguish whether it's a real fallback event or not, still append it.
+		if w.partitionTableAccessor.IsPartitionTable(schema, table) {
+			log.Warn("DML events fallback, but the table is a partition table, still append it",
+				zap.Int32("partition", group.Partition), zap.Any("offset", offset),
+				zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark),
+				zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID),
+				zap.Stringer("eventType", dml.RowTypes[0]), zap.Any("protocol", w.protocol))
+			group.Append(dml, true)
+			return
+		}
+		log.Warn("DML event fallback row, since less than the group high watermark, ignore it",
+			zap.Int32("partition", progress.partition), zap.Any("offset", offset),
+			zap.Uint64("commitTs", commitTs), zap.Uint64("HighWatermark", group.HighWatermark),
+			zap.Any("partitionWatermark", progress.watermark), zap.Any("watermarkOffset", progress.watermarkOffset),
+			zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID),
+			zap.Stringer("eventType", dml.RowTypes[0]),
+			// zap.Any("columns", row.Columns), zap.Any("preColumns", row.PreColumns),
+			zap.Any("protocol", w.protocol), zap.Bool("IsPartition", dml.TableInfo.TableName.IsPartition))
+	default:
+		log.Panic("unknown protocol", zap.Any("protocol", w.protocol))
+	}
+>>>>>>> d220ee9b8 (sink: add debezium-avro protocol (#5475))
 }
 
 func openDB(ctx context.Context, dsn string) (*sql.DB, error) {

cmd/kafka-consumer/writer_test.go

@@ -284,6 +284,7 @@
 	require.Equal(t, "CREATE TABLE `common_1`.`a` (`a` BIGINT PRIMARY KEY,`b` INT)", w.ddlList[0].Query)
 }
 
+<<<<<<< HEAD
 func TestAppendRow2Group_DoesNotDropCommitTsFallbackBeforeApplied(t *testing.T) {
 	// Scenario:
 	// 1) TiCDC writes DML messages to Kafka in commitTs order.
@@ -339,4 +340,105 @@
 	w.appendRow2Group(newDMLEvent(1, 100), progress, kafka.Offset(12))
 	resolved = group.ResolveInto(150, resolvedEvents)
 	require.Empty(t, resolved)
+=======
+func TestOnDDLMarksRoutedCreateTableLikePartitionTableForAvro(t *testing.T) {
+	replicaCfg := config.GetDefaultReplicaConfig()
+	eventRouter, err := eventrouter.NewEventRouter(replicaCfg.Sink, "test-topic", false, true)
+	require.NoError(t, err)
+
+	w := &writer{
+		progresses:             []*partitionProgress{{partition: 0, eventsGroup: make(map[int64]*util.EventsGroup)}},
+		eventRouter:            eventRouter,
+		protocol:               config.ProtocolAvro,
+		partitionTableAccessor: codeccommon.NewPartitionTableAccessor(),
+	}
+
+	ddl := &commonEvent.DDLEvent{
+		Query:      "CREATE TABLE `target`.`dst` LIKE `target`.`src`",
+		SchemaName: "source",
+		TableName:  "dst",
+		Type:       byte(timodel.ActionCreateTable),
+		TableInfo: &common.TableInfo{
+			TableName: common.TableName{
+				Schema:       "source",
+				Table:        "dst",
+				IsPartition:  true,
+				TargetSchema: "target",
+				TargetTable:  "dst",
+			},
+		},
+	}
+	w.onDDL(ddl)
+	require.True(t, w.partitionTableAccessor.IsPartitionTable("target", "dst"))
+
+	newDMLEvent := func(commitTs uint64) *commonEvent.DMLEvent {
+		return &commonEvent.DMLEvent{
+			PhysicalTableID: 1,
+			CommitTs:        commitTs,
+			RowTypes:        []common.RowType{common.RowTypeUpdate},
+			Rows:            chunk.NewChunkWithCapacity(nil, 0),
+			TableInfo: &common.TableInfo{
+				TableName: common.TableName{Schema: "target", Table: "dst"},
+			},
+		}
+	}
+
+	progress := w.progresses[0]
+	w.appendRow2Group(newDMLEvent(200), progress, kafka.Offset(10))
+	w.appendRow2Group(newDMLEvent(100), progress, kafka.Offset(11))
+
+	resolved := progress.eventsGroup[1].ResolveInto(150, nil)
+	require.Len(t, resolved, 1)
+	require.Equal(t, uint64(100), resolved[0].CommitTs)
+}
+
+func TestAppendRow2GroupKeepsDebeziumPartitionTableFallback(t *testing.T) {
+	for _, protocol := range []config.Protocol{
+		config.ProtocolDebezium,
+		config.ProtocolDebeziumAvro,
+	} {
+		t.Run(protocol.String(), func(t *testing.T) {
+			replicaCfg := config.GetDefaultReplicaConfig()
+			eventRouter, err := eventrouter.NewEventRouter(replicaCfg.Sink, "test-topic", false, false)
+			require.NoError(t, err)
+
+			w := &writer{
+				progresses:             []*partitionProgress{{partition: 0, eventsGroup: make(map[int64]*util.EventsGroup)}},
+				eventRouter:            eventRouter,
+				protocol:               protocol,
+				partitionTableAccessor: codeccommon.NewPartitionTableAccessor(),
+			}
+
+			w.partitionTableAccessor.Add("target", "src")
+			ddl := &commonEvent.DDLEvent{
+				Query:      "CREATE TABLE `target`.`dst` LIKE `target`.`src`",
+				SchemaName: "target",
+				TableName:  "dst",
+				Type:       byte(timodel.ActionCreateTable),
+			}
+			w.onDDL(ddl)
+			require.True(t, w.partitionTableAccessor.IsPartitionTable("target", "dst"))
+
+			newDMLEvent := func(commitTs uint64) *commonEvent.DMLEvent {
+				return &commonEvent.DMLEvent{
+					PhysicalTableID: 1,
+					CommitTs:        commitTs,
+					RowTypes:        []common.RowType{common.RowTypeUpdate},
+					Rows:            chunk.NewChunkWithCapacity(nil, 0),
+					TableInfo: &common.TableInfo{
+						TableName: common.TableName{Schema: "target", Table: "dst"},
+					},
+				}
+			}
+
+			progress := w.progresses[0]
+			w.appendRow2Group(newDMLEvent(200), progress, kafka.Offset(10))
+			w.appendRow2Group(newDMLEvent(100), progress, kafka.Offset(11))
+
+			resolved := progress.eventsGroup[1].ResolveInto(150, nil)
+			require.Len(t, resolved, 1)
+			require.Equal(t, uint64(100), resolved[0].CommitTs)
+		})
+	}
+>>>>>>> d220ee9b8 (sink: add debezium-avro protocol (#5475))
 }

downstreamadapter/sink/helper/helper.go

@@ -99,7 +99,7 @@ func GetProtocol(protocolStr string) (config.Protocol, error) {
 // GetFileExtension returns the extension for specific protocol
 func GetFileExtension(protocol config.Protocol) string {
 	switch protocol {
-	case config.ProtocolAvro, config.ProtocolCanalJSON, config.ProtocolMaxwell,
+	case config.ProtocolAvro, config.ProtocolDebeziumAvro, config.ProtocolCanalJSON, config.ProtocolMaxwell,
 		config.ProtocolOpen, config.ProtocolSimple:
 		return ".json"
 	case config.ProtocolCraft:

downstreamadapter/sink/kafka/helper.go

@@ -90,7 +90,7 @@ func newKafkaSinkComponent(
 		return comp, protocol, err
 	}
 
-	isAvroLike := protocol == config.ProtocolAvro
+	isAvroLike := protocol == config.ProtocolAvro || protocol == config.ProtocolDebeziumAvro
 	comp.eventRouter, err = eventrouter.NewEventRouter(
 		sinkConfig, topic, false, isAvroLike)
 	if err != nil {

pkg/config/changefeed.go

@@ -473,8 +473,9 @@ func (info *ChangeFeedInfo) RmUnusedFields() {
 		info.rmMQOnlyFields()
 	} else {
 		// remove schema registry for MQ downstream with
-		// protocol other than avro
-		if util.GetOrZero(info.Config.Sink.Protocol) != ProtocolAvro.String() {
+		// protocol other than avro or debezium-avro
+		protocol := util.GetOrZero(info.Config.Sink.Protocol)
+		if protocol != ProtocolAvro.String() && protocol != ProtocolDebeziumAvro.String() {
 			info.Config.Sink.SchemaRegistry = nil
 		}
 	}