[
https://issues.apache.org/jira/browse/KAFKA-21364?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18133656#comment-18133656
]
Bill Bejeck edited comment on KAFKA-21364 at 10/28/26, 3:58 PM:
----------------------------------------------------------------
Thanks [~zoro30102000] for raising this. I don't think we should allow
{{transaction.timeout.ms}} to be greater than {{{}max.block.ms{}}}. The client
gives up while the transaction is still open at the coordinator, which is an
ambiguous state: the transaction might still commit or abort. We could add a
check in \{{StreamsConfig }}and for now it would log a warning and fail at
startup in the next major release.
That said, you've found a real bug that needs fixing however it's reached. When
{{handleTasksPendingInitialization}} closes the revived task, the pending
offset reset is lost, so the new task never seeks back to the committed offset.
We should get this addressed ASAP so I've assigned this to myself, but let me
know if you'd like to the PR
was (Author: bbejeck):
Thanks [~zoro30102000] for raising this. I don't think we should allow
{{transaction.timeout.ms}} to be greater than {{{}max.block.ms{}}}. The client
gives up while the transaction is still open at the coordinator, which is an
ambiguous state: the transaction might still commit or abort. We could add a
check in {{StreamsConfig }}and for now it would log a warning and fail at
startup in the next major release.
That said, you've found a real bug that needs fixing however it's reached. When
{{handleTasksPendingInitialization}} closes the revived task, the pending
offset reset is lost, so the new task never seeks back to the committed offset.
> 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.2.1, 4.4.0
> Reporter: Eswarar Siva
> Assignee: Bill Bejeck
> Priority: Major
>
> 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)
