I’ve spent quite a bit of time looking for an answer to the following question and found one post in this regard but not sure it applies to the consume-transform-publish model enabled by v0.11+.
Assume Kafka v1.1 and three publishers writing to one processing “input” topic partition. Each is publishing a “batch” of messages and commits only by batch. Further the processing consumer of that input partition must ONLY transform (and commit to an output topic) the entire batch or none. Each batch message however may be interleaved with other batch messages in the topic partition. For example, a two-record batch by three publishers may look like:
P1 writes Message p1-m1
P2 writes Message p2-m1
P1 writes Message p1-m2
P1 commits
P3 writes Message p3-m1
P2 writes Message p2-m2
P2 commits.
P3 writes Message p3-m2
(P3 not yet committed.)
The partition now looks like the following with p1 and p2 messages committed by their publisher and p3 messages not yet committed:
Offset 1: p1-m1
Offset 2: p2-m1
Offset 3: p1-m2
Offset 4: p3-m1 not committed
Offset 5: p2-m2
Offset 6: p3-m2 not committed
The processing consumer polls read_committed messages only and therefore receives messages at offsets 1 through 3. It recognizes (from the data) that it has one complete batch p1 and one incomplete batch p2. It transforms the P1 records, begins a new transaction on an output topic, writes to the output topic, passes offsets 1 and 3 from the input topic partition via sendOffsetsToTransaction, and commits the batch.
Meanwhile P3 then commits – so now all messages have been committed - but…. the consuming processor decides at this point to go away leaving the consuming offset 2 uncommitted.
What happens? On the one hand, we have to pass each consuming offset to sendOffsetToTransaction implying that each is independently marked as committed for consumption. If so, then that seems to imply that the next consumer to access the input partition will poll and receive messages at offset 2, 4, 5 and 6, right?
Or is the message at offset 2 lost because we committed offset 3 before processing offset 2? If so, what is the reason of having to pass each of the consuming offsets in the call to sendOffsetsToTransaction - when simply passing the last one should suffice?