[
https://issues.apache.org/jira/browse/KAFKA-21410?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jun Rao resolved KAFKA-21410.
-----------------------------
Fix Version/s: 4.5.0
Resolution: Fixed
merged the PR to trunk
> Log cleaner ignores a committed tombstone after an earlier aborted
> transaction of the same producer; the deleted key comes back
> -------------------------------------------------------------------------------------------------------------------------------
>
> Key: KAFKA-21410
> URL: https://issues.apache.org/jira/browse/KAFKA-21410
> Project: Kafka
> Issue Type: Bug
> Components: log
> Affects Versions: 4.4.0
> Environment: macOS arm64, OpenJDK 17, single KRaft node; also 1
> controller + 3 brokers (RF 3) on Linux (Docker) and OpenBSD 8.0
> Reporter: Mykyta Bozhenko
> Assignee: Mykyta Bozhenko
> Priority: Major
> Fix For: 4.5.0
>
> Attachments: LogCleanerTest.patch, cluster-observations.txt,
> fix.patch, repro-output.txt, repro.sh, unit-test-output.txt
>
>
> On a compacted topic written by transactional producers, the log cleaner can
> ignore a tombstone (or a newer value) that a producer committed. It keeps the
> key's older value, removes the tombstone after
> delete.retention.ms, and from then on a consumer reading from the beginning
> gets the deleted key back with its old value. Nothing is logged at WARN or
> ERROR. Each replica runs its own cleaner; in a run with
> three replicas all three ended up with the old value
> ([^cluster-observations.txt]).
> Expected, per the compaction guarantee in the documentation
> (https://kafka.apache.org/44/design/design/#what-guarantees-does-log-compaction-provide):
> "Any consumer progressing from the start of the log will
> see at least the final state of all records in the order they were written".
> Actual: the consumer reads the old value.
> It happens when a cleaning pass stops in the middle of a segment because its
> offset map is full (log.cleaner.dedupe.buffer.size; about 5 million keys per
> cleaner thread with the default 128 MiB), the same
> producer id has an aborted transaction that ended earlier in that segment,
> before the position where the next pass starts reading, and the next pass
> reaches a later committed batch of that producer. Typical
> case: exactly-once writers such as Kafka Streams changelogs with
> exactly_once_v2, where aborts follow rebalances, timeouts or crashes.
> h3. Reproduction (attached)
> * [^repro.sh]: one broker+controller from a Kafka distribution, about 3
> minutes. {{KAFKA_HOME=<kafka dir> bash repro.sh p1}} shows the bug; {{p2}} is
> the control, where a different producer commits the
> tombstone. Output on 4.4.0: [^repro-output.txt]. With p1 the consumer returns
> K=v1 although K was deleted; with p2 it does not return K.
> * [^LogCleanerTest.patch]: a unit test for trunk 8aa57006ff. Run {{git apply
> LogCleanerTest.patch}} and {{./gradlew :storage:test --tests
> 'org.apache.kafka.storage.internals.log.LogCleanerTest.testCommittedTombstone*'}}.
> The same-producer case fails, the other-producer case passes:
> [^unit-test-output.txt].
> h3. Cause
> Cleaner#buildOffsetMap seeds CleanedTransactionMetadata with
> log.collectAbortedTransactions(start, end)
> ([Cleaner.java#L730-L732|https://github.com/apache/kafka/blob/8aa57006ff5da4844776d1047bfee1619faea47f/storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java#L730-L732]),
> and
> UnifiedLog#collectAbortedTransactions reads the transaction indexes from the
> segment at or below start but uses the log start offset, not start, as the
> lower bound
> ([UnifiedLog.java#L1676-L1678|https://github.com/apache/kafka/blob/8aa57006ff5da4844776d1047bfee1619faea47f/storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java#L1676-L1678],
> [LocalLog.java#L628-L633|https://github.com/apache/kafka/blob/8aa57006ff5da4844776d1047bfee1619faea47f/storage/src/main/java/org/apache/kafka/storage/internals/log/LocalLog.java#L628-L633]).
> So an aborted
> transaction that ended before start is seeded as ongoing. Reading starts at
> the offset index entry at or below start
> ([Cleaner.java#L788|https://github.com/apache/kafka/blob/8aa57006ff5da4844776d1047bfee1619faea47f/storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java#L788]);
> if the ABORT marker lies
> before that position it is never read, and only an ABORT marker clears the
> producer
> ([CleanedTransactionMetadata.java#L83-L155|https://github.com/apache/kafka/blob/8aa57006ff5da4844776d1047bfee1619faea47f/storage/src/main/java/org/apache/kafka/storage/internals/log/CleanedTransactionMetada
> ta.java#L83-L155]). Every later transactional batch of that producer,
> committed or not, is then treated as aborted and its keys are left out of the
> offset map
> ([Cleaner.java#L809-L814|https://github.com/apache/kafka/blob/8aa57006ff5da4844776d1047bfee1619faea47f/storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java#L809-L814]),
> so the older
> value of the key is kept. The lower bound has been the log start offset since
> KAFKA-5196 (0.11.0.0); KAFKA-13727 fixed the upper bound of the same
> partial-pass case in Cleaner#cleanSegments.
> h3. Proposed fix
> [^fix.patch]: in Cleaner#buildOffsetMap, seed only the aborted transactions
> with lastOffset() >= start. Cleaner#cleanSegments keeps the aborted
> transactions of the whole segment (it rebuilds the transaction
> index of the new segment, see KAFKA-13727), so
> UnifiedLog#collectAbortedTransactions does not change. With the patch, both
> new tests and LogCleanerTest, LogCleanerManagerTest and
> LogCleanerIntegrationTest
> pass.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
