Eswarar Siva created KAFKA-21364:
------------------------------------
Summary: Streams EOS: input records skipped after a commit
timeout, the offset reset of the revived task is lost when the enforced
rebalance closes it
Key: KAFKA-21364
URL: https://issues.apache.org/jira/browse/KAFKA-21364
Project: Kafka
Issue Type: Bug
Components: streams
Affects Versions: 4.4.0, 4.2.1
Reporter: Eswarar Siva
With exactly_once_v2 and the classic group protocol, a commit that times out
against the transaction coordinator can make Streams skip input records. The
input offsets get committed past records that were never processed, so they
never reach the store or the changelog.
When it happens: the commit has to fail with a TimeoutException while the
transaction is still open at the coordinator. So transaction.timeout.ms has to
be above the producer max.block.ms, and the transaction coordinator has to be
unreachable for longer than max.block.ms. With the Streams EOS defaults
(transaction.timeout.ms 10000, max.block.ms 60000) the coordinator aborts
first, the next commit fails with InvalidTxnStateException and the thread dies,
and I saw no loss. With transaction.timeout.ms=120000 and max.block.ms left at
60000, a 70 s outage was enough.
Reproduced on 4.2.1 and 4.4.0. Line numbers at 4.4.0, the same code is on trunk
(not run on trunk). The streams group protocol was not tested; StreamThread
only enforces the rebalance below when streamsRebalanceData is empty.
What happens:
-> The commit fails with "Timeout expired after 60000ms while awaiting
EndTxn(true)" and TaskExecutor marks the task corrupted (188-200).
-> TaskManager.handleCorruption closes the task dirty, which drops the records
it fetched but has not processed yet, adds the input partitions for an offset
reset (TaskManager 320), revives the task and puts it in the tasks pending
initialization (325).
-> StreamThread enforces a rebalance, "Active task(s) got corrupted. Triggering
a rebalance." (1025-1027).
-> On the new assignment handleTasksPendingInitialization closes every task
pending initialization (TaskManager 514-525), the revived one included, and a
new StreamTask is created for the same task id. Its resetOffsetsForPartitions
starts empty (StreamTask 198).
-> The seek back to the committed offset only runs for partitions in that set
(StreamTask 1054-1069), so there is no seek. The new task starts at the
consumer position and the next commit moves the input offset past the dropped
records.
Setup:
# Three KRaft brokers. __transaction_state on broker 1, __consumer_offsets on
broker 2, input topic (1 partition) and changelog on broker 3.
# One instance, one thread, exactly_once_v2, commit.interval.ms=100, producer
transaction.timeout.ms=120000, max.block.ms default. A processor that puts
every input record into a persistent KeyValueStore and sleeps 2 ms per record,
so a commit lands in the middle of a fetched batch.
# Produce 24,000 records with unique keys f-0 to f-23999 at 200 per second.
# 40 s into the feed blackhole the route from the app to broker 1 only, for 70
s, then remove it.
# Compare the distinct keys in the changelog (read_committed) with the input,
and the group's committed offset with the log end offset.
Results:
||Version||Old task last processed offset||New task first processed
offset||Seek on the input after startup||Keys missing from the changelog||Group
committed||
|4.4.0|8102|8200|none|97, f-8103 to f-8199|24,000 of 24,000|
|4.2.1|8055|8200|none|144, f-8056 to f-8199|24,000 of 24,000|
The same happened in 9 more runs with max.block.ms=3000,
transaction.timeout.ms=60000 and two 6 s outages, on both versions, with one
instance and with three instances on six partitions, and also without the
processing delay. Those runs lost 39 to 365 records each. Runs without an
outage lost nothing.
Two things that point at the lost reset rather than at the transaction:
-> In one run the revived task instance was initialized again instead of being
replaced. The consumer logged "Seeking to offset 8223" and nothing was lost.
-> In a lossy run the changelog has the transaction of producer epoch 53 ending
at f-8222 and the next one, epoch 54, starting at f-8400. f-8223 to f-8399 are
in no batch at all and there is no abort marker, so they were never written.
They were fetched, dropped by the dirty close, and skipped.
KAFKA-21159 on trunk clears the next offsets to consume on close so a revived
task cannot commit a stale offset after the consumer is seeked back. Here the
seek back never happens, so it does not cover this.
What I think could be done: keep the pending offset reset when
handleTasksPendingInitialization closes a revived task, for example by carrying
it into the new task, or seek those input partitions to their committed offsets
when the revived task is closed.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
