KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes by m1a2st · Pull Request #21379 · apache/kafka

m1a2st · 2026-01-31T07:13:33Z

Previously, Cleaner#cleanSegments always wrote into a single
destination segment. If that destination segment overflowed in offset
range or size during cleaning—for example, because decompression
expanded the data or because multiple source segments were compacted
into the same destination segment—the cleaner would throw
LogSegmentOffsetOverflowException, invoke splitOverflowedSegment to
split the source segment, abort the entire cleaning pass, and retry
later.

This change removes that restart path. When the cleaner detects that the
current destination segment is about to exceed the configured limits, it
now finalizes the current segment (onBecomeInactiveSegment + flush)
and immediately continues writing into a new destination segment
starting from the overflow point.

Reviewers: Jun Rao junrao@gmail.com, Chia-Ping Tsai
chia7712@gmail.com

chia7712

@m1a2st thanks for this fix

junrao

@m1a2st : Thanks for the PR. Left a comment.

junrao

@m1a2st : Thanks for the updated PR. A couple of more comments.

# Conflicts: # checkstyle/suppressions.xml

junrao

@m1a2st : Thanks for the updated PR. A few more comments.

m1a2st · 2026-03-22T12:34:29Z

I noticed that the integration test we has earlier was removed. Was this intentional?

Since the integration test is expensive and generating 2GB of data is not ideal, I removed it.

chia7712 · 2026-03-22T18:15:45Z

-rw-r--r--. 1 astraea astraea        378 Mar 22 18:06 00000000000002538495.snapshot
-rw-r--r--. 1 astraea astraea      51360 Mar 22 18:06 00000000000002538495.timeindex
-rw-r--r--. 1 astraea astraea      51372 Mar 22 18:06 00000000000002538495.timeindex.deleted
-rw-r--r--. 1 astraea astraea        448 Mar 22 18:06 00000000000002769043.index
-rw-r--r--. 1 astraea astraea     589184 Mar 22 18:06 00000000000002769043.log
-rw-r--r--. 1 astraea astraea         48 Mar 22 18:06 00000000000002769043.timeindex
-rw-r--r--. 1 astraea astraea    1661440 Mar 22 18:06 00000000000002769100.index
-rw-r--r--. 1 astraea astraea 2147474716 Mar 22 18:06 00000000000002769100.log
-rw-r--r--. 1 astraea astraea        378 Mar 22 18:06 00000000000002769100.snapshot
-rw-r--r--. 1 astraea astraea      49248 Mar 22 18:06 00000000000002769100.timeindex
-rw-r--r--. 1 astraea astraea    1661456 Mar 22 18:06 00000000000002999960.index
-rw-r--r--. 1 astraea astraea 2147473581 Mar 22 18:06 00000000000002999960.log

Exercised this patch via a producer flow with positive results. Instead of throwing a exception, the cleaner now yield undersized log files. Typically, these files will be naturally merged in future compaction passes as the dataset accrues duplicate keys

# Conflicts: # storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java

junrao

@m1a2st : Thanks for the updated PR. Very clean PR! Just a couple of minor comments.

junrao · 2026-03-27T21:49:31Z

-    LogTestUtils.initializeLogDirWithOverflowedSegment(dir)
+    val sourceSegments = log.logSegments.asScala.take(2).toSeq
+    val singleBatchSize = sourceSegments.head.log.batches.asScala.map(_.sizeInBytes).max
+    // Allow 1.5 batches: fits the first batch, but adding a second batch overflows.


This line seems redundant given the comment in next line.

junrao · 2026-03-27T22:23:01Z

+            cleanedSegments.add(currentCleaned);
+
+            // swap in all cleaned segments (maybe multiple if overflow occurred)
+            logger.info("Swapping in {} cleaned segment(s) for segment(s) {} in log {}", cleanedSegments.size(), segments, log);


Could we log the cleaned segments, instead of just the size?

chia7712 · 2026-03-27T22:40:31Z

@m1a2st

Could you please take a look at these failing e2e tests?

tests/kafkatest/tests/connect/connect_distributed_test.py::ConnectDistributedTest.test_exactly_once_source
tests/kafkatest/tests/core/consume_bench_test.py::ConsumeBenchTest.test_consume_bench

I noticed they failed during my test run. Since I'm traveling right now, I haven't had the chance to dive deeper and verify if they are related to this PR.

m1a2st · 2026-03-28T05:26:00Z

tests/kafkatest/tests/core/consume_bench_test.py::ConsumeBenchTest.test_consume_bench

When I ran this test on the trunk branch, they were flaky as well.

also create Jira to trace this test.
https://issues.apache.org/jira/browse/KAFKA-20365

tests/kafkatest/tests/connect/connect_distributed_test.py::ConnectDistributedTest.test_exactly_once_source

This one all passed

chia7712 · 2026-03-28T13:38:48Z

-            logger.info("Swapping in cleaned segment {} for segment(s) {} in log {}", cleaned, segments, log);
-            log.replaceSegments(List.of(cleaned), segments);
        } catch (LogCleaningAbortedException e) {
+            cleanedSegments.forEach(segment -> {


Could you eliminate the duplicate code?

Stream.concat(cleanedSegments.stream(), Stream.of(currentCleaned)) .distinct() .forEach(segment -> { try { segment.deleteIfExists(); } catch (Exception deleteException) { e.addSuppressed(deleteException); } }); throw e;

junrao

@m1a2st : Thanks for the updated PR. LGTM. I will wait to see if Chia-Ping has more comments.

chia7712 · 2026-03-31T00:30:38Z

I will wait to see if Chia-Ping has more comments.

I will merge it after all E2E tests pass. I just reran them last night

…level changes (apache#21379) We add a new map to record which topic partitions have experienced overflow. When an overflow occurs, the next time the group is processed, we reduce the segment size by a factor of 0.9 to prevent the overflow from happening again. If the partition still overflows, we continue to multiply the ratio by 0.9 on subsequent attempts until the partition is successfully cleaned. Reviewers: Jun Rao <junrao@gmail.com>, Chia-Ping Tsai <chia7712@gmail.com>

m1a2st · 2026-03-31T11:37:04Z

Thanks for @junrao and @chia7712 review!

…level changes (apache#21379) We add a new map to record which topic partitions have experienced overflow. When an overflow occurs, the next time the group is processed, we reduce the segment size by a factor of 0.9 to prevent the overflow from happening again. If the partition still overflows, we continue to multiply the ratio by 0.9 on subsequent attempts until the partition is successfully cleaned. Reviewers: Jun Rao <junrao@gmail.com>, Chia-Ping Tsai <chia7712@gmail.com>

junrao · 2026-08-06T23:22:14Z

@m1a2st : The description of the PR is outdated and doesn't match the implementation. Could you update it? The outdated description is already in the git log. Not sure if it can be fixed.

chia7712 · 2026-08-07T00:23:17Z

@junrao sorry for the misleading message. We could revert the original patch and then commit a new one with an updated message. WDYT?

junrao · 2026-08-07T15:55:54Z

@chia7712 It's probably not worth it. We can just update the description in the PR.

chia7712 · 2026-08-08T06:36:32Z

It's probably not worth it. We can just update the description in the PR.

Copy that. We will update it after CoC Asia.

junrao

@m1a2st and @chia7712 : There seems to be a bug in this PR and I am not sure how to fix it easily. We probably want to revert this PR from 4.4 and trunk for now.

junrao · 2026-08-11T20:32:30Z

                outputBuffer.flip();
                MemoryRecords retained = MemoryRecords.readableRecords(outputBuffer);
+
+                // While groupSegmentsBySize() ensures source segments don't exceed Integer.MAX_VALUE,


Claude found a bug in this PR. The problem is that checkBatchRetention() modifies transactionMetadata, which stores the ongoing txn state and is reused during cleaning. When we detect an overflow, we rewind the cleaning position to Optional.of(position - result.bytesRead()), but the corresponding transactionMetadata is not rewound. When the cleaning resumes, since transactionMetadata is not accurate, we can have all sorts of bad outcomes. For example, an aborted record could now appear as a committed record after cleaning.

Nice find. This approach does corrupt the txn index ... I think we could revert it from 4.4 first, and then we could discuss the better approach for trunk.

@m1a2st would you mind opening a PR for 4.4? it has some conflicts.

One possibility to fix this issue in trunk is to return the filtered result to the caller on overflow. Instead of rewinding the input segment, the caller will roll a new segment and append the returned result to the new segment. We still need to decouple the updating of the txn index in the filterTo logic. Claude suggested the following.

Stage the index appends instead of writing them during the filter. 1. In CleanedTransactionMetadata, replace the immediate cleanedIndex.ifPresent(index -> index.append(...)) with adding to a List<AbortedTxn> pendingAbortedTxns, plus a flushPendingTo(TransactionIndex) that appends and clears. 2. In cleanInto, flush that list right after a successful dest.append(result.maxOffset(), retained). 3. On overflow, return the saved records + maxOffset and leave the pending list untouched. cleanSegments finalizes the outgoing segment, creates the new one (base offset = first batch of the retained buffer), appends the saved records, then flushes the pending list into the new segment's index. filterTo then runs exactly once per chunk, so the destructive parts of the state machine — ongoingAbortedTxns.remove, ongoingCommittedTxns.remove, consumeAbortedTxnsUpTo's poll — are each consumed once, and the stats double-count goes away with it. The index entry lands in whichever segment its data landed in, by construction. That's the invariant you expected the code to already have.

There is a bug, fyi: #21379 (review) Reviewers: Parker Chang <parkerhiphop027@gmail.com>, Chia-Ping Tsai <chia7712@gmail.com>

m1a2st added 2 commits January 31, 2026 14:23

temp version

21f121c

update the test

23c7bf0

github-actions Bot added triage PRs from the community core Kafka Broker storage Pull requests that target the storage module labels Jan 31, 2026

spotless apply

9ad125d

chia7712 reviewed Jan 31, 2026

View reviewed changes

chia7712 requested review from jolshan and junrao January 31, 2026 19:38

addressed by comments

9b95afe

github-actions Bot removed the triage PRs from the community label Feb 1, 2026

chia7712 reviewed Feb 9, 2026

View reviewed changes

Comment thread storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java Outdated

m1a2st added 2 commits February 10, 2026 20:02

Merge branch 'trunk' into KAFKA-20036

d92d2d6

addressed by comments

ff86cd8

m1a2st force-pushed the KAFKA-20036 branch from 8ebef4a to ff86cd8 Compare February 11, 2026 01:32

m1a2st added 3 commits February 11, 2026 09:38

revert space

ce1433f

fix import

f97d765

Merge branch 'trunk' into KAFKA-20036

cce93a8

junrao reviewed Feb 11, 2026

View reviewed changes

Comment thread storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java Outdated

m1a2st added 6 commits February 15, 2026 15:42

addressed by comment

a17562e

test new approach

eab4c09

fix test

8608992

fix test

9244e71

revert test

e80eab0

remove unused exception

08d7f68

junrao reviewed Mar 3, 2026

View reviewed changes

Comment thread storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java Outdated

Comment thread storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java Outdated

m1a2st added 3 commits March 3, 2026 20:12

Merge branch 'trunk' into KAFKA-20036

0106e70

# Conflicts: # checkstyle/suppressions.xml

refactor the code

b72cd43

update the code

286c4d0

junrao reviewed Mar 10, 2026

View reviewed changes

Merge branch 'trunk' into KAFKA-20036

f2212c3

chia7712 reviewed Mar 22, 2026

View reviewed changes

Comment thread storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java Outdated

m1a2st added 3 commits March 23, 2026 20:08

addressed by comment

51097f2

Merge branch 'trunk' into KAFKA-20036

1a6ce93

Merge branch 'trunk' into KAFKA-20036

44b3598

# Conflicts: # storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java

junrao reviewed Mar 27, 2026

View reviewed changes

m1a2st added 2 commits March 28, 2026 09:54

Merge branch 'trunk' into KAFKA-20036

2dd6025

addressed by comment

3504536

chia7712 reviewed Mar 28, 2026

View reviewed changes

addressed by comment

e4dc30c

junrao approved these changes Mar 31, 2026

View reviewed changes

chia7712 approved these changes Mar 31, 2026

View reviewed changes

chia7712 merged commit 67d1e22 into apache:trunk Mar 31, 2026
24 checks passed

junrao reviewed Aug 11, 2026

View reviewed changes

m1a2st mentioned this pull request Aug 12, 2026

MINOR: Revert KAFKA-20036 on 4.4 #23141

Merged

chia7712 pushed a commit that referenced this pull request Aug 14, 2026

MINOR: Revert KAFKA-20036 on 4.4 (#23141)

c801be7

There is a bug, fyi: #21379 (review) Reviewers: Parker Chang <parkerhiphop027@gmail.com>, Chia-Ping Tsai <chia7712@gmail.com>

Uh oh!

Conversation

m1a2st commented Jan 31, 2026 • edited by github-actions Bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

chia7712 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

m1a2st commented Mar 22, 2026

Uh oh!

chia7712 commented Mar 22, 2026

Uh oh!

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

chia7712 commented Mar 27, 2026

Uh oh!

m1a2st commented Mar 28, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

chia7712 commented Mar 31, 2026

Uh oh!

Uh oh!

m1a2st commented Mar 31, 2026

Uh oh!

junrao commented Aug 6, 2026

Uh oh!

chia7712 commented Aug 7, 2026

Uh oh!

junrao commented Aug 7, 2026

Uh oh!

chia7712 commented Aug 8, 2026

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

m1a2st commented Jan 31, 2026 •

edited by github-actions Bot

Loading

m1a2st commented Mar 28, 2026 •

edited

Loading