Mykyta Bozhenko created KAFKA-21410:
---------------------------------------
Summary: 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
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)
