Skip to content

consumer: change message acknowledgment to cumulative for pulsar consumer #5537

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
wk989898 wants to merge 1 commit into
base: master
Choose a base branch
from
Open

consumer: change message acknowledgment to cumulative for pulsar consumer #5537

Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
2 changes: 1 addition & 1 deletion cmd/pulsar-consumer/consumer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -114,7 +114,7 @@ func (c *consumer) readMessage(ctx context.Context) error {
if !needCommit {
continue
}
err := c.pulsarConsumer.AckID(consumerMsg.ID())
err := c.pulsarConsumer.AckIDCumulative(consumerMsg.ID())

@gemini-code-assist gemini-code-assist Bot Jun 29, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

critical

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))
				}
			}

if err != nil {
log.Panic("Error ack message", zap.Error(err))
}
Expand Down
Loading