Skip to content

feat: support transactional consumers in the Reader - #1446

Open
Darkbat91 wants to merge 2 commits into
segmentio:mainfrom
Darkbat91:fix/skip-control-batches
Open

feat: support transactional consumers in the Reader#1446
Darkbat91 wants to merge 2 commits into
segmentio:mainfrom
Darkbat91:fix/skip-control-batches

Conversation

@Darkbat91

Copy link
Copy Markdown

Problem

kafka-go did not implement the client side of Kafka transactions. We found this when building a go service interconnected with NiFi and transactions enabled on the put Kafka processor.

  1. Control batches are delivered as messages. Any consumer of a topic written by a
    transactional producer sees empty messages interleaved with real data. The
    protocol requires clients to hide them; kafka-go already parses the Control
    attribute bit into messagesHeader.v2.attributes and never tests it.
  2. IsolationLevel: ReadCommitted does nothing (Aborted transactional data included in response when setting read_committed isolation level in reader #1332). The broker bounds
    the fetch at the last stable offset and returns an AbortedTransactions
    list; the client filters. kafka-go parsed that list into a variable it never
    read.

Change

Ports the Java client's CompletedFetch algorithm into messageSetReader, the
single funnel Reader, Batch and Conn all pass through.

Hidden batches consume offsets, so Batch tracks the highest one skipped and
resumes past it.

No exported API is added. dialer.go and consumergroup.go are untouched.
ReadUncommitted stays the default, matching Kafka.

Notes

Hide control batches and, under ReadCommitted, drop records of aborted
transactions in messageSetReader. Track the offsets hidden and emptied
batches consume so the reader resumes past them.

Closes segmentio#1332.
@Darkbat91
Darkbat91 force-pushed the fix/skip-control-batches branch from f6b4ebb to 4e1682c Compare July 31, 2026 20:28
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Aborted transactional data included in response when setting read_committed isolation level in reader

1 participant