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

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

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.

chia7712

@m1a2st thanks for this fix

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

storage/src/main/java/org/apache/kafka/storage/internals/log/SegmentOverflowException.java

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

junrao

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

junrao · 2026-02-11T19:11:53Z

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

        List<List<LogSegment>> groupedSegments = groupSegmentsBySize(
                log.logSegments(0, endOffset),
-                log.config().segmentSize(),
+                effectiveMaxSize,


Hmm, does this approach work in general? When grouping segments, we need to include at least one segment in the group. It's possible that the cleaning of a single segment can cause it to exceed 2GB.

That is a good point. We could follow the approach used for handling offset overflow: split the segment and then restart the cleanup. The trade-off is that the first hale of the segment will be cleaned in isolation, so there might be little to nothing to clean up :)

That approach could work, but one has to guess the size to split the segments into. Have you considered the alternative of creating multiple cleaned segments? log.replaceSegments() already supports replacing multiple segments. If cleanInto() hits a file overflow exception, we could close the current cleaned segment, create a new one and continue the cleaning.

yes, we can roll the new segment when either the "size check" or "overflow check" is triggered.

// 1. Size Check: current size + retained size > config limit // 2. Overflow Check: max offset - base offset > Integer.MAX_VALUE boolean willExceedSize = (long) dest.size() + retained.sizeInBytes() > log.config().segmentSize(); boolean willOverflow = result.maxOffset() - dest.baseOffset() > Integer.MAX_VALUE; if (willExceedSize || willOverflow) { logger.info("Rolling new segment. Condition met: size_exceeded={}, overflow={}. (Segment size: {}, Batch size: {}, BaseOffset: {}, MaxOffset: {})", willExceedSize, willOverflow, dest.size(), retained.sizeInBytes(), dest.baseOffset(), result.maxOffset()); dest = rollNewSegment(log, dest, cleanedSegments, transactionMetadata, retained); }

However, I have another concern regrading "temporary disk usage". If we remove the initial segment grouping entirely, it might requires a significant amount of disk space to hold all the cleaned segments simultaneously before the replacement happens.

I believe the grouping logic should be retained, but simplified to serve as a batch size threshold. This way, we can control the cleaning scope to avoid occupying too much disk space, while still allowing the inner logic to split segment dynamically if needed

I think we should retain groupSegmentsBySize() to control temporary disk usage, and handle overflow dynamically within cleanInto() by creating multiple cleaned segments as needed.

This approach allows us to avoid disk space issues while still handling segment overflow gracefully. The peak disk usage remains bounded by the group size rather than the total log size.

Yes, we will still want to group the segments.

junrao

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

junrao · 2026-03-03T00:26:28Z

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

     * @param currentTime The time at which the clean was initiated
+     * @param log The log instance for creating new segments if overflow occurs
+     *
+     * @return The current active destination segment (maybe different from input dest if overflow occurred)


This api seems awkward. An alternative is to instead passing in a starting position in sourceRecords to cleanInto(). Initially, we can pass in 0 as the position, if cleanInto() hits a size limit, it throws an exception with the current position in sourceRecords. The caller catches this exception, creates a new destination segment and call cleanInto() again with position in the exception and the new destination segment.

Thanks for the feedback! I agree this is a cleaner design. I've updated cleanInto() to accept a starting position and throw an exception containing the current position on overflow. The caller then creates a new destination segment and resumes cleaning from where it left off.

junrao · 2026-03-03T00:27:28Z

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

                MemoryRecords retained = MemoryRecords.readableRecords(outputBuffer);
+
+                // Check for TWO types of overflow BEFORE appending:
+                // 1. Offset overflow: offset range exceeds Integer.MAX_VALUE


The grouping of the segments takes offset overflow into consideration. So, it seems that we can't hit this?

# Conflicts: # checkstyle/suppressions.xml

junrao

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

junrao · 2026-03-10T23:06:52Z

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

+
+                            // Complete current cleaned segment
+                            currentCleaned.onBecomeInactiveSegment();
+                            currentCleaned.flush();


Should we set setLastModified for currentCleaned?

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

storage/src/main/java/org/apache/kafka/storage/internals/log/SegmentSizeOverflowException.java

storage/src/test/java/org/apache/kafka/storage/internals/log/CleanerIntegrationTest.java

m1a2st · 2026-03-12T01:08:06Z

Thanks for @junrao review, I have addressed all the comments

junrao

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

junrao · 2026-03-13T23:20:08Z

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

+                        // If cleanInto completes without exception, we're done with this segment
+                        cleaningComplete = true;
+
+                    } catch (SegmentSizeOverflowException e) {


Where do we catch LogSegmentOffsetOverflowException now?

Using an exception for control flow feels a bit unnatural here. Could we consider returning a result object instead to indicate where the cleaning stopped?

Let's me attach a example to strengthen my comment.

private record Overflow(int position) {}

if (sizeOverflow || offsetOverflow) { // log.xxx return Optional.of(new Overflow(position - result.bytesRead())); } ... return Optional.empty();

while (!cleaningComplete) { Optional<Overflow> overflowOpt = cleanInto( log.topicPartition(), currentSegment.log(), currentCleaned, position, ); if (overflowOpt.isPresent()) { Overflow overflow = overflowOpt.get(); logger.info("Completing cleaned segment {} due to overflow, creating new segment", currentCleaned.baseOffset()); currentCleaned.onBecomeInactiveSegment(); currentCleaned.flush(); currentCleaned.setLastModified(currentSegment.lastModified()); cleanedSegments.add(currentCleaned); Iterator<FileChannelRecordBatch> nextBatches = currentSegment.log().batchesFrom(overflow.position()).iterator(); long nextBaseOffset = nextBatches.hasNext() ? nextBatches.next().baseOffset() : currentCleaned.readNextOffset(); currentCleaned = UnifiedLog.createNewCleanedSegment(log.dir(), log.config(), nextBaseOffset); transactionMetadata.setCleanedIndex(Optional.of(currentCleaned.txnIndex())); position = overflow.position(); } else { cleaningComplete = true; } }

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

junrao · 2026-03-13T23:25:07Z

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

+                        cleanedSegments.add(currentCleaned);
+
+                        // Create new cleaned segment with base offset = next offset of completed segment
+                        long nextBaseOffset = currentCleaned.readNextOffset();


If we use the logic here to handle LogSegmentOffsetOverflowException, it's better to set nextBaseOffset as the base offset of the next batch to be cleaned. Since compaction can leave a hole, currentCleaned.readNextOffset() may not always be the base offset of the next batch to be cleaned.

junrao · 2026-03-13T23:27:57Z

storage/src/main/java/org/apache/kafka/storage/internals/log/LocalLog.java

-            // remove the index entry
-            if (segment.baseOffset() != sortedNewSegments.get(0).baseOffset()) {
+            // remove the index entry; skip removal for base offsets that a new segment is replacing in-place
+            if (!newSegmentBaseOffsets.contains(segment.baseOffset())) {


Could we merge the code in line 1051 here?

+1 to merge them

chia7712

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

chia7712 · 2026-03-21T19:13:08Z

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

+                // recompression during cleaning can cause the cleaned segment to exceed that size.
+                // Similarly, combining multiple source segments into one cleaned segment can cause
+                // the offset range to exceed Integer.MAX_VALUE.
+                boolean sizeOverflow = retained.sizeInBytes() > maxCleanedSegmentSize - dest.size();


Should we allow the write for the first batch of an empty segment?

We should allow this.

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

chia7712 · 2026-03-22T16:08:45Z

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

-     * Clean a group of segments into a single replacement segment.
+     * Clean a group of segments into one or more replacement segments.
+     *
+     * <p>If cleaning would cause the destination segment's size or offset range to exceed the configured limit


* <p>If cleaning causes the destination segment's size or offset range to exceed the configured limit * (e.g., due to recompression or combining multiple source segments), the current cleaned segment is * finalized and a new one is started.

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

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

storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java Outdated Show resolved Hide resolved

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

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

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

m1a2st added 6 commits March 11, 2026 20:52

temp fix

1462eb4

add unit test

845172f

Merge branch 'trunk' into KAFKA-20036

242be0a

combined the logic

3505373

cleanup the code

1a9077a

cleanup the code

4c83c89

junrao reviewed Mar 13, 2026

View reviewed changes

m1a2st added 3 commits March 19, 2026 20:00

fix other comments

739230b

Merge branch 'trunk' into KAFKA-20036

f60a153

addressed by comment

ee138cf

chia7712 reviewed Mar 21, 2026

View reviewed changes

m1a2st added 2 commits March 22, 2026 16:17

addressed vy comment

3b5db07

Merge branch 'trunk' into KAFKA-20036

f2212c3

chia7712 reviewed Mar 22, 2026

View reviewed changes

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

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!

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

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!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

m1a2st commented Mar 12, 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!

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!

chia7712 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!

m1a2st commented Mar 22, 2026

Uh oh!

chia7712 commented Mar 22, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

m1a2st commented Jan 31, 2026 •

edited by github-actions bot

Loading