chia7712 commented on code in PR #23301:
URL: https://github.com/apache/kafka/pull/23301#discussion_r3931018070
##########
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.
+ // We cannot use currentCleaned.readNextOffset() because
compaction may leave holes
+ // in the offset sequence, so the first retained batch's
base offset could be much larger.
+ long nextBaseOffset =
chunk.retained().batches().iterator().next().baseOffset();
+ currentCleaned =
UnifiedLog.createNewCleanedSegment(log.dir(), log.config(), nextBaseOffset);
+ logger.info("Created new cleaned segment with base offset
{} for partition {}", nextBaseOffset, log.topicPartition());
+
+ // The chunk must be appended before the next cleanInto()
call, which would clear the
+ // shared write buffer backing the retained records
+ currentCleaned.append(chunk.maxOffset(), chunk.retained());
+ // Aborted transactions staged while filtering the chunk
belong to the segment the
+ // chunk was appended to
+
transactionMetadata.flushPendingAbortedTxnsTo(currentCleaned.txnIndex());
Review Comment:
This brings a subtle behaviour change. Previously, the `IOException` was
wrapped in a `RuntimeException`, which was caught and converted to a
`LogCleaningException`, so the partition was marked as "uncleanable". This
patch throws the `IOException` directly, so the log dir will be marked
"offline" instead.
I think the change is in the right direction, since it's an IO error either
way, and taking the disk offline seems reasonable.
However, it would be good to mention this behaviour change in the PR
description.
--
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]
