[
https://issues.apache.org/jira/browse/KAFKA-21402?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18136529#comment-18136529
]
seongsu shin commented on KAFKA-21402:
--------------------------------------
Hi Sean, unless you're already working on this, I'd like to pick it up.
Proposed fix: don't install a partially loaded map. {{loadTransactionMetadata}}
would propagate the failure instead of swallowing it, and
{{loadTransactionsForTxnTopicPartition}} would log the error, remove the
partition from {{loadingPartitions}} and skip {{addLoadedTransactionsToCache}}
and the {{PREPARE_COMMIT}}/{{PREPARE_ABORT}} marker sends. Requests for
transactional ids in that partition then get {{NOT_COORDINATOR}} until the
partition is loaded again (next leader change), instead of being served from
stale state that then gets written back to the log.
On surfacing failures: agreed that a load-failure metric would need a KIP, so
I'd leave it out of this change. Until such a metric exists, this at least
turns a silent corruption into a visible unavailability: clients keep getting
{{NOT_COORDINATOR}} for that partition, alongside the ERROR log, rather than
the coordinator acting on stale state. Retrying the load automatically is also
out of scope here, but I'm happy to add it if preferred.
I have a unit test that reproduces it (a read failure after the first batch
installs a cache entry holding the {{ONGOING}} state) and passes with the fix.
> Transaction coordinator installs partially loaded transaction state when
> loading __transaction_state fails
> ----------------------------------------------------------------------------------------------------------
>
> Key: KAFKA-21402
> URL: https://issues.apache.org/jira/browse/KAFKA-21402
> Project: Kafka
> Issue Type: Bug
> Affects Versions: 0.11.0.0
> Reporter: Sean Quah
> Priority: Major
>
> If \{{TransactionStateManager.loadTransactionMetadata}} throws while reading
> a \{{__transaction_state}} partition, it logs the error and returns the
> partially built map:
> {code:scala}
> } catch {
> case t: Throwable => error(s"Error loading transactions from
> transaction log $topicPartition", t)
> }
> }
> loadedTransactions
> {code}
> {\{loadTransactionsForTxnTopicPartition}} then installs the partial map as
> the coordinator's cache and completes the load. For any transactional ID with
> records past the failure point, the in-memory cache holds an older state than
> the log does. Any transition the coordinator makes from that stale state is
> appended to the log and supersedes the correct, newer records. The stale
> state is then persisted, and it survives coordinator failover and reloads.
> Observed case (TV2):
> # Loading failed part-way with \{{OutOfMemoryError}}. The partial map held a
> transactional ID as \{{ONGOING}} at epoch 22969, a transaction that had
> completed hours earlier.
> # The coordinator timed it out and wrote \{{PREPARE_ABORT}} at epoch 22970.
> # The data partition had already seen the producer at epoch 24018, so it
> rejected the marker with \{{InvalidProducerEpochException}}.
> # \{{PREPARE_ABORT}} is now the latest record for the ID, so every later
> coordinator hits the same failure. The transaction is stuck in
> \{{PrepareAbort}}, and \{{InitProducerId}} returns
> \{{CONCURRENT_TRANSACTIONS}} indefinitely.
> Relates to KAFKA-21349.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
