[
https://issues.apache.org/jira/browse/KAFKA-21363?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18133828#comment-18133828
]
Eswarar Siva commented on KAFKA-21363:
--------------------------------------
??That would be a pickle... Maybe Justine Olshan 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).??
Could I ask for your view on the anti-pattern point, since it's how we run
expiry today? Two of our EOS v2 services keep a large key-value store (over 200
million keys in our load tests) whose entries have to go after a fixed
retention say for ex 100 days ... A persistent key-value store has no TTL, and
expiring at the RocksDB level would bypass the changelog, so the entries would
come back after a restore. So we keep a second store as an expiry index (expiry
bucket plus key) and a wall-clock punctuator every few minutes that range-scans
it up to now and deletes from both stores through the store API, so both
changelogs get the tombstones.
We already learned one lesson there: a cleanup run of 14 to 18 seconds held the
transaction open past the 10 second transaction timeout, the producer got
fenced and the task restored from scratch, again and again. Each run now has a
time budget and keeps a watermark, so the next run picks up where the last one
stopped.
What would you recommend instead? A stream-time punctuator or a bounded cleanup
inside process() would commit input offsets together with the deletes, but both
stop when a partition goes quiet. A periodic tick record per partition on an
input topic would keep the cleanup running as normal processing... Is that the
pattern you'd suggest, or is there something planned for TTL on key-value
stores? (for us , expired entry cleanup is a pain at our volume). if there is a
better place to have this discussion , we can do that too if you want .
> 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)
