[
https://issues.apache.org/jira/browse/KAFKA-21363?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18133807#comment-18133807
]
Matthias J. Sax commented on KAFKA-21363:
-----------------------------------------
{quote}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.
{quote}
That's what I expected...
{quote}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.
{quote}
That would be a pickle... Maybe [~jolshan] can chime in? (Given that
punctuations and EOS are in general kind of an anti-pattern, I would personally
be ok to only fix for newer version is we need a broker change).
> 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)
