[
https://issues.apache.org/jira/browse/KAFKA-21363?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18133200#comment-18133200
]
Eswarar Siva commented on KAFKA-21363:
--------------------------------------
[~mjsax] I had a look at what an empty map does, in case it helps.
On the producer side, if I read it right, sendOffsetsToTransaction returns
straight away when the map is empty, so neither AddOffsetsToTxn nor
TxnOffsetCommit goes out. That seems to have come in with KAFKA-13319 (3.1.0),
to avoid markers piling up in __consumer_offsets.
On the broker side, KafkaApis.handleTxnOffsetCommitRequest looks like it
answers a TxnOffsetCommit without topics by itself, without going to the group
coordinator (I saw the same on 3.0 and trunk). So my understanding is that
allowing an empty map would need a broker change too, and older brokers would
still skip the check. Please correct me if I missed something there.
That makes me think one real offset might be the simpler route. From what I see
the group coordinator validates member id and generation once per request
(OffsetMetadataManager.commitTransactionalOffset), so one offset could already
be enough for the check.
About the value, I wonder if StreamTask.findOffsetAndMetadata could be reused.
It takes the head of the task's queue, or the next offset from the last fetch
when the queue is empty, not the consumer position. If nothing was processed
since the last commit, I think that comes out as the offset committed last time
(maybe moved past control records), so the commit would be idempotent. It
returns empty for a partition where nothing was fetched yet since the
assignment, which might need some thought.
Just an idea, you know this code much better than me, so happy to be told if
this doesn't hold .
> Streams EOS: member cut off from the group coordinator keeps committing
> punctuator writes unfenced, and its store diverges from the changelog after a
> clean restart
> -------------------------------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: KAFKA-21363
> URL: https://issues.apache.org/jira/browse/KAFKA-21363
> Project: Kafka
> Issue Type: Bug
> Components: streams
> Affects Versions: 4.4.0
> Reporter: Eswarar Siva
> Priority: Major
>
> With exactly_once_v2 and the classic group protocol, a Streams instance that
> cannot reach the group coordinator keeps its active task and keeps committing
> what its wall clock punctuator writes, while the group has already given the
> same task to another instance. Both instances write the same changelog
> partition and neither is fenced. If the cut off instance is then closed
> cleanly, it keeps its state directory, and on restart it restores only from
> its own checkpoint offset onwards. The other instance's records below that
> offset are never applied, so the store no longer matches its changelog.
> Reproduced twice end to end with 4.4.0 clients against 4.3.0 brokers. Line
> numbers below are at 4.4.0, the same code is on trunk. The streams group
> protocol (KIP-1071) was not tested.
> Why the instance keeps the task:
> -> ConsumerCoordinator.poll calls pollHeartbeat (525), then returns early
> while the coordinator is unknown (526-528), before the rejoin check at 530.
> The task is never revoked or marked lost.
> -> In the heartbeat thread (AbstractCoordinator 1513-1531) the
> coordinatorUnknown() branch only looks up the coordinator again and backs
> off. The session timeout branch at 1523 only marks the coordinator unknown,
> and the poll timeout leave at 1528 is an else-if after 1513, so it is not
> reached while the coordinator is unknown.
> Why EOS does not fence it:
> -> A punctuator only commit has no input offsets.
> TaskExecutor.commitOffsetsOrTransaction still commits because a transaction
> is in flight (181-186) and passes an empty map to
> StreamsProducer.commitTransaction, which calls sendOffsetsToTransaction (249).
> -> KafkaProducer.sendOffsetsToTransaction only contacts the group coordinator
> when the map is not empty (800, from KAFKA-13319). The transaction commits
> against the transaction coordinator, which is still reachable, and group
> membership is never checked.
> -> The two instances have different transactional ids, so they do not fence
> each other either. Their changelog records interleave.
> Why the store stays wrong:
> -> A clean close keeps the task directory. StateManagerUtil only wipes on an
> unclean close (168-169, 183-189; transactional state stores were off).
> -> On restart ProcessorStateManager initialises the store offset from the
> store's own committed offset (334-338) and StoreChangelogReader seeks to that
> offset + 1 (1028-1032). Nothing below it is read again.
> Setup:
> # Three KRaft brokers. __consumer_offsets on broker 2, __transaction_state on
> broker 1, the input topic (1 partition) and the changelog on broker 3.
> # Two instances A and B, exactly_once_v2, commit.interval.ms=100,
> session.timeout.ms=6000, heartbeat.interval.ms=2000, one persistent
> KeyValueStore, a 1 s wall clock punctuator that puts a new key each time. No
> input traffic.
> # Blackhole the route from the instance that owns 0_0 (A) to broker 2 only.
> Within 30 s B has 0_0 active while A still reports 0_0 active, stays RUNNING
> and keeps punctuating.
> # After one to two minutes close A cleanly (end of run in the first run,
> SIGTERM in the second). One "Closed clean", task directory kept.
> # Restart A on the same state directory. Restore a third instance from an
> empty state directory as the reference.
> Results:
> ||Run||A records in changelog||B records in changelog||A restarted from
> checkpoint||B keys in reference store||B keys in restarted A||Missing||
> |1|208 (offsets 0 to 626)|554 (offsets 204 to 1522)|626|554|448|106, all at
> offsets 204 to 624|
> |2|178 (offsets 0 to 522)|85 (offsets 188 to 524)|522|85|1|84, all at offsets
> 188 to 520|
> Restarted A logs "State store store initialized from checkpoint with offset
> 626" (run 1) and restores from 627 on. A's own keys are all there. Nothing
> repairs this later since the checkpoint looks valid to Streams.
> Not run, only follows from the same path: a key B deleted below A's
> checkpoint offset keeps A's old value after the restart.
> What I think could be done: check group membership on the transactional
> commit even when there are no offsets to send, or treat a coordinator that
> stays unknown past the session timeout as losing the assignment, or let the
> poll timeout leave run outside the coordinatorUnknown() branch. Not sure
> which one fits best.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
