consumer: change message acknowledgment to cumulative for pulsar consumer

[wk989898]

What problem does this PR solve?

Issue Number: close #xxx

What is changed and how it works?

Check List

Tests

• Unit test
• Integration test
• Manual test (add detailed scripts or steps below)
• No code

Questions

Will it cause performance regression or break compatibility?

Do you need to update user documentation, design documentation or monitoring documentation?

Release note

Please refer to [Release Notes Language Style Guide](https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/release-notes-style-guide.html) to write a quality release note.

If you don't think this PR needs a release note then fill it with `None`.

Summary by CodeRabbit

• Bug Fixes
  • Updated message acknowledgment behavior for Pulsar consumption so committed messages are acknowledged cumulatively, improving consistency in processing and reducing the chance of duplicate reprocessing.

[ti-chi-bot]

[APPROVALNOTIFIER] This PR is NOT APPROVED

This pull-request has been approved by:
Once this PR has been reviewed and has the lgtm label, please assign 3aceshowhand for approval. For more information see the Code Review Process.
Please ensure that each of them provides their approval before proceeding.

The full list of commands accepted by this bot can be found here.

Details

Needs approval from an approver in each of these files:

• OWNERS

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: 917f8075-e9c0-488b-b7b2-5a96ad081569

📥 Commits

Reviewing files that changed from the base of the PR and between a0066dc and 7623ce9.

📒 Files selected for processing (1)

• cmd/pulsar-consumer/consumer.go

---

📝 Walkthrough

Walkthrough

In cmd/pulsar-consumer/consumer.go, the Pulsar acknowledgment call in the needCommit branch is changed from AckID (single-message acknowledgment) to AckIDCumulative (cumulative acknowledgment up to that message ID).

Changes

Pulsar Consumer Ack Semantics

Layer / File(s)	Summary
AckID → AckIDCumulative on commit cmd/pulsar-consumer/consumer.go	When needCommit is true after WriteMessage, the acknowledgment call switches from AckID to AckIDCumulative, committing all messages up to the current message ID.

Estimated code review effort

🎯 2 (Simple) | ⏱️ ~5 minutes

Poem

A rabbit hops past the Pulsar queue,
One ack becomes cumulative — who knew?
All messages prior now swept in one swoop,
The broker breathes easy, confirmed in a loop.
🐇✉️ Hop, hop, committed!

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Description check	⚠️ Warning	The description is mostly placeholder template text and lacks concrete issue, change details, tests, compatibility, and release note.	Fill in the issue number, summarize the change, add at least one test, answer compatibility/documentation questions, and provide a release note.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title clearly states the main change: switching Pulsar consumer acknowledgments to cumulative.
Docstring Coverage	✅ Passed	No functions found in the changed files to evaluate docstring coverage. Skipping docstring coverage check.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

✨ Finishing Touches

📝 Generate docstrings

• Create stacked PR
• Commit on current branch

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch wk989898-patch-1

---

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.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[gemini-code-assist]

Code Review

This pull request replaces individual message acknowledgment (AckID) with cumulative acknowledgment (AckIDCumulative) in the Pulsar consumer. The reviewer identified a critical bug in this change: using cumulative acknowledgment for all messages, including DDL messages, can lead to silent data loss if the consumer crashes before buffered DMLs of other tables are flushed. The reviewer recommends using cumulative acknowledgment only for ResolvedTS messages and individual acknowledgment for DDL messages, and has provided a detailed code suggestion to implement this strategy.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[gemini-code-assist]

Critical Bug: Potential Silent Data Loss with Cumulative Acknowledgment on DDLs

Changing the acknowledgment to AckIDCumulative for all committed messages introduces a critical data loss vulnerability when processing DDL messages.

Why this happens:

1. When a DDL message is received, WriteMessage only flushes the DMLs of the tables blocked by this DDL (via flushDDLEvent).
2. DMLs of other (non-blocked) tables remain buffered in memory (progress.eventsGroup) and are not flushed to the downstream database.
3. Since the DDL is executed successfully, WriteMessage returns needCommit = true.
4. The consumer then calls AckIDCumulative(DDL_Msg.ID()), which cumulatively acknowledges all messages up to the DDL message, including the unflushed DMLs of other tables.
5. If the consumer crashes or restarts before those other DMLs are flushed (which only happens when a subsequent ResolvedTS is processed), those DMLs will never be redelivered because they were already acknowledged in Pulsar. This results in silent data loss.

Recommended Solution:

• ResolvedTS Messages: Use AckIDCumulative because a ResolvedTS flush guarantees that all DMLs of all tables up to the watermark are fully flushed downstream.
• DDL Messages: Use individual acknowledgment (AckID) so that only the DDL message itself is acknowledged, leaving any unflushed DMLs of other tables unacknowledged and safe for redelivery upon restart.

To implement this, you can modify WriteMessage to return an acknowledgment strategy or type, and apply the appropriate ack method in consumer.go:

// In writer.go
type AckStrategy int
const (
	AckStrategyNone AckStrategy = iota
	AckStrategyIndividual
	AckStrategyCumulative
)

func (w *writer) WriteMessage(ctx context.Context, message pulsar.Message) AckStrategy {
	// ...
	if needFlush {
		if w.Write(ctx, messageType) {
			if messageType == common.MessageTypeResolved {
				return AckStrategyCumulative
			}
			return AckStrategyIndividual
		}
	}
	return AckStrategyNone
}

And in consumer.go:

			ackStrategy := c.writer.WriteMessage(ctx, consumerMsg)
			switch ackStrategy {
			case AckStrategyCumulative:
				err := c.pulsarConsumer.AckIDCumulative(consumerMsg.ID())
				if err != nil {
					log.Panic("Error ack message cumulatively", zap.Error(err))
				}
			case AckStrategyIndividual:
				err := c.pulsarConsumer.AckID(consumerMsg.ID())
				if err != nil {
					log.Panic("Error ack message individually", zap.Error(err))
				}
			}

[ti-chi-bot]

[FORMAT CHECKER NOTIFICATION]

Notice: To remove the do-not-merge/needs-linked-issue label, please provide the linked issue number on one line in the PR body, for example: Issue Number: close #123 or Issue Number: ref #456.

_{📖 For more info, you can check the "Contribute Code" section in the development guide.}