[
https://issues.apache.org/jira/browse/KAFKA-21363?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18132694#comment-18132694
]
Eswarar Siva commented on KAFKA-21363:
--------------------------------------
[~mjsax] Thanks , makes sense. I also checked whether an empty TxnOffsetCommit
could do the job, but the broker answers it straight away without going to the
group coordinator (same on 3.0 and 4.4), so we do need real offsets in there.
One thing to be careful about is which offset gets sent. If it's the consumer
position, records that were fetched but not processed yet would be skipped
after a failover. StreamTask.findOffsetAndMetadata already gives the head of
the task's queue and committableOffsetsAndMetadata already commits all input
partitions when processor metadata needs a commit. Maybe the same path can be
reused when a task has writes under eos v2 but consumed nothing? Then the extra
AddOffsetsToTxn/TxnOffsetCommit only happens in that case.
> 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)
