mattwong949 commented on a change in pull request #10914:
URL: https://github.com/apache/kafka/pull/10914#discussion_r700663048
##########
File path:
clients/src/test/java/org/apache/kafka/common/record/MemoryRecordsTest.java
##########
@@ -483,6 +499,53 @@ public void testBuildEndTxnMarker() {
assertEquals(coordinatorEpoch, deserializedMarker.coordinatorEpoch());
}
+ /**
+ * This test is used to see if the base timestamp of the batch has been
successfully
+ * converted to a delete horizon for the tombstones / transaction markers
of the batch.
+ * It also verifies that the record timestamps remain correct as a delta
relative to the delete horizon.
+ */
+ @ParameterizedTest
+ @ArgumentsSource(V2MemoryRecordsArgumentsProvider.class)
+ public void testBaseTimestampToDeleteHorizonConversion(Args args) {
+ int partitionLeaderEpoch = 998;
+ ByteBuffer buffer = ByteBuffer.allocate(2048);
+ MemoryRecordsBuilder builder = MemoryRecords.builder(buffer,
args.magic, args.compression, TimestampType.CREATE_TIME,
+ 0L, RecordBatch.NO_TIMESTAMP, partitionLeaderEpoch);
+ builder.append(5L, "0".getBytes(), "0".getBytes());
+ builder.append(10L, "1".getBytes(), null);
+ builder.append(15L, "2".getBytes(), "2".getBytes());
+
+ ByteBuffer filtered = ByteBuffer.allocate(2048);
+ final long deleteHorizon = Integer.MAX_VALUE / 2;
+ final RecordFilter recordFilter = new
MemoryRecords.RecordFilter(deleteHorizon - 1, 1) {
+ @Override
+ protected boolean shouldRetainRecord(RecordBatch recordBatch,
Record record) {
+ return true;
+ }
+
+ @Override
+ protected BatchRetentionResult checkBatchRetention(RecordBatch
batch) {
+ return new BatchRetentionResult(BatchRetention.RETAIN_EMPTY,
true);
Review comment:
makes sense, I can change this to return `false`
--
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]
