You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "m1a2st (via GitHub)" <gi...@apache.org> on 2026/01/31 07:13:35 UTC
[PR] KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes [kafka]
m1a2st opened a new pull request, #21379:
URL: https://github.com/apache/kafka/pull/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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #21379:
URL: https://github.com/apache/kafka/pull/21379#discussion_r2805377490
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java:
##########
@@ -169,9 +176,17 @@ public Map.Entry<Long, CleanerStats> doClean(LogToClean cleanable, long currentT
log.name(), new Date(cleanableHorizonMs), new Date(legacyDeleteHorizonMs));
CleanedTransactionMetadata transactionMetadata = new CleanedTransactionMetadata();
+ double sizeRatio = segmentOverflowPartitions.getOrDefault(log.topicPartition(), 1.0);
+ if (sizeRatio != 1.0) {
+ logger.info("Partition {} has overflow history. " + "Reducing effective segment size to {}% for this round.",
+ log.topicPartition(), sizeRatio * 100);
+ }
+
+ int effectiveMaxSize = (int) (log.config().segmentSize() * sizeRatio);
+
List<List<LogSegment>> groupedSegments = groupSegmentsBySize(
log.logSegments(0, endOffset),
- log.config().segmentSize(),
+ effectiveMaxSize,
Review Comment:
Yes, we will still want to group the segments.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #21379:
URL: https://github.com/apache/kafka/pull/21379#discussion_r2937034411
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/LocalLog.java:
##########
@@ -1034,8 +1034,8 @@ public static List<LogSegment> replaceSegments(LogSegments existingSegments,
// delete the old files
List<LogSegment> deletedNotReplaced = new ArrayList<>();
for (LogSegment segment : sortedOldSegments) {
- // 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())) {
Review Comment:
+1 to merge them
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java:
##########
@@ -253,46 +277,74 @@ public void cleanSegments(UnifiedLog log,
upperBoundOffsetOfCleaningRound,
stats,
currentTime
- );
- } catch (LogSegmentOffsetOverflowException e) {
- // Split the current segment. It's also safest to abort the current cleaning process, so that we retry from
- // scratch once the split is complete.
- logger.info("Caught segment overflow error during cleaning: {}", e.getMessage());
- log.splitOverflowedSegment(currentSegment);
- throw new LogCleaningAbortedException();
+ );
+
+ // If cleanInto completes without exception, we're done with this segment
+ cleaningComplete = true;
+
+ } catch (SegmentSizeOverflowException e) {
Review Comment:
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?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #21379:
URL: https://github.com/apache/kafka/pull/21379#discussion_r2800694490
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java:
##########
@@ -169,9 +176,17 @@ public Map.Entry<Long, CleanerStats> doClean(LogToClean cleanable, long currentT
log.name(), new Date(cleanableHorizonMs), new Date(legacyDeleteHorizonMs));
CleanedTransactionMetadata transactionMetadata = new CleanedTransactionMetadata();
+ double sizeRatio = segmentOverflowPartitions.getOrDefault(log.topicPartition(), 1.0);
+ if (sizeRatio != 1.0) {
+ logger.info("Partition {} has overflow history. " + "Reducing effective segment size to {}% for this round.",
+ log.topicPartition(), sizeRatio * 100);
+ }
+
+ int effectiveMaxSize = (int) (log.config().segmentSize() * sizeRatio);
+
List<List<LogSegment>> groupedSegments = groupSegmentsBySize(
log.logSegments(0, endOffset),
- log.config().segmentSize(),
+ effectiveMaxSize,
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #21379:
URL: https://github.com/apache/kafka/pull/21379#discussion_r2795004624
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java:
##########
@@ -169,9 +176,17 @@ public Map.Entry<Long, CleanerStats> doClean(LogToClean cleanable, long currentT
log.name(), new Date(cleanableHorizonMs), new Date(legacyDeleteHorizonMs));
CleanedTransactionMetadata transactionMetadata = new CleanedTransactionMetadata();
+ double sizeRatio = segmentOverflowPartitions.getOrDefault(log.topicPartition(), 1.0);
+ if (sizeRatio != 1.0) {
+ logger.info("Partition {} has overflow history. " + "Reducing effective segment size to {}% for this round.",
+ log.topicPartition(), sizeRatio * 100);
+ }
+
+ int effectiveMaxSize = (int) (log.config().segmentSize() * sizeRatio);
+
List<List<LogSegment>> groupedSegments = groupSegmentsBySize(
log.logSegments(0, endOffset),
- log.config().segmentSize(),
+ effectiveMaxSize,
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on PR #21379:
URL: https://github.com/apache/kafka/pull/21379#issuecomment-4147200424
> 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 these tests.
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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #21379:
URL: https://github.com/apache/kafka/pull/21379#discussion_r2785102395
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java:
##########
@@ -60,6 +61,11 @@ public class Cleaner {
private final Time time;
private final Consumer<TopicPartition> checkDone;
+ /**
+ * The topic partitions that have segment overflow history mapped to their segment size ratio
+ */
+ private final Map<TopicPartition, Double> segmentOverflowPartitions = new HashMap<>();
Review Comment:
If we cache overflowed partitions at the `Cleaner` level, they won't be visible to other cleaners. This means another cleaner might hit the same error. I think it's better to maintain this state in `LogCleaner`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #21379:
URL: https://github.com/apache/kafka/pull/21379#discussion_r2961426218
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java:
##########
@@ -253,46 +277,74 @@ public void cleanSegments(UnifiedLog log,
upperBoundOffsetOfCleaningRound,
stats,
currentTime
- );
- } catch (LogSegmentOffsetOverflowException e) {
- // Split the current segment. It's also safest to abort the current cleaning process, so that we retry from
- // scratch once the split is complete.
- logger.info("Caught segment overflow error during cleaning: {}", e.getMessage());
- log.splitOverflowedSegment(currentSegment);
- throw new LogCleaningAbortedException();
+ );
+
+ // If cleanInto completes without exception, we're done with this segment
+ cleaningComplete = true;
+
+ } catch (SegmentSizeOverflowException e) {
Review Comment:
Let's me attach a example to strengthen my comment.
```java
private record Overflow(int position) {}
```
```java
if (sizeOverflow || offsetOverflow) {
// log.xxx
return Optional.of(new Overflow(position - result.bytesRead()));
}
...
return Optional.empty();
```
```java
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;
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #21379:
URL: https://github.com/apache/kafka/pull/21379#discussion_r2879135421
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java:
##########
@@ -302,19 +327,26 @@ public void cleanSegments(UnifiedLog log,
* @param upperBoundOffsetOfCleaningRound Next offset of the last batch in the source segment
* @param stats Collector for cleaning statistics
* @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)
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #21379:
URL: https://github.com/apache/kafka/pull/21379#discussion_r3761497110
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java:
##########
@@ -401,6 +460,19 @@ public boolean shouldRetainRecord(RecordBatch batch, Record record) {
if (outputBuffer.position() > 0) {
outputBuffer.flip();
MemoryRecords retained = MemoryRecords.readableRecords(outputBuffer);
+
+ // While groupSegmentsBySize() ensures source segments don't exceed Integer.MAX_VALUE,
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #21379:
URL: https://github.com/apache/kafka/pull/21379#discussion_r2934194389
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/LocalLog.java:
##########
@@ -1034,8 +1034,8 @@ public static List<LogSegment> replaceSegments(LogSegments existingSegments,
// delete the old files
List<LogSegment> deletedNotReplaced = new ArrayList<>();
for (LogSegment segment : sortedOldSegments) {
- // 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())) {
Review Comment:
Could we merge the code in line 1051 here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #21379:
URL: https://github.com/apache/kafka/pull/21379#discussion_r2914995491
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java:
##########
@@ -235,25 +235,56 @@ public void cleanSegments(UnifiedLog log,
logger.info(
"Cleaning {} in log {} into {} with an upper bound deletion horizon {} computed from " +
"the segment last modified time of {},{} deletes.",
- currentSegment, log.name(), cleaned.baseOffset(), legacyDeleteHorizonMs, currentSegment.lastModified(),
+ currentSegment, log.name(), currentCleaned.baseOffset(), legacyDeleteHorizonMs, currentSegment.lastModified(),
retainLegacyDeletesAndTxnMarkers ? "retaining" : "discarding"
);
try {
- cleanInto(
- log.topicPartition(),
- currentSegment.log(),
- cleaned,
- map,
- retainLegacyDeletesAndTxnMarkers,
- log.config().deleteRetentionMs,
- log.config().maxMessageSize(),
- transactionMetadata,
- lastOffsetOfActiveProducers,
- upperBoundOffsetOfCleaningRound,
- stats,
- currentTime
- );
+ // Start cleaning from position 0
+ int position = 0;
+ boolean cleaningComplete = false;
+
+ while (!cleaningComplete) {
+ try {
+ cleanInto(
+ log.topicPartition(),
+ currentSegment.log(),
+ currentCleaned,
+ position,
+ map,
+ retainLegacyDeletesAndTxnMarkers,
+ log.config().deleteRetentionMs,
+ log.config().maxMessageSize(),
+ transactionMetadata,
+ lastOffsetOfActiveProducers,
+ upperBoundOffsetOfCleaningRound,
+ stats,
+ currentTime
+ );
+
+ // If cleanInto completes without exception, we're done with this segment
+ cleaningComplete = true;
+
+ } catch (SegmentSizeOverflowException e) {
+ // Overflow detected - complete current segment and create new one
Review Comment:
Could we handling LogSegmentOffsetOverflowException in the same way here? This will simplify the logic and will be more efficient.
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java:
##########
@@ -235,25 +235,56 @@ public void cleanSegments(UnifiedLog log,
logger.info(
"Cleaning {} in log {} into {} with an upper bound deletion horizon {} computed from " +
"the segment last modified time of {},{} deletes.",
- currentSegment, log.name(), cleaned.baseOffset(), legacyDeleteHorizonMs, currentSegment.lastModified(),
+ currentSegment, log.name(), currentCleaned.baseOffset(), legacyDeleteHorizonMs, currentSegment.lastModified(),
retainLegacyDeletesAndTxnMarkers ? "retaining" : "discarding"
);
try {
- cleanInto(
- log.topicPartition(),
- currentSegment.log(),
- cleaned,
- map,
- retainLegacyDeletesAndTxnMarkers,
- log.config().deleteRetentionMs,
- log.config().maxMessageSize(),
- transactionMetadata,
- lastOffsetOfActiveProducers,
- upperBoundOffsetOfCleaningRound,
- stats,
- currentTime
- );
+ // Start cleaning from position 0
+ int position = 0;
+ boolean cleaningComplete = false;
+
+ while (!cleaningComplete) {
+ try {
+ cleanInto(
+ log.topicPartition(),
+ currentSegment.log(),
+ currentCleaned,
+ position,
+ map,
+ retainLegacyDeletesAndTxnMarkers,
+ log.config().deleteRetentionMs,
+ log.config().maxMessageSize(),
+ transactionMetadata,
+ lastOffsetOfActiveProducers,
+ upperBoundOffsetOfCleaningRound,
+ stats,
+ currentTime
+ );
+
+ // If cleanInto completes without exception, we're done with this segment
+ cleaningComplete = true;
+
+ } catch (SegmentSizeOverflowException e) {
+ // Overflow detected - complete current segment and create new one
+ logger.info("Completing cleaned segment {} due to overflow, creating new segment", currentCleaned.baseOffset());
+
+ // Complete current cleaned segment
+ currentCleaned.onBecomeInactiveSegment();
+ currentCleaned.flush();
Review Comment:
Should we set setLastModified for currentCleaned?
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java:
##########
@@ -400,6 +445,21 @@ public boolean shouldRetainRecord(RecordBatch batch, Record record) {
if (outputBuffer.position() > 0) {
outputBuffer.flip();
MemoryRecords retained = MemoryRecords.readableRecords(outputBuffer);
+
+ // Check for size overflow: file size would exceed Integer.MAX_VALUE. While
+ // groupSegmentsBySize() ensures source segments don't exceed maxSize, the cleaned
+ // segment contains retained data from multiple sources, and low compaction ratios
+ // can result in cleaned segment sizes approaching or exceeding Integer.MAX_VALUE.
Review Comment:
How about the following?
```
// Make sure that file size won't exceed Integer.MAX_VALUE. While groupSegmentsBySize()
// ensures source segments don't exceed Integer.MAX_VALUE, recompression during cleaning
// can result in the size of cleaned segments exceeding Integer.MAX_VALUE.
```
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/SegmentSizeOverflowException.java:
##########
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.storage.internals.log;
+
+import org.apache.kafka.common.KafkaException;
+
+/**
+ * Exception thrown when cleaning a segment would cause size overflow.
+ */
+public class SegmentSizeOverflowException extends KafkaException {
+
Review Comment:
extra new line
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/SegmentSizeOverflowException.java:
##########
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.storage.internals.log;
+
+import org.apache.kafka.common.KafkaException;
+
+/**
+ * Exception thrown when cleaning a segment would cause size overflow.
+ */
+public class SegmentSizeOverflowException extends KafkaException {
+
+ private final int position;
Review Comment:
Could we add a comment what this field represents?
##########
storage/src/test/java/org/apache/kafka/storage/internals/log/CleanerIntegrationTest.java:
##########
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.storage.internals.log;
+
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.test.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.common.test.api.Type;
+import org.apache.kafka.server.metrics.KafkaYammerMetrics;
+
+import com.yammer.metrics.core.Gauge;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.concurrent.CompletableFuture;
+import java.util.stream.IntStream;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class CleanerIntegrationTest {
+
+ @SuppressWarnings("unchecked")
+ @ClusterTest(types = Type.CO_KRAFT)
+ public void testCleanerSegmentCompactionOverflow(ClusterInstance cluster) throws Exception {
Review Comment:
Hmm, this test passes without the actual fix. Also, integration test is expensive and it's not ideal to generate 2GB worth of data. Could we design a simpler and more effective unit test?
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/SegmentSizeOverflowException.java:
##########
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.storage.internals.log;
+
+import org.apache.kafka.common.KafkaException;
+
+/**
+ * Exception thrown when cleaning a segment would cause size overflow.
+ */
+public class SegmentSizeOverflowException extends KafkaException {
+
+ private final int position;
+ private final LogSegment segment;
+
+ public SegmentSizeOverflowException(LogSegment segment, int position) {
+ super(String.format("Segment %s size would overflow at position %d", segment, position));
+ this.segment = segment;
+ this.position = position;
+ }
+
+ public int position() {
+ return position;
+ }
+
+ public LogSegment segment() {
Review Comment:
This is unused?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #21379:
URL: https://github.com/apache/kafka/pull/21379#issuecomment-5210297893
@junrao sorry for the misleading message. We could revert the original patch and then commit a new one with an updated message. WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on PR #21379:
URL: https://github.com/apache/kafka/pull/21379#issuecomment-5209877720
@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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on PR #21379:
URL: https://github.com/apache/kafka/pull/21379#issuecomment-4106187756
> 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #21379:
URL: https://github.com/apache/kafka/pull/21379#issuecomment-4145739681
@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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #21379:
URL: https://github.com/apache/kafka/pull/21379#discussion_r2803957651
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java:
##########
@@ -169,9 +176,17 @@ public Map.Entry<Long, CleanerStats> doClean(LogToClean cleanable, long currentT
log.name(), new Date(cleanableHorizonMs), new Date(legacyDeleteHorizonMs));
CleanedTransactionMetadata transactionMetadata = new CleanedTransactionMetadata();
+ double sizeRatio = segmentOverflowPartitions.getOrDefault(log.topicPartition(), 1.0);
+ if (sizeRatio != 1.0) {
+ logger.info("Partition {} has overflow history. " + "Reducing effective segment size to {}% for this round.",
+ log.topicPartition(), sizeRatio * 100);
+ }
+
+ int effectiveMaxSize = (int) (log.config().segmentSize() * sizeRatio);
+
List<List<LogSegment>> groupedSegments = groupSegmentsBySize(
log.logSegments(0, endOffset),
- log.config().segmentSize(),
+ effectiveMaxSize,
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #21379:
URL: https://github.com/apache/kafka/pull/21379#discussion_r3853991144
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java:
##########
@@ -401,6 +460,19 @@ public boolean shouldRetainRecord(RecordBatch batch, Record record) {
if (outputBuffer.position() > 0) {
outputBuffer.flip();
MemoryRecords retained = MemoryRecords.readableRecords(outputBuffer);
+
+ // While groupSegmentsBySize() ensures source segments don't exceed Integer.MAX_VALUE,
Review Comment:
Thanks for pointing this out. I'll investigate this issue further.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #21379:
URL: https://github.com/apache/kafka/pull/21379#discussion_r2934175647
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java:
##########
@@ -253,46 +277,74 @@ public void cleanSegments(UnifiedLog log,
upperBoundOffsetOfCleaningRound,
stats,
currentTime
- );
- } catch (LogSegmentOffsetOverflowException e) {
- // Split the current segment. It's also safest to abort the current cleaning process, so that we retry from
- // scratch once the split is complete.
- logger.info("Caught segment overflow error during cleaning: {}", e.getMessage());
- log.splitOverflowedSegment(currentSegment);
- throw new LogCleaningAbortedException();
+ );
+
+ // If cleanInto completes without exception, we're done with this segment
+ cleaningComplete = true;
+
+ } catch (SegmentSizeOverflowException e) {
Review Comment:
Where do we catch LogSegmentOffsetOverflowException now?
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java:
##########
@@ -253,46 +277,74 @@ public void cleanSegments(UnifiedLog log,
upperBoundOffsetOfCleaningRound,
stats,
currentTime
- );
- } catch (LogSegmentOffsetOverflowException e) {
- // Split the current segment. It's also safest to abort the current cleaning process, so that we retry from
- // scratch once the split is complete.
- logger.info("Caught segment overflow error during cleaning: {}", e.getMessage());
- log.splitOverflowedSegment(currentSegment);
- throw new LogCleaningAbortedException();
+ );
+
+ // If cleanInto completes without exception, we're done with this segment
+ cleaningComplete = true;
+
+ } catch (SegmentSizeOverflowException e) {
+ // Overflow detected - complete current segment and create new one
+ logger.info("Completing cleaned segment {} due to overflow, creating new segment", currentCleaned.baseOffset());
+
+ // Complete current cleaned segment
+ currentCleaned.onBecomeInactiveSegment();
+ currentCleaned.flush();
+ currentCleaned.setLastModified(segments.get(segments.size() - 1).lastModified());
+ cleanedSegments.add(currentCleaned);
+
+ // Create new cleaned segment with base offset = next offset of completed segment
+ long nextBaseOffset = currentCleaned.readNextOffset();
Review Comment:
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.
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java:
##########
@@ -253,46 +277,74 @@ public void cleanSegments(UnifiedLog log,
upperBoundOffsetOfCleaningRound,
stats,
currentTime
- );
- } catch (LogSegmentOffsetOverflowException e) {
- // Split the current segment. It's also safest to abort the current cleaning process, so that we retry from
- // scratch once the split is complete.
- logger.info("Caught segment overflow error during cleaning: {}", e.getMessage());
- log.splitOverflowedSegment(currentSegment);
- throw new LogCleaningAbortedException();
+ );
+
+ // If cleanInto completes without exception, we're done with this segment
+ cleaningComplete = true;
+
+ } catch (SegmentSizeOverflowException e) {
+ // Overflow detected - complete current segment and create new one
+ logger.info("Completing cleaned segment {} due to overflow, creating new segment", currentCleaned.baseOffset());
+
+ // Complete current cleaned segment
+ currentCleaned.onBecomeInactiveSegment();
+ currentCleaned.flush();
+ currentCleaned.setLastModified(segments.get(segments.size() - 1).lastModified());
Review Comment:
We want to use the timestamp of the last segment replacing currentCleaned.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #21379:
URL: https://github.com/apache/kafka/pull/21379#discussion_r2797080701
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java:
##########
@@ -169,9 +176,17 @@ public Map.Entry<Long, CleanerStats> doClean(LogToClean cleanable, long currentT
log.name(), new Date(cleanableHorizonMs), new Date(legacyDeleteHorizonMs));
CleanedTransactionMetadata transactionMetadata = new CleanedTransactionMetadata();
+ double sizeRatio = segmentOverflowPartitions.getOrDefault(log.topicPartition(), 1.0);
+ if (sizeRatio != 1.0) {
+ logger.info("Partition {} has overflow history. " + "Reducing effective segment size to {}% for this round.",
+ log.topicPartition(), sizeRatio * 100);
+ }
+
+ int effectiveMaxSize = (int) (log.config().segmentSize() * sizeRatio);
+
List<List<LogSegment>> groupedSegments = groupSegmentsBySize(
log.logSegments(0, endOffset),
- log.config().segmentSize(),
+ effectiveMaxSize,
Review Comment:
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 :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #21379:
URL: https://github.com/apache/kafka/pull/21379#issuecomment-4106721505
```
-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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #21379:
URL: https://github.com/apache/kafka/pull/21379#discussion_r2971711448
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java:
##########
@@ -189,7 +208,11 @@ public Map.Entry<Long, CleanerStats> doClean(LogToClean cleanable, long currentT
}
/**
- * 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
Review Comment:
```
* <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.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #21379:
URL: https://github.com/apache/kafka/pull/21379#discussion_r3766839820
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java:
##########
@@ -401,6 +460,19 @@ public boolean shouldRetainRecord(RecordBatch batch, Record record) {
if (outputBuffer.position() > 0) {
outputBuffer.flip();
MemoryRecords retained = MemoryRecords.readableRecords(outputBuffer);
+
+ // While groupSegmentsBySize() ensures source segments don't exceed Integer.MAX_VALUE,
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #21379:
URL: https://github.com/apache/kafka/pull/21379#discussion_r2875399185
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java:
##########
@@ -302,19 +327,26 @@ public void cleanSegments(UnifiedLog log,
* @param upperBoundOffsetOfCleaningRound Next offset of the last batch in the source segment
* @param stats Collector for cleaning statistics
* @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)
Review Comment:
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.
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java:
##########
@@ -400,6 +432,36 @@ public boolean shouldRetainRecord(RecordBatch batch, Record record) {
if (outputBuffer.position() > 0) {
outputBuffer.flip();
MemoryRecords retained = MemoryRecords.readableRecords(outputBuffer);
+
+ // Check for TWO types of overflow BEFORE appending:
+ // 1. Offset overflow: offset range exceeds Integer.MAX_VALUE
Review Comment:
The grouping of the segments takes offset overflow into consideration. So, it seems that we can't hit this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #21379:
URL: https://github.com/apache/kafka/pull/21379
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #21379:
URL: https://github.com/apache/kafka/pull/21379#issuecomment-4159005088
> 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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #21379:
URL: https://github.com/apache/kafka/pull/21379#issuecomment-5224952121
> It's probably not worth it. We can just update the description in the PR.
Copy that. We will update it after CoC Asia.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on PR #21379:
URL: https://github.com/apache/kafka/pull/21379#issuecomment-5219220799
@chia7712 It's probably not worth it. We can just update the description in the PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #21379:
URL: https://github.com/apache/kafka/pull/21379#discussion_r3003390788
##########
core/src/test/scala/unit/kafka/log/LogCleanerTest.scala:
##########
@@ -1634,47 +1634,71 @@ class LogCleanerTest extends Logging {
}
@Test
- def testSegmentWithOffsetOverflow(): Unit = {
- val cleaner = makeCleaner(Int.MaxValue)
+ def testCleanedSegmentSizeOverflow(): Unit = {
+ // Put one record per source segment so each filterTo() call reads exactly one batch.
+ // After cleaning source segment 0 into currentCleaned, cleaning source segment 1 would push
+ // currentCleaned over maxCleanedSegmentSize, triggering overflow and
+ // rolling to a second cleaned segment.
val logProps = new Properties()
- logProps.put(TopicConfig.INDEX_INTERVAL_BYTES_CONFIG, 1: java.lang.Integer)
- logProps.put(TopicConfig.FILE_DELETE_DELAY_MS_CONFIG, 1000: java.lang.Integer)
- val config = LogConfig.fromProps(logConfig.originals, logProps)
+ logProps.put(LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG, 4096: java.lang.Integer)
+ val log = makeLog(config = LogConfig.fromProps(logConfig.originals, logProps))
+
+ log.appendAsLeader(record(0, 0), 0)
+ log.roll()
+ log.appendAsLeader(record(1, 1), 0)
+ log.roll()
- 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.
Review Comment:
This line seems redundant given the comment in next line.
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java:
##########
@@ -255,45 +283,74 @@ public void cleanSegments(UnifiedLog log,
stats,
currentTime
);
- } catch (LogSegmentOffsetOverflowException e) {
- // Split the current segment. It's also safest to abort the current cleaning process, so that we retry from
- // scratch once the split is complete.
- logger.info("Caught segment overflow error during cleaning: {}", e.getMessage());
- log.splitOverflowedSegment(currentSegment);
- throw new LogCleaningAbortedException();
+
+ if (overflowOpt.isPresent()) {
+ // Overflow detected - complete current segment and create new one
+ logger.info("Completing cleaned segment {} due to overflow, creating new segment", currentCleaned.baseOffset());
+
+ currentCleaned.onBecomeInactiveSegment();
+ currentCleaned.flush();
+ currentCleaned.setLastModified(currentSegment.lastModified());
+ cleanedSegments.add(currentCleaned);
+
+ // Use the base offset of the next batch to be cleaned as the new segment's base offset.
+ // We cannot use currentCleaned.readNextOffset() because compaction may leave holes
+ // in the offset sequence, so the next batch's base offset could be much larger.
+ int overflowPosition = overflowOpt.get();
+ Iterator<FileChannelRecordBatch> nextBatches = currentSegment.log().batchesFrom(overflowPosition).iterator();
+ long nextBaseOffset = nextBatches.hasNext() ? nextBatches.next().baseOffset() : currentCleaned.readNextOffset();
+ currentCleaned = UnifiedLog.createNewCleanedSegment(log.dir(), log.config(), nextBaseOffset);
+ transactionMetadata.setCleanedIndex(Optional.of(currentCleaned.txnIndex()));
+
+ logger.info("Created new cleaned segment with base offset {} for partition {}", nextBaseOffset, log.topicPartition());
+ position = overflowPosition;
+ } else {
+ break;
+ }
}
+
currentSegmentOpt = nextSegmentOpt;
}
- cleaned.onBecomeInactiveSegment();
- // flush new segment to disk before swap
- cleaned.flush();
+ // Process the final segment
+ currentCleaned.onBecomeInactiveSegment();
+ currentCleaned.flush();
// update the modification date to retain the last modified date of the original files
long modified = segments.get(segments.size() - 1).lastModified();
- cleaned.setLastModified(modified);
+ currentCleaned.setLastModified(modified);
+
+ 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);
Review Comment:
Could we log the cleaned segments, instead of just the size?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #21379:
URL: https://github.com/apache/kafka/pull/21379#discussion_r3004842593
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java:
##########
@@ -255,45 +283,74 @@ public void cleanSegments(UnifiedLog log,
stats,
currentTime
);
- } catch (LogSegmentOffsetOverflowException e) {
- // Split the current segment. It's also safest to abort the current cleaning process, so that we retry from
- // scratch once the split is complete.
- logger.info("Caught segment overflow error during cleaning: {}", e.getMessage());
- log.splitOverflowedSegment(currentSegment);
- throw new LogCleaningAbortedException();
+
+ if (overflowOpt.isPresent()) {
+ // Overflow detected - complete current segment and create new one
+ logger.info("Completing cleaned segment {} due to overflow, creating new segment", currentCleaned.baseOffset());
+
+ currentCleaned.onBecomeInactiveSegment();
+ currentCleaned.flush();
+ currentCleaned.setLastModified(currentSegment.lastModified());
+ cleanedSegments.add(currentCleaned);
+
+ // Use the base offset of the next batch to be cleaned as the new segment's base offset.
+ // We cannot use currentCleaned.readNextOffset() because compaction may leave holes
+ // in the offset sequence, so the next batch's base offset could be much larger.
+ int overflowPosition = overflowOpt.get();
+ Iterator<FileChannelRecordBatch> nextBatches = currentSegment.log().batchesFrom(overflowPosition).iterator();
+ long nextBaseOffset = nextBatches.hasNext() ? nextBatches.next().baseOffset() : currentCleaned.readNextOffset();
+ currentCleaned = UnifiedLog.createNewCleanedSegment(log.dir(), log.config(), nextBaseOffset);
+ transactionMetadata.setCleanedIndex(Optional.of(currentCleaned.txnIndex()));
+
+ logger.info("Created new cleaned segment with base offset {} for partition {}", nextBaseOffset, log.topicPartition());
+ position = overflowPosition;
+ } else {
+ break;
+ }
}
+
currentSegmentOpt = nextSegmentOpt;
}
- cleaned.onBecomeInactiveSegment();
- // flush new segment to disk before swap
- cleaned.flush();
+ // Process the final segment
+ currentCleaned.onBecomeInactiveSegment();
+ currentCleaned.flush();
// update the modification date to retain the last modified date of the original files
long modified = segments.get(segments.size() - 1).lastModified();
- cleaned.setLastModified(modified);
+ currentCleaned.setLastModified(modified);
+
+ 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, segments, log);
+ log.replaceSegments(cleanedSegments, segments);
- // swap in new segment
- 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 -> {
Review Comment:
Could you eliminate the duplicate code?
```java
Stream.concat(cleanedSegments.stream(), Stream.of(currentCleaned))
.distinct()
.forEach(segment -> {
try {
segment.deleteIfExists();
} catch (Exception deleteException) {
e.addSuppressed(deleteException);
}
});
throw e;
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #21379:
URL: https://github.com/apache/kafka/pull/21379#discussion_r2787629725
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java:
##########
@@ -60,6 +61,11 @@ public class Cleaner {
private final Time time;
private final Consumer<TopicPartition> checkDone;
+ /**
+ * The topic partitions that have segment overflow history mapped to their segment size ratio
+ */
+ private final Map<TopicPartition, Double> segmentOverflowPartitions = new HashMap<>();
Review Comment:
Yes, you’re right. I should move the map into `LogCleaner`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #21379:
URL: https://github.com/apache/kafka/pull/21379#discussion_r3768388018
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java:
##########
@@ -401,6 +460,19 @@ public boolean shouldRetainRecord(RecordBatch batch, Record record) {
if (outputBuffer.position() > 0) {
outputBuffer.flip();
MemoryRecords retained = MemoryRecords.readableRecords(outputBuffer);
+
+ // While groupSegmentsBySize() ensures source segments don't exceed Integer.MAX_VALUE,
Review Comment:
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.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #21379:
URL: https://github.com/apache/kafka/pull/21379#discussion_r3766842767
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java:
##########
@@ -401,6 +460,19 @@ public boolean shouldRetainRecord(RecordBatch batch, Record record) {
if (outputBuffer.position() > 0) {
outputBuffer.flip();
MemoryRecords retained = MemoryRecords.readableRecords(outputBuffer);
+
+ // While groupSegmentsBySize() ensures source segments don't exceed Integer.MAX_VALUE,
Review Comment:
@m1a2st would you mind opening a PR for 4.4? it has some conflicts.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #21379:
URL: https://github.com/apache/kafka/pull/21379#discussion_r3767359026
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java:
##########
@@ -401,6 +460,19 @@ public boolean shouldRetainRecord(RecordBatch batch, Record record) {
if (outputBuffer.position() > 0) {
outputBuffer.flip();
MemoryRecords retained = MemoryRecords.readableRecords(outputBuffer);
+
+ // While groupSegmentsBySize() ensures source segments don't exceed Integer.MAX_VALUE,
Review Comment:
https://github.com/apache/kafka/pull/23141
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on PR #21379:
URL: https://github.com/apache/kafka/pull/21379#issuecomment-4043214741
Thanks for @junrao review, I have addressed all the comments
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #21379:
URL: https://github.com/apache/kafka/pull/21379#discussion_r2970014295
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java:
##########
@@ -400,6 +460,17 @@ public boolean shouldRetainRecord(RecordBatch batch, Record record) {
if (outputBuffer.position() > 0) {
outputBuffer.flip();
MemoryRecords retained = MemoryRecords.readableRecords(outputBuffer);
+
+ // While groupSegmentsBySize() ensures source segments don't exceed Integer.MAX_VALUE,
+ // 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();
Review Comment:
Should we allow the write for the first batch of an empty segment?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #21379:
URL: https://github.com/apache/kafka/pull/21379#discussion_r2971473466
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java:
##########
@@ -400,6 +460,17 @@ public boolean shouldRetainRecord(RecordBatch batch, Record record) {
if (outputBuffer.position() > 0) {
outputBuffer.flip();
MemoryRecords retained = MemoryRecords.readableRecords(outputBuffer);
+
+ // While groupSegmentsBySize() ensures source segments don't exceed Integer.MAX_VALUE,
+ // 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();
Review Comment:
We should allow this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #21379:
URL: https://github.com/apache/kafka/pull/21379#discussion_r2801941737
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java:
##########
@@ -169,9 +176,17 @@ public Map.Entry<Long, CleanerStats> doClean(LogToClean cleanable, long currentT
log.name(), new Date(cleanableHorizonMs), new Date(legacyDeleteHorizonMs));
CleanedTransactionMetadata transactionMetadata = new CleanedTransactionMetadata();
+ double sizeRatio = segmentOverflowPartitions.getOrDefault(log.topicPartition(), 1.0);
+ if (sizeRatio != 1.0) {
+ logger.info("Partition {} has overflow history. " + "Reducing effective segment size to {}% for this round.",
+ log.topicPartition(), sizeRatio * 100);
+ }
+
+ int effectiveMaxSize = (int) (log.config().segmentSize() * sizeRatio);
+
List<List<LogSegment>> groupedSegments = groupSegmentsBySize(
log.logSegments(0, endOffset),
- log.config().segmentSize(),
+ effectiveMaxSize,
Review Comment:
yes, we can roll the new segment when either the "size check" or "overflow check" is triggered.
```java
// 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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #21379:
URL: https://github.com/apache/kafka/pull/21379#discussion_r2749907193
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/SegmentOverflowException.java:
##########
@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.storage.internals.log;
+
+import org.apache.kafka.common.KafkaException;
+
+/**
+ * Exception thrown when segment size would overflow during compaction
+ */
+public class SegmentOverflowException extends KafkaException {
+ public final LogSegment segment;
+
+ public SegmentOverflowException(LogSegment segment) {
+ super("Segment size would overflow during compaction for segment " + segment);
+ this.segment = segment;
Review Comment:
why we need it?
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java:
##########
@@ -400,10 +436,15 @@ public boolean shouldRetainRecord(RecordBatch batch, Record record) {
if (outputBuffer.position() > 0) {
outputBuffer.flip();
MemoryRecords retained = MemoryRecords.readableRecords(outputBuffer);
- // it's OK not to hold the Log's lock in this case, because this segment is only accessed by other threads
- // after `Log.replaceSegments` (which acquires the lock) is called
- dest.append(result.maxOffset(), retained);
- throttler.maybeThrottle(outputBuffer.limit());
+ try {
+ // it's OK not to hold the Log's lock in this case, because this segment is only accessed by other threads
+ // after `Log.replaceSegments` (which acquires the lock) is called
+ dest.append(result.maxOffset(), retained);
Review Comment:
Could you wrap only `dest.append` in the try-catch block to avoid catching unrelated error?
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java:
##########
@@ -180,6 +195,13 @@ public Map.Entry<Long, CleanerStats> doClean(LogToClean cleanable, long currentT
cleanSegments(log, group, offsetMap, currentTime, stats, transactionMetadata, legacyDeleteHorizonMs, upperBoundOffset);
}
+ if (segmentOverflowPartitions.containsKey(log.topicPartition())) {
Review Comment:
```java
if (segmentOverflowPartitions.remove(log.topicPartition()) != null) {
logger.info("Successfully cleaned log {} with degraded size (ratio: {}%). " +
"Cleared overflow marker. Next cleaning will use normal size.",
log.name(), sizeRatio * 100);
}
```
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java:
##########
@@ -254,6 +276,20 @@ public void cleanSegments(UnifiedLog log,
stats,
currentTime
);
+ } catch (SegmentOverflowException e) {
+ if (segmentOverflowPartitions.containsKey(log.topicPartition())) {
Review Comment:
```java
var previousRatio = segmentOverflowPartitions.put(log.topicPartition(),
segmentOverflowPartitions.getOrDefault(log.topicPartition(), 1.0) * 0.9);
if (previousRatio == null) {
logger.warn("Segment overflow detected for partition {}: {}. " +
"Marked for degradation to 90% size in next cleaning round.",
log.topicPartition(), e.getMessage());
} else {
logger.warn("Repeated segment overflow for partition {}: {}. " +
"Further degrading to {}% size in next cleaning round.",
log.topicPartition(), e.getMessage(), previousRatio * 0.9 * 100);
}
```
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java:
##########
@@ -169,9 +175,18 @@ public Map.Entry<Long, CleanerStats> doClean(LogToClean cleanable, long currentT
log.name(), new Date(cleanableHorizonMs), new Date(legacyDeleteHorizonMs));
CleanedTransactionMetadata transactionMetadata = new CleanedTransactionMetadata();
+ double sizeRatio = 1.0;
Review Comment:
Would something like this work better?
```java
double sizeRatio = segmentOverflowPartitions.getOrDefault(log.topicPartition(), 1.0);
if (sizeRatio != 1.0) {
logger.info("Partition {} has overflow history. " + "Reducing effective segment size to {}% for this round.",
log.topicPartition(), sizeRatio * 100);
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20036 Handle LogCleaner segment overflow caused by compression level changes [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on PR #21379:
URL: https://github.com/apache/kafka/pull/21379#issuecomment-4161994889
Thanks for @junrao and @chia7712 review!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org