From 90947b632ff7d9dac0c7d610509731800707c186 Mon Sep 17 00:00:00 2001 From: Ziqian Qin Date: Fri, 7 Aug 2026 11:41:44 +0800 Subject: [PATCH 1/4] debezium: add debezium-include-start-ts to emit source.start_ts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a new changefeed-level option debezium-include-start-ts (sink URI parameter or changefeed config, default false). When enabled, the Debezium JSON value message carries source.start_ts — the start TSO of the transaction that made the change — and the source schema declares it, so schema-validated consumers (e.g. Kafka Connect) can see the field without enable-tidb-extension. The option only takes effect with protocol debezium; the Avro protocol is unaffected (its payload does not carry the field and its schema does not declare it). Explicit sink URI values (including false) override the config file. - EncodeValue / writeSourceSchema: emit and declare start_ts under debezium-include-start-ts (JSON only, guarded for Avro). - decoder: read source.start_ts as the decoded event's StartTs, falling back to commit_ts for messages without the field (pre-feature format), preserving old behavior. - config: add the option (URI param + changefeed config) with protocol validation; apply explicit URI values after the mergo merge because mergo cannot override a *bool true with false. - tests: encode/schema/round-trip coverage, pre-feature fallback coverage, Avro negative coverage, and config parse/precedence/ validation coverage. Existing golden tests pass unchanged (default config does not emit the field). --- api/v2/changefeed_toml_test.go | 12 ++- api/v2/model.go | 7 ++ pkg/config/sink.go | 3 + pkg/sink/codec/common/config.go | 26 ++++++- pkg/sink/codec/common/config_test.go | 36 +++++++++ pkg/sink/codec/debezium/avro_test.go | 6 ++ pkg/sink/codec/debezium/codec.go | 15 ++++ pkg/sink/codec/debezium/debezium_test.go | 95 ++++++++++++++++++++++++ pkg/sink/codec/debezium/decoder.go | 24 +++++- 9 files changed, 218 insertions(+), 6 deletions(-) diff --git a/api/v2/changefeed_toml_test.go b/api/v2/changefeed_toml_test.go index 4a8539ef4a..7ed1d4a31a 100644 --- a/api/v2/changefeed_toml_test.go +++ b/api/v2/changefeed_toml_test.go @@ -88,10 +88,13 @@ func TestChangeFeedInfoTOMLRoundTripToInternal(t *testing.T) { SinkURI: "blackhole://", StartTs: 449999999999999999, Config: &ReplicaConfig{ - MemoryQuota: util.AddressOf(uint64(1024)), - CaseSensitive: util.AddressOf(true), - ForceReplicate: util.AddressOf(true), - CheckGCSafePoint: util.AddressOf(false), + MemoryQuota: util.AddressOf(uint64(1024)), + CaseSensitive: util.AddressOf(true), + ForceReplicate: util.AddressOf(true), + CheckGCSafePoint: util.AddressOf(false), + Sink: &SinkConfig{ + DebeziumIncludeStartTs: util.AddressOf(true), + }, SyncPointInterval: &JSONDuration{duration: 10 * time.Minute}, Integrity: &IntegrityConfig{ IntegrityCheckLevel: util.AddressOf("correctness"), @@ -113,6 +116,7 @@ func TestChangeFeedInfoTOMLRoundTripToInternal(t *testing.T) { // Top-level kebab-case keys and runtime field omissions. require.Contains(t, out, `sink-uri = "blackhole://"`) require.Contains(t, out, "start-ts") + require.Contains(t, out, "debezium-include-start-ts") require.NotContains(t, out, "gid") // GID is omitted from TOML (toml:"-") // The [config] section must decode into the internal ReplicaConfig used by diff --git a/api/v2/model.go b/api/v2/model.go index 0575a73e23..4e5290d757 100644 --- a/api/v2/model.go +++ b/api/v2/model.go @@ -580,6 +580,9 @@ func (c *ReplicaConfig) toInternalReplicaConfigWithOriginConfig( if c.Sink.DebeziumDisableSchema != nil { res.Sink.DebeziumDisableSchema = util.AddressOf(*c.Sink.DebeziumDisableSchema) } + if c.Sink.DebeziumIncludeStartTs != nil { + res.Sink.DebeziumIncludeStartTs = util.AddressOf(*c.Sink.DebeziumIncludeStartTs) + } if c.Sink.SendBootstrapIntervalInSec != nil { res.Sink.SendBootstrapIntervalInSec = util.AddressOf(*c.Sink.SendBootstrapIntervalInSec) @@ -959,6 +962,9 @@ func ToAPIReplicaConfig(c *config.ReplicaConfig) *ReplicaConfig { if cloned.Sink.DebeziumDisableSchema != nil { res.Sink.DebeziumDisableSchema = util.AddressOf(*cloned.Sink.DebeziumDisableSchema) } + if cloned.Sink.DebeziumIncludeStartTs != nil { + res.Sink.DebeziumIncludeStartTs = util.AddressOf(*cloned.Sink.DebeziumIncludeStartTs) + } } if cloned.Consistent != nil { if res.Consistent == nil { @@ -1198,6 +1204,7 @@ type SinkConfig struct { SendBootstrapToAllPartition *bool `json:"send_bootstrap_to_all_partition,omitempty" toml:"send-bootstrap-to-all-partition,omitempty"` SendAllBootstrapAtStart *bool `json:"send_all_bootstrap_at_start,omitempty" toml:"send-all-bootstrap-at-start,omitempty"` DebeziumDisableSchema *bool `json:"debezium_disable_schema,omitempty" toml:"debezium-disable-schema,omitempty"` + DebeziumIncludeStartTs *bool `json:"debezium_include_start_ts,omitempty" toml:"debezium-include-start-ts,omitempty"` DebeziumConfig *DebeziumConfig `json:"debezium,omitempty" toml:"debezium,omitempty"` OpenProtocolConfig *OpenProtocolConfig `json:"open,omitempty" toml:"open,omitempty"` } diff --git a/pkg/config/sink.go b/pkg/config/sink.go index dca4d50572..9d6779508d 100644 --- a/pkg/config/sink.go +++ b/pkg/config/sink.go @@ -200,6 +200,9 @@ type SinkConfig struct { SendAllBootstrapAtStart *bool `toml:"send-all-bootstrap-at-start" json:"send-all-bootstrap-at-start,omitempty"` // Debezium only. Whether schema should be excluded in the output. DebeziumDisableSchema *bool `toml:"debezium-disable-schema" json:"debezium-disable-schema,omitempty"` + // Debezium only. Whether the transaction start_ts should be included in + // the source block of the output. JSON protocol only. + DebeziumIncludeStartTs *bool `toml:"debezium-include-start-ts" json:"debezium-include-start-ts,omitempty"` // CSVConfig is only available when the downstream is Storage. CSVConfig *CSVConfig `toml:"csv" json:"csv,omitempty"` diff --git a/pkg/sink/codec/common/config.go b/pkg/sink/codec/common/config.go index 4f5cee429d..dadb8a9c87 100644 --- a/pkg/sink/codec/common/config.go +++ b/pkg/sink/codec/common/config.go @@ -98,6 +98,9 @@ type Config struct { DebeziumDisableSchema bool // Debezium only. Whether before value should be included in the output. DebeziumOutputOldValue bool + // Debezium only. Whether the transaction start_ts should be included in + // the source block of the output. JSON protocol only. + DebeziumIncludeStartTs bool // CSV only. Whether header should be included in the output. CSVOutputFieldHeader bool } @@ -143,6 +146,7 @@ func NewConfig(protocol config.Protocol) *Config { DebeziumOutputOldValue: true, OpenOutputOldValue: true, DebeziumDisableSchema: false, + DebeziumIncludeStartTs: false, CSVOutputFieldHeader: false, } } @@ -182,7 +186,8 @@ type urlConfig struct { OnlyOutputUpdatedColumns *bool `form:"only-output-updated-columns"` ContentCompatible *bool `form:"content-compatible"` - DebeziumDisableSchema *bool `form:"debezium-disable-schema"` + DebeziumDisableSchema *bool `form:"debezium-disable-schema"` + DebeziumIncludeStartTs *bool `form:"debezium-include-start-ts"` // EncodingFormatType is only works for the simple protocol, // can be `json` and `avro`, default to `json`. EncodingFormatType *string `form:"encoding-format"` @@ -200,6 +205,10 @@ func (c *Config) Apply(sinkURI *url.URL, sinkConfig *config.SinkConfig) error { if err = binding.Query.Bind(req, urlParameter); err != nil { return errors.WrapError(errors.ErrSinkInvalidConfig, err) } + // Keep the raw URI parameters: mergeConfig uses mergo, which cannot + // override a *bool "true" (from the config file) with an explicit + // "false" from the sink URI, so explicit URI values are applied last. + rawURLParameter := urlParameter if urlParameter, err = mergeConfig(sinkConfig, urlParameter); err != nil { return err } @@ -307,6 +316,12 @@ func (c *Config) Apply(sinkURI *url.URL, sinkConfig *config.SinkConfig) error { if urlParameter.DebeziumDisableSchema != nil { c.DebeziumDisableSchema = *urlParameter.DebeziumDisableSchema } + if urlParameter.DebeziumIncludeStartTs != nil { + c.DebeziumIncludeStartTs = *urlParameter.DebeziumIncludeStartTs + } + if rawURLParameter.DebeziumIncludeStartTs != nil { + c.DebeziumIncludeStartTs = *rawURLParameter.DebeziumIncludeStartTs + } return nil } @@ -338,6 +353,9 @@ func mergeConfig( if sinkConfig.DebeziumDisableSchema != nil { dest.DebeziumDisableSchema = sinkConfig.DebeziumDisableSchema } + if sinkConfig.DebeziumIncludeStartTs != nil { + dest.DebeziumIncludeStartTs = sinkConfig.DebeziumIncludeStartTs + } } if err := mergo.Merge(dest, urlParameters, mergo.WithOverride); err != nil { return nil, err @@ -381,6 +399,12 @@ func (c *Config) Validate() error { ) } + if c.DebeziumIncludeStartTs && c.Protocol != config.ProtocolDebezium { + return errors.ErrCodecInvalidConfig.GenWithStack( + `debezium-include-start-ts only takes effect with protocol "debezium"`, + ) + } + if c.Protocol == config.ProtocolAvro || c.Protocol == config.ProtocolDebeziumAvro { if c.AvroConfluentSchemaRegistry != "" && c.AvroGlueSchemaRegistry != nil { protocol := "Avro" diff --git a/pkg/sink/codec/common/config_test.go b/pkg/sink/codec/common/config_test.go index 4369de7216..a0ad34a782 100644 --- a/pkg/sink/codec/common/config_test.go +++ b/pkg/sink/codec/common/config_test.go @@ -158,3 +158,39 @@ func TestDebeziumAvroWatermarkConfig(t *testing.T) { require.True(t, cfg.AvroEnableWatermark) require.Equal(t, "http://127.0.0.1:8081", cfg.AvroConfluentSchemaRegistry) } + +func TestDebeziumIncludeStartTsConfig(t *testing.T) { + // URI parameter + cfg := NewConfig(config.ProtocolDebezium) + sinkURI, err := url.Parse("kafka://127.0.0.1:9092/topic?protocol=debezium&debezium-include-start-ts=true") + require.NoError(t, err) + require.NoError(t, cfg.Apply(sinkURI, config.GetDefaultReplicaConfig().Sink)) + require.True(t, cfg.DebeziumIncludeStartTs) + require.NoError(t, cfg.Validate()) + + // changefeed config file + on := true + cfg2 := NewConfig(config.ProtocolDebezium) + sinkConfig := config.GetDefaultReplicaConfig().Sink + sinkConfig.DebeziumIncludeStartTs = &on + sinkURI2, err := url.Parse("kafka://127.0.0.1:9092/topic?protocol=debezium") + require.NoError(t, err) + require.NoError(t, cfg2.Apply(sinkURI2, sinkConfig)) + require.True(t, cfg2.DebeziumIncludeStartTs) + + // URI parameter overrides the config file + cfg3 := NewConfig(config.ProtocolDebezium) + sinkConfig3 := config.GetDefaultReplicaConfig().Sink + sinkConfig3.DebeziumIncludeStartTs = &on + sinkURI3, err := url.Parse("kafka://127.0.0.1:9092/topic?protocol=debezium&debezium-include-start-ts=false") + require.NoError(t, err) + require.NoError(t, cfg3.Apply(sinkURI3, sinkConfig3)) + require.False(t, cfg3.DebeziumIncludeStartTs) + + // only supported by the debezium (JSON) protocol + cfg4 := NewConfig(config.ProtocolDebeziumAvro) + cfg4.DebeziumIncludeStartTs = true + errCode, ok := errors.RFCCode(cfg4.Validate()) + require.True(t, ok) + require.Equal(t, errors.ErrCodecInvalidConfig.RFCCode(), errCode) +} diff --git a/pkg/sink/codec/debezium/avro_test.go b/pkg/sink/codec/debezium/avro_test.go index a61e2094ed..524084cd7d 100644 --- a/pkg/sink/codec/debezium/avro_test.go +++ b/pkg/sink/codec/debezium/avro_test.go @@ -60,6 +60,8 @@ func TestDebeziumConfluentAvroEncodeRowEvent(t *testing.T) { cfg.AvroConfluentSchemaRegistry = "http://127.0.0.1:8081" cfg.AvroBigintUnsignedHandlingMode = common.BigintUnsignedHandlingModeString cfg.DebeziumDisableSchema = true + // debezium-include-start-ts must not affect the Avro protocol. + cfg.DebeziumIncludeStartTs = true cfg.TimeZone = time.UTC encoder, err := NewAvroBatchEncoder(ctx, cfg, "dbserver1") @@ -103,6 +105,9 @@ func TestDebeziumConfluentAvroEncodeRowEvent(t *testing.T) { require.Nil(t, source["snapshot"]) require.Nil(t, source["thread"]) require.Equal(t, "dbserver1", source["name"]) + // start_ts is a JSON-protocol-only field: the Avro payload and its + // registered schema must not carry it, even with debezium-include-start-ts on. + require.NotContains(t, source, "start_ts") valueSchema := decodeConfluentAvroSchemaForTest(t, messages[0].Value) require.Contains(t, valueSchema, `"name":"fooEnvelope"`) @@ -110,6 +115,7 @@ func TestDebeziumConfluentAvroEncodeRowEvent(t *testing.T) { require.Contains(t, valueSchema, `"name":"Source"`) require.Contains(t, valueSchema, `"logicalType":"decimal"`) require.NotContains(t, valueSchema, `"field":"transaction"`) + require.NotContains(t, valueSchema, `"field":"start_ts"`) } func TestDebeziumConfluentAvroSanitizesFullNameAndUnionBranch(t *testing.T) { diff --git a/pkg/sink/codec/debezium/codec.go b/pkg/sink/codec/debezium/codec.go index 1ceea135e8..5bc7a8910c 100644 --- a/pkg/sink/codec/debezium/codec.go +++ b/pkg/sink/codec/debezium/codec.go @@ -987,6 +987,16 @@ func (c *dbzCodec) writeSourceSchema(writer *util.JSONWriter, schemaName string) writer.WriteStringField("field", "cluster_id") }) } + // start_ts is gated by debezium-include-start-ts and declared only for + // the JSON protocol: the Avro payload does not carry it, so its schema + // must not declare it either. + if c.config.DebeziumIncludeStartTs && !c.isDebeziumAvro() { + writer.WriteObjectElement(func() { + writer.WriteStringField("type", "int64") + writer.WriteBoolField("optional", false) + writer.WriteStringField("field", "start_ts") + }) + } }) writer.WriteBoolField("optional", false) writer.WriteStringField("name", c.sourceSchemaName(schemaName)) @@ -1084,6 +1094,11 @@ func (c *dbzCodec) EncodeValue( // The followings are TiDB extended fields jWriter.WriteUint64Field("commit_ts", e.CommitTs) + // start_ts: the start TSO of the transaction that made this change, + // exposed for downstream consumers that need transaction correlation. + if c.config.DebeziumIncludeStartTs { + jWriter.WriteUint64Field("start_ts", e.StartTs) + } jWriter.WriteStringField("cluster_id", c.clusterID) }) diff --git a/pkg/sink/codec/debezium/debezium_test.go b/pkg/sink/codec/debezium/debezium_test.go index 80380bfebf..5b3dcdbd92 100644 --- a/pkg/sink/codec/debezium/debezium_test.go +++ b/pkg/sink/codec/debezium/debezium_test.go @@ -14,6 +14,7 @@ package debezium import ( + "bytes" "context" "encoding/json" "os" @@ -207,3 +208,97 @@ func (s *debeziumSuite) TestDataTypes() { s.requireDebeziumJSONEq(dataDbzOutput, messages[0].Value) s.requireDebeziumJSONEq(keyDbzOutput, messages[0].Key) } + +func TestEncodeStartTsInSource(t *testing.T) { + // The field is emitted when debezium-include-start-ts is enabled, + // independent of enable-tidb-extension. + cfg := common.NewConfig(config.ProtocolDebezium) + cfg.DebeziumIncludeStartTs = true + cfg.TimeZone = time.UTC + + encoder := NewBatchEncoder(cfg, "dbserver1") + rowEvent := common.NewRoutedRowEvent4Test() + rowEvent.StartTs = 5 + require.NoError(t, encoder.AppendRowChangedEvent(context.Background(), "", rowEvent)) + + messages := encoder.Build() + require.Len(t, messages, 1) + + dec := json.NewDecoder(bytes.NewReader(messages[0].Value)) + dec.UseNumber() + var value map[string]any + require.NoError(t, dec.Decode(&value)) + payload := value["payload"].(map[string]any) + source := payload["source"].(map[string]any) + startTs, err := source["start_ts"].(json.Number).Int64() + require.NoError(t, err) + require.Equal(t, int64(5), startTs) + + // The source schema declares start_ts under the same switch, so + // schema-validated consumers can see it without enable-tidb-extension. + schema := value["schema"].(map[string]any) + sourceSchema := schemaFieldsByName(t, schema, "source") + require.NotNil(t, sourceSchema) + require.NotNil(t, schemaFieldsByName(t, sourceSchema, "start_ts")) + + // round-trip: decoding restores the true start ts. The TiCDC-side decoder + // requires enable-tidb-extension: it relies on the per-column tidb_type in + // the schema to reconstruct column types, so the encoded message must + // carry the extension fields as well. + cfg2 := common.NewConfig(config.ProtocolDebezium) + cfg2.DebeziumIncludeStartTs = true + cfg2.EnableTiDBExtension = true + cfg2.TimeZone = time.UTC + encoder2 := NewBatchEncoder(cfg2, "dbserver1") + require.NoError(t, encoder2.AppendRowChangedEvent(context.Background(), "", rowEvent)) + messages2 := encoder2.Build() + require.Len(t, messages2, 1) + + decoder := NewDecoder(cfg2, 0, nil) + decoder.AddKeyValue(messages2[0].Key, messages2[0].Value) + messageType, hasNext := decoder.HasNext() + require.True(t, hasNext) + require.Equal(t, common.MessageTypeRow, messageType) + decoded := decoder.NextDMLMessage().ToDMLEvent() + require.Equal(t, uint64(5), decoded.GetStartTs()) +} + +func TestDecodeStartTsFallbackToCommitTs(t *testing.T) { + // A message produced without debezium-include-start-ts (the pre-feature + // format) has no start_ts in the source block; decoding it must fall back + // to commit_ts, keeping the old behavior. + cfg := common.NewConfig(config.ProtocolDebezium) + cfg.EnableTiDBExtension = true // required to decode the message back + cfg.TimeZone = time.UTC + + encoder := NewBatchEncoder(cfg, "dbserver1") + rowEvent := common.NewRoutedRowEvent4Test() + rowEvent.StartTs = 5 + require.NoError(t, encoder.AppendRowChangedEvent(context.Background(), "", rowEvent)) + + messages := encoder.Build() + require.Len(t, messages, 1) + + decoder := NewDecoder(cfg, 0, nil) + decoder.AddKeyValue(messages[0].Key, messages[0].Value) + messageType, hasNext := decoder.HasNext() + require.True(t, hasNext) + require.Equal(t, common.MessageTypeRow, messageType) + decoded := decoder.NextDMLMessage().ToDMLEvent() + require.Equal(t, decoded.GetCommitTs(), decoded.GetStartTs()) + require.NotEqual(t, uint64(5), decoded.GetStartTs()) +} + +// schemaFieldsByName returns the sub-schema object of a field inside a Debezium +// struct schema, or nil when the field is not declared. +func schemaFieldsByName(t *testing.T, schema map[string]any, name string) map[string]any { + fields, ok := schema["fields"].([]any) + require.True(t, ok) + for _, f := range fields { + fm := f.(map[string]any) + if fm["field"] == name { + return fm + } + } + return nil +} diff --git a/pkg/sink/codec/debezium/decoder.go b/pkg/sink/codec/debezium/decoder.go index 5b80cb8047..9e8163743e 100644 --- a/pkg/sink/codec/debezium/decoder.go +++ b/pkg/sink/codec/debezium/decoder.go @@ -202,9 +202,15 @@ func (d *decoder) assembleDMLEventFromPayload( ) *commonEvent.DMLEvent { tableInfo := queryTableInfoFromPayload(keyPayload, valuePayload, valueSchema) commitTs := getCommitTsFromPayload(valuePayload) + startTs := getStartTsFromPayload(valuePayload) + if startTs == 0 { + // Fall back to commit_ts for messages that do not carry start_ts, + // keeping the pre-feature behavior for old messages. + startTs = commitTs + } event := &commonEvent.DMLEvent{ Rows: chunk.NewChunkFromPoolWithCapacity(tableInfo.GetFieldSlice(), chunk.InitialCapacity), - StartTs: commitTs, + StartTs: startTs, CommitTs: commitTs, TableInfo: tableInfo, PhysicalTableID: tableInfo.TableName.TableID, @@ -250,6 +256,22 @@ func getCommitTsFromPayload(valuePayload map[string]any) uint64 { return uint64(commitTs) } +// getStartTsFromPayload returns the start_ts carried in the source block, or 0 +// when the field is absent (messages produced before the start_ts field existed). +func getStartTsFromPayload(valuePayload map[string]any) uint64 { + source := valuePayload["source"].(map[string]any) + startTs, ok := source["start_ts"].(json.Number) + if !ok { + return 0 + } + ts, err := startTs.Int64() + if err != nil { + log.Error("decode value failed", zap.Error(err), zap.String("value", util.RedactAny(source))) + return 0 + } + return uint64(ts) +} + func (d *decoder) getSchemaName() string { return getSchemaNameFromPayload(d.valuePayload) } From bc7c56bcd5d0ba488aac632ee6ce4e87732ec397 Mon Sep 17 00:00:00 2001 From: Ziqian Qin Date: Tue, 11 Aug 2026 11:55:02 +0800 Subject: [PATCH 2/4] debezium: do not declare start_ts in DDL/checkpoint schemas writeSourceSchema is shared by DML, DDL and checkpoint (watermark) messages. Declaring start_ts for all of them while only DML payloads carry the field breaks schema-validating consumers (declared non-optional field missing from payload). Parameterize writeSourceSchema with includeStartTs and declare start_ts only for DML row events. Add a regression test covering DDL and checkpoint messages with debezium-include-start-ts enabled. --- pkg/sink/codec/debezium/avro.go | 2 +- pkg/sink/codec/debezium/codec.go | 13 +++++---- pkg/sink/codec/debezium/codec_test.go | 41 +++++++++++++++++++++++++++ 3 files changed, 50 insertions(+), 6 deletions(-) diff --git a/pkg/sink/codec/debezium/avro.go b/pkg/sink/codec/debezium/avro.go index fcadacf00b..8ea205e9be 100644 --- a/pkg/sink/codec/debezium/avro.go +++ b/pkg/sink/codec/debezium/avro.go @@ -454,7 +454,7 @@ func (c *dbzCodec) buildDebeziumConnectSourceSchema( ) (*debeziumConnectSchema, error) { buf := &bytes.Buffer{} writer := util.BorrowJSONWriter(buf) - c.writeSourceSchema(writer, schemaName) + c.writeSourceSchema(writer, schemaName, false) util.ReturnJSONWriter(writer) return decodeDebeziumConnectSchema(buf.Bytes()) diff --git a/pkg/sink/codec/debezium/codec.go b/pkg/sink/codec/debezium/codec.go index 5bc7a8910c..bcc8180774 100644 --- a/pkg/sink/codec/debezium/codec.go +++ b/pkg/sink/codec/debezium/codec.go @@ -886,7 +886,10 @@ func (c *dbzCodec) writeBinaryField(writer *util.JSONWriter, fieldName string, v writer.WriteBase64StringField(fieldName, value) } -func (c *dbzCodec) writeSourceSchema(writer *util.JSONWriter, schemaName string) { +// includeStartTs should only be true for DML row events: DDL and checkpoint +// (watermark) messages have no per-row transaction, so their payloads never +// carry start_ts and their schemas must not declare it. +func (c *dbzCodec) writeSourceSchema(writer *util.JSONWriter, schemaName string, includeStartTs bool) { writer.WriteObjectElement(func() { writer.WriteStringField("type", "struct") writer.WriteArrayField("fields", func() { @@ -990,7 +993,7 @@ func (c *dbzCodec) writeSourceSchema(writer *util.JSONWriter, schemaName string) // start_ts is gated by debezium-include-start-ts and declared only for // the JSON protocol: the Avro payload does not carry it, so its schema // must not declare it either. - if c.config.DebeziumIncludeStartTs && !c.isDebeziumAvro() { + if includeStartTs && c.config.DebeziumIncludeStartTs && !c.isDebeziumAvro() { writer.WriteObjectElement(func() { writer.WriteStringField("type", "int64") writer.WriteBoolField("optional", false) @@ -1191,7 +1194,7 @@ func (c *dbzCodec) EncodeValue( jWriter.WriteRaw(fieldsJSON) }) }) - c.writeSourceSchema(jWriter, schemaName) + c.writeSourceSchema(jWriter, schemaName, true) jWriter.WriteObjectElement(func() { jWriter.WriteStringField("type", "string") jWriter.WriteBoolField("optional", false) @@ -1479,7 +1482,7 @@ func (c *dbzCodec) EncodeDDLEvent( jWriter.WriteIntField("version", 1) jWriter.WriteStringField("name", "io.debezium.connector.mysql.SchemaChangeValue") jWriter.WriteArrayField("fields", func() { - c.writeSourceSchema(jWriter, dbName) + c.writeSourceSchema(jWriter, dbName, false) jWriter.WriteObjectElement(func() { jWriter.WriteStringField("field", "ts_ms") jWriter.WriteBoolField("optional", false) @@ -1718,7 +1721,7 @@ func (c *dbzCodec) EncodeCheckpointEvent( fmt.Sprintf("%s.%s.Envelope", common.SanitizeName(c.clusterID), "watermark")) jWriter.WriteIntField("version", 1) jWriter.WriteArrayField("fields", func() { - c.writeSourceSchema(jWriter, "watermark") + c.writeSourceSchema(jWriter, "watermark", false) jWriter.WriteObjectElement(func() { jWriter.WriteStringField("type", "string") jWriter.WriteBoolField("optional", false) diff --git a/pkg/sink/codec/debezium/codec_test.go b/pkg/sink/codec/debezium/codec_test.go index b23167d37e..67d9f0e97d 100644 --- a/pkg/sink/codec/debezium/codec_test.go +++ b/pkg/sink/codec/debezium/codec_test.go @@ -1544,3 +1544,44 @@ func BenchmarkEncodeLargeBinary(b *testing.B) { codec.EncodeValue(e, buf) } } + +func TestStartTsNotInDDLAndCheckpointEvents(t *testing.T) { + // Even with debezium-include-start-ts enabled, DDL and checkpoint + // (watermark) messages must not declare start_ts in their schemas: + // their payloads never carry the field (no per-row transaction), and a + // declared-but-absent non-optional field breaks schema-validating consumers. + codec := &dbzCodec{ + config: common.NewConfig(config.ProtocolDebezium), + clusterID: "test_cluster", + nowFunc: func() time.Time { return time.Unix(1701326309, 0) }, + } + codec.config.DebeziumIncludeStartTs = true + codec.config.DebeziumDisableSchema = false + + helper := commonEvent.NewEventTestHelper(t) + defer helper.Close() + helper.Tk().MustExec("use test") + helper.DDL2Job(`create table test.table1(id int(10) primary key)`) + job := helper.DDL2Job(`RENAME TABLE test.table1 to test.table2`) + tableInfo := helper.GetTableInfo(job) + + e := &commonEvent.DDLEvent{ + FinishedTs: 1, + TableInfo: tableInfo, + SchemaName: "test", + TableName: "table2", + ExtraSchemaName: "test", + ExtraTableName: "table1", + Type: byte(timodel.ActionRenameTable), + Query: job.Query, + } + keyBuf := bytes.NewBuffer(nil) + buf := bytes.NewBuffer(nil) + require.NoError(t, codec.EncodeDDLEvent(e, keyBuf, buf)) + require.NotContains(t, buf.String(), "start_ts") + + keyBuf.Reset() + buf.Reset() + require.NoError(t, codec.EncodeCheckpointEvent(3, keyBuf, buf)) + require.NotContains(t, buf.String(), "start_ts") +} From 492dbbcb004fee57b2a738dde34d2c331f82d213 Mon Sep 17 00:00:00 2001 From: Ziqian Qin Date: Wed, 12 Aug 2026 12:34:46 +0800 Subject: [PATCH 3/4] debezium: nest start ts config and validate decoded values --- api/v2/changefeed_toml_test.go | 8 +++-- api/v2/model.go | 16 ++++----- api/v2/model_test.go | 5 +++ pkg/config/sink.go | 6 ++-- pkg/sink/codec/common/config.go | 4 +-- pkg/sink/codec/common/config_test.go | 4 +-- pkg/sink/codec/debezium/debezium_test.go | 43 ++++++++++++++++++++++++ pkg/sink/codec/debezium/decoder.go | 34 +++++++++++++------ 8 files changed, 92 insertions(+), 28 deletions(-) diff --git a/api/v2/changefeed_toml_test.go b/api/v2/changefeed_toml_test.go index 7ed1d4a31a..b4be90bfbe 100644 --- a/api/v2/changefeed_toml_test.go +++ b/api/v2/changefeed_toml_test.go @@ -93,7 +93,9 @@ func TestChangeFeedInfoTOMLRoundTripToInternal(t *testing.T) { ForceReplicate: util.AddressOf(true), CheckGCSafePoint: util.AddressOf(false), Sink: &SinkConfig{ - DebeziumIncludeStartTs: util.AddressOf(true), + DebeziumConfig: &DebeziumConfig{ + IncludeStartTs: util.AddressOf(true), + }, }, SyncPointInterval: &JSONDuration{duration: 10 * time.Minute}, Integrity: &IntegrityConfig{ @@ -116,7 +118,8 @@ func TestChangeFeedInfoTOMLRoundTripToInternal(t *testing.T) { // Top-level kebab-case keys and runtime field omissions. require.Contains(t, out, `sink-uri = "blackhole://"`) require.Contains(t, out, "start-ts") - require.Contains(t, out, "debezium-include-start-ts") + require.Contains(t, out, "[config.sink.debezium]") + require.Contains(t, out, "include-start-ts = true") require.NotContains(t, out, "gid") // GID is omitted from TOML (toml:"-") // The [config] section must decode into the internal ReplicaConfig used by @@ -133,6 +136,7 @@ func TestChangeFeedInfoTOMLRoundTripToInternal(t *testing.T) { require.Equal(t, 10*time.Minute, *wrapper.Config.SyncPointInterval) require.Equal(t, "correctness", util.GetOrZero(wrapper.Config.Integrity.IntegrityCheckLevel)) require.Equal(t, "eventual", util.GetOrZero(wrapper.Config.Consistent.Level)) + require.True(t, util.GetOrZero(wrapper.Config.Sink.Debezium.IncludeStartTs)) } // TestDefaultConfigTOMLRoundTripToInternal encodes the full default replica diff --git a/api/v2/model.go b/api/v2/model.go index 4e5290d757..8274fe649f 100644 --- a/api/v2/model.go +++ b/api/v2/model.go @@ -540,6 +540,9 @@ func (c *ReplicaConfig) toInternalReplicaConfigWithOriginConfig( debeziumConfig = &config.DebeziumConfig{ OutputOldValue: c.Sink.DebeziumConfig.OutputOldValue, } + if c.Sink.DebeziumConfig.IncludeStartTs != nil { + debeziumConfig.IncludeStartTs = util.AddressOf(*c.Sink.DebeziumConfig.IncludeStartTs) + } } var openProtocolConfig *config.OpenProtocolConfig if c.Sink.OpenProtocolConfig != nil { @@ -580,9 +583,6 @@ func (c *ReplicaConfig) toInternalReplicaConfigWithOriginConfig( if c.Sink.DebeziumDisableSchema != nil { res.Sink.DebeziumDisableSchema = util.AddressOf(*c.Sink.DebeziumDisableSchema) } - if c.Sink.DebeziumIncludeStartTs != nil { - res.Sink.DebeziumIncludeStartTs = util.AddressOf(*c.Sink.DebeziumIncludeStartTs) - } if c.Sink.SendBootstrapIntervalInSec != nil { res.Sink.SendBootstrapIntervalInSec = util.AddressOf(*c.Sink.SendBootstrapIntervalInSec) @@ -906,6 +906,9 @@ func ToAPIReplicaConfig(c *config.ReplicaConfig) *ReplicaConfig { debeziumConfig = &DebeziumConfig{ OutputOldValue: cloned.Sink.Debezium.OutputOldValue, } + if cloned.Sink.Debezium.IncludeStartTs != nil { + debeziumConfig.IncludeStartTs = util.AddressOf(*cloned.Sink.Debezium.IncludeStartTs) + } } var openProtocolConfig *OpenProtocolConfig if cloned.Sink.OpenProtocol != nil { @@ -962,9 +965,6 @@ func ToAPIReplicaConfig(c *config.ReplicaConfig) *ReplicaConfig { if cloned.Sink.DebeziumDisableSchema != nil { res.Sink.DebeziumDisableSchema = util.AddressOf(*cloned.Sink.DebeziumDisableSchema) } - if cloned.Sink.DebeziumIncludeStartTs != nil { - res.Sink.DebeziumIncludeStartTs = util.AddressOf(*cloned.Sink.DebeziumIncludeStartTs) - } } if cloned.Consistent != nil { if res.Consistent == nil { @@ -1204,7 +1204,6 @@ type SinkConfig struct { SendBootstrapToAllPartition *bool `json:"send_bootstrap_to_all_partition,omitempty" toml:"send-bootstrap-to-all-partition,omitempty"` SendAllBootstrapAtStart *bool `json:"send_all_bootstrap_at_start,omitempty" toml:"send-all-bootstrap-at-start,omitempty"` DebeziumDisableSchema *bool `json:"debezium_disable_schema,omitempty" toml:"debezium-disable-schema,omitempty"` - DebeziumIncludeStartTs *bool `json:"debezium_include_start_ts,omitempty" toml:"debezium-include-start-ts,omitempty"` DebeziumConfig *DebeziumConfig `json:"debezium,omitempty" toml:"debezium,omitempty"` OpenProtocolConfig *OpenProtocolConfig `json:"open,omitempty" toml:"open,omitempty"` } @@ -1587,7 +1586,8 @@ type OpenProtocolConfig struct { // DebeziumConfig represents the configurations for debezium protocol encoding type DebeziumConfig struct { - OutputOldValue bool `json:"output_old_value" toml:"output-old-value"` + OutputOldValue bool `json:"output_old_value" toml:"output-old-value"` + IncludeStartTs *bool `json:"include_start_ts,omitempty" toml:"include-start-ts,omitempty"` } type DispatcherCount struct { diff --git a/api/v2/model_test.go b/api/v2/model_test.go index 88fb821cdb..e0ca589623 100644 --- a/api/v2/model_test.go +++ b/api/v2/model_test.go @@ -41,6 +41,9 @@ func TestReplicaConfigConversion(t *testing.T) { SpoolDiskQuota: util.AddressOf(int64(1024)), SpoolBaseDir: util.AddressOf("/tmp/ticdc-spool"), }, + DebeziumConfig: &DebeziumConfig{ + IncludeStartTs: util.AddressOf(true), + }, }, Mounter: &MounterConfig{ WorkerNum: util.AddressOf(16), @@ -73,6 +76,7 @@ func TestReplicaConfigConversion(t *testing.T) { require.True(t, util.GetOrZero(internalCfg.Sink.CloudStorageConfig.UseTableIDAsPath)) require.Equal(t, int64(1024), util.GetOrZero(internalCfg.Sink.CloudStorageConfig.SpoolDiskQuota)) require.Equal(t, "/tmp/ticdc-spool", util.GetOrZero(internalCfg.Sink.CloudStorageConfig.SpoolBaseDir)) + require.True(t, util.GetOrZero(internalCfg.Sink.Debezium.IncludeStartTs)) require.Equal(t, internalCfg.Mounter.WorkerNum, *apiCfg.Mounter.WorkerNum) require.True(t, util.GetOrZero(internalCfg.Scheduler.EnableTableAcrossNodes)) require.Equal(t, 1000, util.GetOrZero(internalCfg.Scheduler.RegionThreshold)) @@ -100,6 +104,7 @@ func TestReplicaConfigConversion(t *testing.T) { require.True(t, *apiCfgBack.Sink.CloudStorageConfig.UseTableIDAsPath) require.Equal(t, int64(1024), *apiCfgBack.Sink.CloudStorageConfig.SpoolDiskQuota) require.Equal(t, "/tmp/ticdc-spool", *apiCfgBack.Sink.CloudStorageConfig.SpoolBaseDir) + require.True(t, util.GetOrZero(apiCfgBack.Sink.DebeziumConfig.IncludeStartTs)) require.Equal(t, 16, *apiCfgBack.Mounter.WorkerNum) require.True(t, *apiCfgBack.Scheduler.EnableTableAcrossNodes) require.Equal(t, "correctness", *apiCfgBack.Integrity.IntegrityCheckLevel) diff --git a/pkg/config/sink.go b/pkg/config/sink.go index 9d6779508d..535286b849 100644 --- a/pkg/config/sink.go +++ b/pkg/config/sink.go @@ -200,9 +200,6 @@ type SinkConfig struct { SendAllBootstrapAtStart *bool `toml:"send-all-bootstrap-at-start" json:"send-all-bootstrap-at-start,omitempty"` // Debezium only. Whether schema should be excluded in the output. DebeziumDisableSchema *bool `toml:"debezium-disable-schema" json:"debezium-disable-schema,omitempty"` - // Debezium only. Whether the transaction start_ts should be included in - // the source block of the output. JSON protocol only. - DebeziumIncludeStartTs *bool `toml:"debezium-include-start-ts" json:"debezium-include-start-ts,omitempty"` // CSVConfig is only available when the downstream is Storage. CSVConfig *CSVConfig `toml:"csv" json:"csv,omitempty"` @@ -1170,6 +1167,9 @@ type OpenProtocolConfig struct { // DebeziumConfig represents the configurations for debezium protocol encoding type DebeziumConfig struct { OutputOldValue bool `toml:"output-old-value" json:"output-old-value"` + // IncludeStartTs controls whether the transaction start_ts is included in + // the source block of Debezium JSON output. + IncludeStartTs *bool `toml:"include-start-ts" json:"include-start-ts,omitempty"` } // validRoutingExpressionRegexp accepts routing expressions made of literal text diff --git a/pkg/sink/codec/common/config.go b/pkg/sink/codec/common/config.go index dadb8a9c87..5671d6c186 100644 --- a/pkg/sink/codec/common/config.go +++ b/pkg/sink/codec/common/config.go @@ -353,8 +353,8 @@ func mergeConfig( if sinkConfig.DebeziumDisableSchema != nil { dest.DebeziumDisableSchema = sinkConfig.DebeziumDisableSchema } - if sinkConfig.DebeziumIncludeStartTs != nil { - dest.DebeziumIncludeStartTs = sinkConfig.DebeziumIncludeStartTs + if sinkConfig.Debezium != nil && sinkConfig.Debezium.IncludeStartTs != nil { + dest.DebeziumIncludeStartTs = sinkConfig.Debezium.IncludeStartTs } } if err := mergo.Merge(dest, urlParameters, mergo.WithOverride); err != nil { diff --git a/pkg/sink/codec/common/config_test.go b/pkg/sink/codec/common/config_test.go index a0ad34a782..b387c30532 100644 --- a/pkg/sink/codec/common/config_test.go +++ b/pkg/sink/codec/common/config_test.go @@ -172,7 +172,7 @@ func TestDebeziumIncludeStartTsConfig(t *testing.T) { on := true cfg2 := NewConfig(config.ProtocolDebezium) sinkConfig := config.GetDefaultReplicaConfig().Sink - sinkConfig.DebeziumIncludeStartTs = &on + sinkConfig.Debezium.IncludeStartTs = &on sinkURI2, err := url.Parse("kafka://127.0.0.1:9092/topic?protocol=debezium") require.NoError(t, err) require.NoError(t, cfg2.Apply(sinkURI2, sinkConfig)) @@ -181,7 +181,7 @@ func TestDebeziumIncludeStartTsConfig(t *testing.T) { // URI parameter overrides the config file cfg3 := NewConfig(config.ProtocolDebezium) sinkConfig3 := config.GetDefaultReplicaConfig().Sink - sinkConfig3.DebeziumIncludeStartTs = &on + sinkConfig3.Debezium.IncludeStartTs = &on sinkURI3, err := url.Parse("kafka://127.0.0.1:9092/topic?protocol=debezium&debezium-include-start-ts=false") require.NoError(t, err) require.NoError(t, cfg3.Apply(sinkURI3, sinkConfig3)) diff --git a/pkg/sink/codec/debezium/debezium_test.go b/pkg/sink/codec/debezium/debezium_test.go index 5b3dcdbd92..1fca4b24d3 100644 --- a/pkg/sink/codec/debezium/debezium_test.go +++ b/pkg/sink/codec/debezium/debezium_test.go @@ -289,6 +289,49 @@ func TestDecodeStartTsFallbackToCommitTs(t *testing.T) { require.NotEqual(t, uint64(5), decoded.GetStartTs()) } +func TestDecodeNonPositiveStartTsFallbackToCommitTs(t *testing.T) { + cfg := common.NewConfig(config.ProtocolDebezium) + cfg.DebeziumIncludeStartTs = true + cfg.EnableTiDBExtension = true + cfg.TimeZone = time.UTC + + encoder := NewBatchEncoder(cfg, "dbserver1") + rowEvent := common.NewRoutedRowEvent4Test() + rowEvent.StartTs = 5 + require.NoError(t, encoder.AppendRowChangedEvent(context.Background(), "", rowEvent)) + + messages := encoder.Build() + require.Len(t, messages, 1) + + for _, tc := range []struct { + name string + startTs json.Number + }{ + {name: "zero", startTs: json.Number("0")}, + {name: "negative", startTs: json.Number("-1")}, + } { + t.Run(tc.name, func(t *testing.T) { + dec := json.NewDecoder(bytes.NewReader(messages[0].Value)) + dec.UseNumber() + var value map[string]any + require.NoError(t, dec.Decode(&value)) + payload := value["payload"].(map[string]any) + source := payload["source"].(map[string]any) + source["start_ts"] = tc.startTs + valueBytes, err := json.Marshal(value) + require.NoError(t, err) + + decoder := NewDecoder(cfg, 0, nil) + decoder.AddKeyValue(messages[0].Key, valueBytes) + messageType, hasNext := decoder.HasNext() + require.True(t, hasNext) + require.Equal(t, common.MessageTypeRow, messageType) + decoded := decoder.NextDMLMessage().ToDMLEvent() + require.Equal(t, decoded.GetCommitTs(), decoded.GetStartTs()) + }) + } +} + // schemaFieldsByName returns the sub-schema object of a field inside a Debezium // struct schema, or nil when the field is not declared. func schemaFieldsByName(t *testing.T, schema map[string]any, name string) map[string]any { diff --git a/pkg/sink/codec/debezium/decoder.go b/pkg/sink/codec/debezium/decoder.go index 9e8163743e..ac3a536b15 100644 --- a/pkg/sink/codec/debezium/decoder.go +++ b/pkg/sink/codec/debezium/decoder.go @@ -202,10 +202,11 @@ func (d *decoder) assembleDMLEventFromPayload( ) *commonEvent.DMLEvent { tableInfo := queryTableInfoFromPayload(keyPayload, valuePayload, valueSchema) commitTs := getCommitTsFromPayload(valuePayload) - startTs := getStartTsFromPayload(valuePayload) - if startTs == 0 { - // Fall back to commit_ts for messages that do not carry start_ts, - // keeping the pre-feature behavior for old messages. + startTs, hasStartTs := getStartTsFromPayload(valuePayload) + if !hasStartTs { + // Keep old messages consumable when start_ts is absent. Invalid values + // are logged by getStartTsFromPayload and also fall back so a malformed + // message does not stop production consumption. startTs = commitTs } event := &commonEvent.DMLEvent{ @@ -256,20 +257,31 @@ func getCommitTsFromPayload(valuePayload map[string]any) uint64 { return uint64(commitTs) } -// getStartTsFromPayload returns the start_ts carried in the source block, or 0 -// when the field is absent (messages produced before the start_ts field existed). -func getStartTsFromPayload(valuePayload map[string]any) uint64 { +// getStartTsFromPayload returns the start_ts carried in the source block. +// It returns false when the field is absent or invalid. Invalid values are +// logged before returning so callers can fall back without stopping consumption. +func getStartTsFromPayload(valuePayload map[string]any) (uint64, bool) { source := valuePayload["source"].(map[string]any) - startTs, ok := source["start_ts"].(json.Number) + rawStartTs, exists := source["start_ts"] + if !exists { + return 0, false + } + startTs, ok := rawStartTs.(json.Number) if !ok { - return 0 + log.Error("decode value failed", + zap.String("reason", "start_ts is not an integer"), + zap.String("value", util.RedactAny(source))) + return 0, false } ts, err := startTs.Int64() + if err == nil && ts <= 0 { + err = errors.Errorf("start_ts must be positive: %d", ts) + } if err != nil { log.Error("decode value failed", zap.Error(err), zap.String("value", util.RedactAny(source))) - return 0 + return 0, false } - return uint64(ts) + return uint64(ts), true } func (d *decoder) getSchemaName() string { From 7bda3cafd7064dc861465913663062a5a4ac7d14 Mon Sep 17 00:00:00 2001 From: Ziqian Qin Date: Thu, 13 Aug 2026 14:46:16 +0800 Subject: [PATCH 4/4] debezium: simplify start ts schema condition --- pkg/sink/codec/debezium/codec.go | 13 +++++-------- pkg/sink/codec/debezium/debezium_test.go | 12 ++++++++++++ 2 files changed, 17 insertions(+), 8 deletions(-) diff --git a/pkg/sink/codec/debezium/codec.go b/pkg/sink/codec/debezium/codec.go index bcc8180774..f20a6fc136 100644 --- a/pkg/sink/codec/debezium/codec.go +++ b/pkg/sink/codec/debezium/codec.go @@ -886,9 +886,9 @@ func (c *dbzCodec) writeBinaryField(writer *util.JSONWriter, fieldName string, v writer.WriteBase64StringField(fieldName, value) } -// includeStartTs should only be true for DML row events: DDL and checkpoint -// (watermark) messages have no per-row transaction, so their payloads never -// carry start_ts and their schemas must not declare it. +// includeStartTs indicates whether start_ts should be declared in the source +// schema. DML callers pass the configured value, while DDL, checkpoint, and +// Avro callers pass false because their payloads do not carry the field. func (c *dbzCodec) writeSourceSchema(writer *util.JSONWriter, schemaName string, includeStartTs bool) { writer.WriteObjectElement(func() { writer.WriteStringField("type", "struct") @@ -990,10 +990,7 @@ func (c *dbzCodec) writeSourceSchema(writer *util.JSONWriter, schemaName string, writer.WriteStringField("field", "cluster_id") }) } - // start_ts is gated by debezium-include-start-ts and declared only for - // the JSON protocol: the Avro payload does not carry it, so its schema - // must not declare it either. - if includeStartTs && c.config.DebeziumIncludeStartTs && !c.isDebeziumAvro() { + if includeStartTs { writer.WriteObjectElement(func() { writer.WriteStringField("type", "int64") writer.WriteBoolField("optional", false) @@ -1194,7 +1191,7 @@ func (c *dbzCodec) EncodeValue( jWriter.WriteRaw(fieldsJSON) }) }) - c.writeSourceSchema(jWriter, schemaName, true) + c.writeSourceSchema(jWriter, schemaName, c.config.DebeziumIncludeStartTs) jWriter.WriteObjectElement(func() { jWriter.WriteStringField("type", "string") jWriter.WriteBoolField("optional", false) diff --git a/pkg/sink/codec/debezium/debezium_test.go b/pkg/sink/codec/debezium/debezium_test.go index 1fca4b24d3..44cefdb9af 100644 --- a/pkg/sink/codec/debezium/debezium_test.go +++ b/pkg/sink/codec/debezium/debezium_test.go @@ -279,6 +279,18 @@ func TestDecodeStartTsFallbackToCommitTs(t *testing.T) { messages := encoder.Build() require.Len(t, messages, 1) + dec := json.NewDecoder(bytes.NewReader(messages[0].Value)) + dec.UseNumber() + var value map[string]any + require.NoError(t, dec.Decode(&value)) + payload := value["payload"].(map[string]any) + source := payload["source"].(map[string]any) + require.NotContains(t, source, "start_ts") + schema := value["schema"].(map[string]any) + sourceSchema := schemaFieldsByName(t, schema, "source") + require.NotNil(t, sourceSchema) + require.Nil(t, schemaFieldsByName(t, sourceSchema, "start_ts")) + decoder := NewDecoder(cfg, 0, nil) decoder.AddKeyValue(messages[0].Key, messages[0].Value) messageType, hasNext := decoder.HasNext()