Eswarar Siva created KAFKA-21363:
------------------------------------
Summary: 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
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)
