Skip to content

debezium: output start_ts in the source block of DML messages #5903

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ekexium wants to merge 6 commits into
base: master
Choose a base branch
from
Open

debezium: output start_ts in the source block of DML messages #5903

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
90947b6
debezium: add debezium-include-start-ts to emit source.start_ts
ekexium Aug 7, 2026
bc7c56b
debezium: do not declare start_ts in DDL/checkpoint schemas
ekexium Aug 11, 2026
492dbbc
debezium: nest start ts config and validate decoded values
ekexium Aug 12, 2026
56d0dc5
Merge remote-tracking branch 'upstream/master' into feat/debezium-sta…
ekexium Aug 12, 2026
7bda3ca
debezium: simplify start ts schema condition
ekexium Aug 13, 2026
947ad62
Merge remote-tracking branch 'upstream/master' into feat/debezium-sta…
ekexium Aug 13, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 13 additions & 5 deletions api/v2/changefeed_toml_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -88,11 +88,16 @@ func TestChangeFeedInfoTOMLRoundTripToInternal(t *testing.T) {
SinkURI: "blackhole://",
StartTs: 449999999999999999,
Config: &ReplicaConfig{
PerformanceMode: util.AddressOf(config.PerformanceModeLowLatency),
MemoryQuota: util.AddressOf(uint64(1024)),
CaseSensitive: util.AddressOf(true),
ForceReplicate: util.AddressOf(true),
CheckGCSafePoint: util.AddressOf(false),
PerformanceMode: util.AddressOf(config.PerformanceModeLowLatency),
MemoryQuota: util.AddressOf(uint64(1024)),
CaseSensitive: util.AddressOf(true),
ForceReplicate: util.AddressOf(true),
CheckGCSafePoint: util.AddressOf(false),
Sink: &SinkConfig{
DebeziumConfig: &DebeziumConfig{
IncludeStartTs: util.AddressOf(true),
},
},
SyncPointInterval: &JSONDuration{duration: 10 * time.Minute},
Integrity: &IntegrityConfig{
IntegrityCheckLevel: util.AddressOf("correctness"),
Expand All @@ -114,6 +119,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, "[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
Expand All @@ -131,6 +138,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))
}

func TestCodecConfigTOMLRoundTripToInternal(t *testing.T) {
Expand Down
9 changes: 8 additions & 1 deletion api/v2/model.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -546,6 +546,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 {
Expand Down Expand Up @@ -912,6 +915,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 {
Expand Down Expand Up @@ -1591,7 +1597,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"`

@lidezhu lidezhu Aug 13, 2026 •
edited
Loading

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Preserve output_old_value when only include_start_ts is provided

A partial v2 update such as:

  {
    "replica_config": {
      "sink": {
        "debezium": {
          "include_start_ts": true
        }
      }
    }
  }

silently changes output_old_value from its default true to false. UpdateChangefeed decodes into an empty config, and DebeziumConfig.OutputOldValue is a non-pointer bool, so an omitted field becomes false. The conversion here then copies that zero value into a newly constructed internal DebeziumConfig, overriding the default. Consequently, Debezium update events stop carrying the before value even though the user only enabled start_ts.

Please make the API field optional (for example, OutputOldValue *bool) and only override the existing/default internal value when it is non-nil. Please also add a unit test to cover this case.

}

type DispatcherCount struct {
Expand Down
5 changes: 5 additions & 0 deletions api/v2/model_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,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),
Expand Down Expand Up @@ -75,6 +78,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))
Expand Down Expand Up @@ -103,6 +107,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)
Expand Down
3 changes: 3 additions & 0 deletions pkg/config/sink.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1169,6 +1169,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
Expand Down
26 changes: 25 additions & 1 deletion pkg/sink/codec/common/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,6 +99,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
}
Expand Down Expand Up @@ -145,6 +148,7 @@ func NewConfig(protocol config.Protocol) *Config {
DebeziumOutputOldValue: true,
OpenOutputOldValue: true,
DebeziumDisableSchema: false,
DebeziumIncludeStartTs: false,
CSVOutputFieldHeader: false,
}
}
Expand Down Expand Up @@ -185,7 +189,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"`
Expand All @@ -203,6 +208,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
}
Expand Down Expand Up @@ -313,6 +322,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
}
Expand Down Expand Up @@ -345,6 +360,9 @@ func mergeConfig(
if sinkConfig.DebeziumDisableSchema != nil {
dest.DebeziumDisableSchema = sinkConfig.DebeziumDisableSchema
}
if sinkConfig.Debezium != nil && sinkConfig.Debezium.IncludeStartTs != nil {
dest.DebeziumIncludeStartTs = sinkConfig.Debezium.IncludeStartTs
}
}
if err := mergo.Merge(dest, urlParameters, mergo.WithOverride); err != nil {
return nil, err
Expand Down Expand Up @@ -388,6 +406,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"
Expand Down
37 changes: 37 additions & 0 deletions pkg/sink/codec/common/config_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ import (
"testing"

"github.com/pingcap/ticdc/pkg/config"
"github.com/pingcap/ticdc/pkg/errors"
"github.com/pingcap/ticdc/pkg/util"
"github.com/stretchr/testify/require"
)
Expand Down Expand Up @@ -53,3 +54,39 @@ func TestAvroIncludeBeforeValueConfigFile(t *testing.T) {
require.False(t, cfg.EnableTiDBExtension)
require.True(t, cfg.AvroIncludeBeforeValue)
}

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.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))
require.True(t, cfg2.DebeziumIncludeStartTs)

// URI parameter overrides the config file
cfg3 := NewConfig(config.ProtocolDebezium)
sinkConfig3 := config.GetDefaultReplicaConfig().Sink
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))
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)
}
2 changes: 1 addition & 1 deletion pkg/sink/codec/debezium/avro.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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())
Expand Down
6 changes: 6 additions & 0 deletions pkg/sink/codec/debezium/avro_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down Expand Up @@ -103,13 +105,17 @@ 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"`)
require.Contains(t, valueSchema, `"name":"foo"`)
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) {
Expand Down
23 changes: 19 additions & 4 deletions pkg/sink/codec/debezium/codec.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 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) {

@wk989898 wk989898 Aug 12, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It looks like only dml will carry the start ts field?

@ekexium ekexium Aug 12, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, only DML events

writer.WriteObjectElement(func() {
writer.WriteStringField("type", "struct")
writer.WriteArrayField("fields", func() {
Expand Down Expand Up @@ -987,6 +990,13 @@ func (c *dbzCodec) writeSourceSchema(writer *util.JSONWriter, schemaName string)
writer.WriteStringField("field", "cluster_id")
})
}
if includeStartTs {
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))
Expand Down Expand Up @@ -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)
}
Comment on lines +1097 to +1101

@coderabbitai coderabbitai Bot Aug 10, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win

Exclude start_ts from Debezium Avro payloads.

Line 1099 does not check !c.isDebeziumAvro(). When DebeziumOutputStartTs is true, this writes source.start_ts into an Avro record although Lines 990-999 exclude it from the Avro schema.

This breaks the Avro payload/schema contract and fails the Avro exclusion test. Gate the payload write with the same Avro condition as the schema.

Proposed fix 
-				if c.config.DebeziumOutputStartTs {
+				if c.config.DebeziumOutputStartTs && !c.isDebeziumAvro() {
 					jWriter.WriteUint64Field("start_ts", e.StartTs)
 				}
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
// start_ts: the start TSO of the transaction that made this change,
// exposed for downstream consumers that need transaction correlation.
if c.config.DebeziumOutputStartTs {
jWriter.WriteUint64Field("start_ts", e.StartTs)
}
// start_ts: the start TSO of the transaction that made this change,
// exposed for downstream consumers that need transaction correlation.
if c.config.DebeziumOutputStartTs && !c.isDebeziumAvro() {
jWriter.WriteUint64Field("start_ts", e.StartTs)
}
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@pkg/sink/codec/debezium/codec.go` around lines 1097 - 1101, Update the
start_ts payload write in the Debezium encoding flow to require
DebeziumOutputStartTs and a non-Avro output, matching the schema exclusion
logic. Use c.isDebeziumAvro() in the condition around
jWriter.WriteUint64Field("start_ts", e.StartTs), preserving start_ts for
non-Avro payloads.

jWriter.WriteStringField("cluster_id", c.clusterID)
})

Expand Down Expand Up @@ -1176,7 +1191,7 @@ func (c *dbzCodec) EncodeValue(
jWriter.WriteRaw(fieldsJSON)
})
})
c.writeSourceSchema(jWriter, schemaName)
c.writeSourceSchema(jWriter, schemaName, c.config.DebeziumIncludeStartTs)

@wk989898 wk989898 Aug 13, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Does this affect the Debezium-Avro protocol?

@ekexium ekexium Aug 13, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No. Debezium Avro explicitly calls writeSourceSchema(writer, schemaName, false) in buildDebeziumAvroValueMessage. And there is a test TestDebeziumConfluentAvroEncodeRowEvent

jWriter.WriteObjectElement(func() {
jWriter.WriteStringField("type", "string")
jWriter.WriteBoolField("optional", false)
Expand Down Expand Up @@ -1464,7 +1479,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)
Expand Down Expand Up @@ -1703,7 +1718,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)
Expand Down
41 changes: 41 additions & 0 deletions pkg/sink/codec/debezium/codec_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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")
}
Loading
Loading