chia7712 commented on code in PR #23301:
URL: https://github.com/apache/kafka/pull/23301#discussion_r3931102000
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java:
##########
@@ -285,29 +283,33 @@ public void cleanSegments(UnifiedLog log,
currentTime
);
- 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 {
+ if (overflowOpt.isEmpty())
break;
- }
+
+ // Overflow detected - complete the current segment and
continue with the filtered chunk in a new cleaned segment
+ OverflowedChunk chunk = 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);
+
+ // Use the base offset of the first retained batch as the
new segment's base offset.
Review Comment:
Could you add a unit test for this behaviour? Previously we used the
unfiltered chunk's first batch offset as the new segment's base offset, and
this PR changes it to the first retained batch's offset.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
