junrao commented on code in PR #23301:
URL: https://github.com/apache/kafka/pull/23301#discussion_r3899195938
##########
storage/src/test/java/org/apache/kafka/storage/internals/log/LogCleanerTest.java:
##########
@@ -1886,6 +1886,59 @@ public void testCleanedSegmentOffsetOverflow() throws
IOException {
log.close();
}
+ @Test
+ public void testCleanedSegmentSizeOverflowWithAbortedTransaction() throws
IOException {
+ // The overflowing chunk contains aborted batches and their abort
marker. Since
+ // CleanedTransactionMetadata state is consumed during filtering, the
chunk must
+ // not be filtered twice; otherwise the aborted data could be retained
as
+ // committed, and its transaction index entry could be written to the
segment
+ // preceding the marker.
+ Properties logProps = new Properties();
+ logProps.put(LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG, 4096);
+ UnifiedLog log = makeLog(LogConfig.fromProps(logConfig.originals(),
logProps));
+
+ short producerEpoch = 0;
+ long producerId = 1L;
+
+ log.appendAsLeader(record(0, 0), 0);
+ log.roll();
+
+ var appendProducer = appendTransactionalAsLeader(log, producerId,
producerEpoch);
+ appendProducer.append(List.of(1));
+ log.appendAsLeader(abortMarker(producerId, producerEpoch), 0,
AppendOrigin.REPLICATION,
+ RequestLocal.noCaching(), VerificationGuard.SENTINEL,
TransactionVersion.TV_UNKNOWN);
+ log.appendAsLeader(record(2, 2), 0);
+ log.roll();
+
+ List<LogSegment> sourceSegments = log.logSegments().subList(0, 2);
+ // maxCleanedSize fits the first source segment only; the second
segment's chunk overflows.
+ long maxCleanedSize = (long) sourceSegments.get(0).size() + 1L;
+
+ // No deletions; the offset map is empty.
+ Cleaner cleaner = makeCleaner(Integer.MAX_VALUE, tp -> { }, 64 * 1024,
maxCleanedSize, Integer.MAX_VALUE);
+ var offsetMap = new LogTestUtils.FakeOffsetMap(Integer.MAX_VALUE);
+
+ AbortedTxn expectedAbortedTxn = new
AbortedTxn().setProducerId(producerId)
+ .setFirstOffset(1).setLastOffset(2).setLastStableOffset(3);
+ assertAllAbortedTxns(List.of(expectedAbortedTxn), log);
+
+ log.updateHighWatermark(sourceSegments.get(1).readNextOffset());
+ cleaner.cleanSegments(log, sourceSegments, offsetMap, 0L,
+ new CleanerStats(Time.SYSTEM), new CleanedTransactionMetadata(),
-1);
+
+ // The aborted data must stay removed and the abort marker retained.
Review Comment:
The aborted data must stay removed => The aborted data must be removed
##########
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
Review Comment:
extra space after "continue with the filtered"
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java:
##########
@@ -341,6 +343,16 @@ public void cleanSegments(UnifiedLog log,
}
}
+ /**
+ * A filtered chunk that could not be appended to the current cleaned
segment without overflowing it.
+ *
+ * @param retained The filtered records to append to the next cleaned
segment. They are backed by
+ * the cleaner's shared write buffer and must be appended
before the next read.
Review Comment:
It's not clear what "the next read" refers to.
--
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]
