debezium: output start_ts in the source block of DML messages - #5903
debezium: output start_ts in the source block of DML messages#5903ekexium wants to merge 4 commits into
Conversation
|
Skipping CI for Draft Pull Request. |
ti-chi-bot
Bot
added
do-not-merge/needs-linked-issue
do-not-merge/work-in-progress
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (8)
🚧 Files skipped from review as they are similar to previous changes (5)
📝 WalkthroughWalkthroughAdds optional ChangesDebezium start timestamp
Estimated code review effort: 3 (Moderate) | ~25 minutes Sequence Diagram(s)sequenceDiagram
participant SinkConfig
participant DebeziumCodec
participant DebeziumMessage
participant DebeziumDecoder
participant DMLEvent
SinkConfig->>DebeziumCodec: enable DebeziumIncludeStartTs
DebeziumCodec->>DebeziumMessage: encode RowEvent.StartTs as source.start_ts
DebeziumMessage->>DebeziumDecoder: provide source.start_ts
DebeziumDecoder->>DMLEvent: restore StartTs
DebeziumDecoder->>DMLEvent: use commit_ts when start_ts is absent or invalid
Possibly related PRs
Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 2⚔️ Resolve merge conflicts 💡
🛠️ Fix failing CI checks 💡
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
|
|
ti-chi-bot
Bot
added
the
size/L
1b27000 to
dc43cf0
Compare
ti-chi-bot
Bot
added
do-not-merge/needs-triage-completed
release-note
dc43cf0 to
f3a3988
Compare
|
/check-issue-triage-complete |
a569d07 to
c364a44
Compare
|
[APPROVALNOTIFIER] This PR is NOT APPROVED This pull-request has been approved by: The full list of commands accepted by this bot can be found here. DetailsNeeds approval from an approver in each of these files:Approvers can indicate their approval by writing |
ti-chi-bot
Bot
removed
the
do-not-merge/work-in-progress
There was a problem hiding this comment.
Pull request overview
Adds an opt-in Debezium (JSON) output enhancement to include a transaction’s start_ts in the source block of DML messages, and updates decoding/config to support the new field while preserving backward compatibility for older messages.
Changes:
- Add
debezium-output-start-tssink option/config to emitsource.start_tsin Debezium JSON DML value messages. - Update Debezium JSON schema generation (JSON-only; guarded for Avro) and decoder logic to read
start_tswith fallback tocommit_tswhen absent. - Add unit tests covering JSON emission/schema, decoder fallback, Avro non-impact, and config parsing/validation.
Validation:
- Not run here (no CI access in this environment). Suggested:
go test -tags intest ./pkg/sink/codec/debezium/... ./pkg/sink/codec/common/... ./pkg/config/... ./cmd/kafka-consumer/...
Reviewed changes
Copilot reviewed 7 out of 7 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
| pkg/sink/codec/debezium/decoder.go | Decode source.start_ts when present; fall back to commit_ts for older messages. |
| pkg/sink/codec/debezium/debezium_test.go | Add unit tests for start_ts emission/schema and decoder fallback behavior. |
| pkg/sink/codec/debezium/codec.go | Emit source.start_ts in JSON payload + declare in JSON schema; guard against Avro schema/payload changes. |
| pkg/sink/codec/debezium/avro_test.go | Assert Avro payload/schema do not include start_ts even when option is enabled. |
| pkg/sink/codec/common/config.go | Add DebeziumOutputStartTs config field, URI parsing, and validation gating by protocol. |
| pkg/sink/codec/common/config_test.go | Test URI/config-file parsing and protocol validation for debezium-output-start-ts. |
| pkg/config/sink.go | Add [sink.debezium].output-start-ts config field to Debezium sink config. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| 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) | ||
|
|
||
| // Simulate a pre-feature message: drop start_ts from the source block. | ||
| dec := json.NewDecoder(bytes.NewReader(messages[0].Value)) | ||
| dec.UseNumber() | ||
| var value map[string]any | ||
| require.NoError(t, dec.Decode(&value)) | ||
| source := value["payload"].(map[string]any)["source"].(map[string]any) | ||
| require.Equal(t, json.Number("5"), source["start_ts"]) | ||
| delete(source, "start_ts") | ||
| valueBytes, err := json.Marshal(value) | ||
| require.NoError(t, err) | ||
|
|
||
| decoder := NewDecoder(cfg, 0, nil) | ||
| decoder.AddKeyValue(messages[0].Key, valueBytes) |
| // round-trip: decoding restores the true start ts. The TiCDC-side decoder | ||
| // requires enable-tidb-extension (to read commit_ts and tidb_type), so the | ||
| // encoded message must carry the extension fields as well. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (1)
pkg/sink/codec/common/config_test.go (1)
162-179: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd a URI precedence test.
The separate tests do not verify the override contract when both configuration sources set
debezium-output-start-ts.Add cases where the sink configuration and URI parameter have opposite values. Assert that the URI value wins.
🤖 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/common/config_test.go` around lines 162 - 179, Extend TestDebeziumOutputStartTsConfig with cases where sinkConfig.Debezium.OutputStartTs and the debezium-output-start-ts URI parameter are set to opposite boolean values. Apply each configuration and assert cfg.DebeziumOutputStartTs matches the URI value, covering both true-over-false and false-over-true precedence.
🤖 Prompt for all review comments with 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.
Inline comments:
In `@pkg/sink/codec/debezium/codec.go`:
- Around line 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.
In `@pkg/sink/codec/debezium/decoder.go`:
- Around line 267-272: Update the startTs conversion logic in the decoder to
reject negative signed values before converting to uint64, ensuring negative
start_ts values return 0 and trigger the existing commit_ts fallback. Add a
decoder test covering negative start_ts and verify the resulting timestamp
equals commit_ts.
---
Nitpick comments:
In `@pkg/sink/codec/common/config_test.go`:
- Around line 162-179: Extend TestDebeziumOutputStartTsConfig with cases where
sinkConfig.Debezium.OutputStartTs and the debezium-output-start-ts URI parameter
are set to opposite boolean values. Apply each configuration and assert
cfg.DebeziumOutputStartTs matches the URI value, covering both true-over-false
and false-over-true precedence.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 510e587e-ef39-4457-aef6-1f5226a3180e
📒 Files selected for processing (7)
pkg/config/sink.gopkg/sink/codec/common/config.gopkg/sink/codec/common/config_test.gopkg/sink/codec/debezium/avro_test.gopkg/sink/codec/debezium/codec.gopkg/sink/codec/debezium/debezium_test.gopkg/sink/codec/debezium/decoder.go
| // 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) | ||
| } |
There was a problem hiding this comment.
🗄️ 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.
| // 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.
| func TestDebeziumOutputStartTsConfig(t *testing.T) { | ||
| // URI parameter | ||
| cfg := NewConfig(config.ProtocolDebezium) | ||
| sinkURI, err := url.Parse("kafka://127.0.0.1:9092/topic?protocol=debezium&debezium-output-start-ts=true") |
There was a problem hiding this comment.
Can we rename to debezium-include-start-ts? I remember we have avro-include-before-value, so I think to keep it consistent. #5154
There was a problem hiding this comment.
In the changefeed config, do you prefer [sink] debezium-include-start-ts or [sink.debezium] include-start-ts = true? #5154 is the in the former form, but I think the latter is more reasonable.
There was a problem hiding this comment.
Agree, please go with the latter one.
c364a44 to
039dffa
Compare
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with 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.
Inline comments:
In `@pkg/config/sink.go`:
- Around line 203-205: Move DebeziumIncludeStartTs from SinkConfig into
DebeziumConfig, or explicitly map it through the existing SinkConfig.Debezium
field so the [sink.debezium] path populates it. Update mergeConfig to read the
nested value and extend the configuration test to verify DebeziumIncludeStartTs
is enabled through the nested form.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 909e3c37-18f3-47fb-820c-b9a1e2a52dbc
📒 Files selected for processing (6)
pkg/config/sink.gopkg/sink/codec/common/config.gopkg/sink/codec/common/config_test.gopkg/sink/codec/debezium/avro_test.gopkg/sink/codec/debezium/codec.gopkg/sink/codec/debezium/debezium_test.go
🚧 Files skipped from review as they are similar to previous changes (3)
- pkg/sink/codec/debezium/codec.go
- pkg/sink/codec/debezium/debezium_test.go
- pkg/sink/codec/debezium/avro_test.go
039dffa to
2d4e12c
Compare
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).
2d4e12c to
90947b6
Compare
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.
ti-chi-bot
Bot
added
size/XL
There was a problem hiding this comment.
🧹 Nitpick comments (1)
api/v2/changefeed_toml_test.go (1)
119-119: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAssert the decoded sink value.
Line 119 only verifies text serialization. Assert
wrapper.Config.Sink.DebeziumIncludeStartTsafter TOML decoding. This verifies the TOML path and the internal field mapping.Proposed test addition
_, err := toml.Decode(out, &wrapper) require.NoError(t, err) + require.NotNil(t, wrapper.Config.Sink) + require.True(t, util.GetOrZero(wrapper.Config.Sink.DebeziumIncludeStartTs)) require.Equal(t, uint64(1024), util.GetOrZero(wrapper.Config.MemoryQuota))🤖 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 `@api/v2/changefeed_toml_test.go` at line 119, Extend the TOML decoding test around the existing serialization assertion to also validate wrapper.Config.Sink.DebeziumIncludeStartTs after decoding, confirming the sink field mapping and decoded value rather than only checking serialized text.
🤖 Prompt for all review comments with 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.
Nitpick comments:
In `@api/v2/changefeed_toml_test.go`:
- Line 119: Extend the TOML decoding test around the existing serialization
assertion to also validate wrapper.Config.Sink.DebeziumIncludeStartTs after
decoding, confirming the sink field mapping and decoded value rather than only
checking serialized text.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: c4bde877-ddc6-4c92-935f-f66f34a75c7e
📒 Files selected for processing (5)
api/v2/changefeed_toml_test.goapi/v2/model.gopkg/sink/codec/debezium/avro.gopkg/sink/codec/debezium/codec.gopkg/sink/codec/debezium/codec_test.go
…rt-ts # Conflicts: # api/v2/changefeed_toml_test.go
|
@ekexium: The following test failed, say
Full PR test history. Your PR dashboard. DetailsInstructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. I understand the commands that are listed here. |
|
@wk989898 PTAL |
| // 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) { |
There was a problem hiding this comment.
It looks like only dml will carry the start ts field?
What problem does this PR solve?
Issue Number: close #5904
What is changed and how it works?
Add a new changefeed-level option for Debezium JSON output (default
false). Configure it with the sink URI parameterdebezium-include-start-ts=trueor in the changefeed config:When enabled, the Debezium JSON value message carries the start TSO of the transaction that made the change as
source.start_ts, next tocommit_ts:source.start_ts = e.StartTswhenDebeziumIncludeStartTsis enabled.start_tsunder the same option (JSON only), so schema-validated consumers (e.g. Kafka Connect) can see the field withoutenable-tidb-extension. The JSON source-schema writer is shared with Avro, so the declaration is guarded against Avro — the Avro payload does not carry the field and its schema must not declare it.source.start_tsand use it as the decoded event'sStartTs(previouslyStartTs = CommitTs). Messages without the field (pre-feature format) fall back toCommitTs, preserving existing behavior. Present but invalid values (non-positive, out ofint64range, or not an integer) emit an error log and fall back toCommitTsso consumption can continue.debezium-include-start-tsin the sink URI, orinclude-start-tsunder[sink.debezium]/ the nested API v2 Debezium config. Explicit URI values (includingfalse) override the config file — applied after the mergo merge because mergo cannot override a*booltruewithfalse. Validation rejects the option for non-debeziumprotocols.Check List
Tests
Questions
Will it cause performance regression or break compatibility?
No measurable performance impact (one extra uint64 field per DML message, only when the option is enabled). With the default config (option off) nothing changes at all; with the option on, the
sourceblock gains one field and the schema declares it consistently, so schema-validated consumers see a matching declaration. DDL/checkpoint messages, key messages, row checksums and the Avro protocol are unchanged. The decoder falls back toCommitTswhenstart_tsis absent, so messages produced before this change decode exactly as before. Invalid present values are logged at error level and also fall back to keep production consumption running.Do you need to update user documentation, design documentation or monitoring documentation?
Yes — the TiCDC Debezium protocol and changefeed configuration documentation will be updated in a follow-up PR to describe both forms: the URI parameter
debezium-include-start-tsand[sink.debezium] include-start-ts.Release note
Summary by CodeRabbit
New Features
start_ts) in Debezium JSON events.start_tsand falls back to the commit timestamp when unavailable.Improvements