[
https://issues.apache.org/jira/browse/KAFKA-21364?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18133656#comment-18133656
]
Bill Bejeck commented on KAFKA-21364:
-------------------------------------
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
> 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)
