From 9185f8b6a0937411b29d72290e50aa6565b893e5 Mon Sep 17 00:00:00 2001 From: Kamal Chandraprakash Date: Sun, 3 Nov 2024 21:51:15 +0530 Subject: [PATCH 1/5] KAFKA-17801: RemoteLogManager may compute inaccurate upperBoundOffset for aborted txns --- .../kafka/log/remote/RemoteLogManager.java | 50 ++++++++++++++----- .../log/remote/RemoteLogManagerTest.java | 8 +-- 2 files changed, 42 insertions(+), 16 deletions(-) diff --git a/core/src/main/java/kafka/log/remote/RemoteLogManager.java b/core/src/main/java/kafka/log/remote/RemoteLogManager.java index f31a67260ea25..205c98e9fdac8 100644 --- a/core/src/main/java/kafka/log/remote/RemoteLogManager.java +++ b/core/src/main/java/kafka/log/remote/RemoteLogManager.java @@ -1654,22 +1654,22 @@ public FetchDataInfo read(RemoteStorageFetchInfo remoteStorageFetchInfo) throws InputStream remoteSegInputStream = null; try { int startPos = 0; - RecordBatch firstBatch = null; - + EnrichedRecordBatch enrichedRecordBatch = new EnrichedRecordBatch(); // Iteration over multiple RemoteSegmentMetadata is required in case of log compaction. // It may be possible the offset is log compacted in the current RemoteLogSegmentMetadata // And we need to iterate over the next segment metadata to fetch messages higher than the given offset. - while (firstBatch == null && rlsMetadataOptional.isPresent()) { + while (enrichedRecordBatch.getBatch() == null && rlsMetadataOptional.isPresent()) { remoteLogSegmentMetadata = rlsMetadataOptional.get(); // Search forward for the position of the last offset that is greater than or equal to the target offset startPos = lookupPositionForOffset(remoteLogSegmentMetadata, offset); remoteSegInputStream = remoteLogStorageManager.fetchLogSegment(remoteLogSegmentMetadata, startPos); RemoteLogInputStream remoteLogInputStream = getRemoteLogInputStream(remoteSegInputStream); - firstBatch = findFirstBatch(remoteLogInputStream, offset); - if (firstBatch == null) { + enrichedRecordBatch = findFirstBatch(remoteLogInputStream, offset); + if (enrichedRecordBatch.getBatch() == null) { rlsMetadataOptional = findNextSegmentMetadata(rlsMetadataOptional.get(), logOptional.get().leaderEpochCache()); } } + RecordBatch firstBatch = enrichedRecordBatch.getBatch(); if (firstBatch == null) return new FetchDataInfo(new LogOffsetMetadata(offset), MemoryRecords.EMPTY, false, includeAbortedTxns ? Optional.of(Collections.emptyList()) : Optional.empty()); @@ -1704,7 +1704,7 @@ public FetchDataInfo read(RemoteStorageFetchInfo remoteStorageFetchInfo) throws new LogOffsetMetadata(offset, remoteLogSegmentMetadata.startOffset(), startPos), MemoryRecords.readableRecords(buffer)); if (includeAbortedTxns) { - fetchDataInfo = addAbortedTransactions(firstBatch.baseOffset(), remoteLogSegmentMetadata, fetchDataInfo, logOptional.get()); + fetchDataInfo = addAbortedTransactions(firstBatch.baseOffset(), remoteLogSegmentMetadata, fetchDataInfo, logOptional.get(), enrichedRecordBatch.getSkippedBytes()); } return fetchDataInfo; @@ -1725,13 +1725,14 @@ int lookupPositionForOffset(RemoteLogSegmentMetadata remoteLogSegmentMetadata, l private FetchDataInfo addAbortedTransactions(long startOffset, RemoteLogSegmentMetadata segmentMetadata, FetchDataInfo fetchInfo, - UnifiedLog log) throws RemoteStorageException { + UnifiedLog log, + int skippedBytes) throws RemoteStorageException { int fetchSize = fetchInfo.records.sizeInBytes(); OffsetPosition startOffsetPosition = new OffsetPosition(fetchInfo.fetchOffsetMetadata.messageOffset, fetchInfo.fetchOffsetMetadata.relativePositionInSegment); OffsetIndex offsetIndex = indexCache.getIndexEntry(segmentMetadata).offsetIndex(); - long upperBoundOffset = offsetIndex.fetchUpperBoundOffset(startOffsetPosition, fetchSize) + long upperBoundOffset = offsetIndex.fetchUpperBoundOffset(startOffsetPosition, fetchSize + skippedBytes) .map(position -> position.offset).orElse(segmentMetadata.endOffset() + 1); final Set abortedTransactions = new HashSet<>(); @@ -1804,15 +1805,18 @@ Optional findNextSegmentMetadata(RemoteLogSegmentMetad } // Visible for testing - RecordBatch findFirstBatch(RemoteLogInputStream remoteLogInputStream, long offset) throws IOException { - RecordBatch nextBatch; + EnrichedRecordBatch findFirstBatch(RemoteLogInputStream remoteLogInputStream, long offset) throws IOException { + int skippedBytes = 0; + RecordBatch nextBatch = null; // Look for the batch which has the desired offset // We will always have a batch in that segment as it is a non-compacted topic. do { + if (nextBatch != null) { + skippedBytes += nextBatch.sizeInBytes(); + } nextBatch = remoteLogInputStream.nextBatch(); } while (nextBatch != null && nextBatch.lastOffset() < offset); - - return nextBatch; + return new EnrichedRecordBatch(nextBatch, skippedBytes); } OffsetAndEpoch findHighestRemoteOffset(TopicIdPartition topicIdPartition, UnifiedLog log) throws RemoteStorageException { @@ -2163,4 +2167,26 @@ public String toString() { '}'; } } + + static class EnrichedRecordBatch { + private final RecordBatch batch; + private final int skippedBytes; + + public EnrichedRecordBatch() { + this(null, 0); + } + + public EnrichedRecordBatch(RecordBatch batch, int skippedBytes) { + this.batch = batch; + this.skippedBytes = skippedBytes; + } + + public RecordBatch getBatch() { + return batch; + } + + public int getSkippedBytes() { + return skippedBytes; + } + } } diff --git a/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java b/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java index 4ea373327d968..9b78d426b9a9c 100644 --- a/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java +++ b/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java @@ -3021,8 +3021,8 @@ int lookupPositionForOffset(RemoteLogSegmentMetadata remoteLogSegmentMetadata, l } // This is the key scenario that we are testing here - RecordBatch findFirstBatch(RemoteLogInputStream remoteLogInputStream, long offset) { - return null; + EnrichedRecordBatch findFirstBatch(RemoteLogInputStream remoteLogInputStream, long offset) { + return new EnrichedRecordBatch(null, 0); } }) { FetchDataInfo fetchDataInfo = remoteLogManager.read(fetchInfo); @@ -3088,10 +3088,10 @@ int lookupPositionForOffset(RemoteLogSegmentMetadata remoteLogSegmentMetadata, l return 1; } - RecordBatch findFirstBatch(RemoteLogInputStream remoteLogInputStream, long offset) { + EnrichedRecordBatch findFirstBatch(RemoteLogInputStream remoteLogInputStream, long offset) { when(firstBatch.sizeInBytes()).thenReturn(recordBatchSizeInBytes); doNothing().when(firstBatch).writeTo(capture.capture()); - return firstBatch; + return new EnrichedRecordBatch(firstBatch, 0); } }) { FetchDataInfo fetchDataInfo = remoteLogManager.read(fetchInfo); From 080c6c276f53bc6d14c83d4af36768e0a2f60e81 Mon Sep 17 00:00:00 2001 From: Kamal Chandraprakash Date: Wed, 6 Nov 2024 09:58:24 +0530 Subject: [PATCH 2/5] Addressed the review comments --- .../main/java/kafka/log/remote/RemoteLogManager.java | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/core/src/main/java/kafka/log/remote/RemoteLogManager.java b/core/src/main/java/kafka/log/remote/RemoteLogManager.java index 205c98e9fdac8..1ec0248649820 100644 --- a/core/src/main/java/kafka/log/remote/RemoteLogManager.java +++ b/core/src/main/java/kafka/log/remote/RemoteLogManager.java @@ -1700,11 +1700,12 @@ public FetchDataInfo read(RemoteStorageFetchInfo remoteStorageFetchInfo) throws } buffer.flip(); + startPos = startPos + enrichedRecordBatch.skippedBytes; FetchDataInfo fetchDataInfo = new FetchDataInfo( - new LogOffsetMetadata(offset, remoteLogSegmentMetadata.startOffset(), startPos), + new LogOffsetMetadata(firstBatch.baseOffset(), remoteLogSegmentMetadata.startOffset(), startPos), MemoryRecords.readableRecords(buffer)); if (includeAbortedTxns) { - fetchDataInfo = addAbortedTransactions(firstBatch.baseOffset(), remoteLogSegmentMetadata, fetchDataInfo, logOptional.get(), enrichedRecordBatch.getSkippedBytes()); + fetchDataInfo = addAbortedTransactions(firstBatch.baseOffset(), remoteLogSegmentMetadata, fetchDataInfo, logOptional.get()); } return fetchDataInfo; @@ -1725,14 +1726,13 @@ int lookupPositionForOffset(RemoteLogSegmentMetadata remoteLogSegmentMetadata, l private FetchDataInfo addAbortedTransactions(long startOffset, RemoteLogSegmentMetadata segmentMetadata, FetchDataInfo fetchInfo, - UnifiedLog log, - int skippedBytes) throws RemoteStorageException { + UnifiedLog log) throws RemoteStorageException { int fetchSize = fetchInfo.records.sizeInBytes(); OffsetPosition startOffsetPosition = new OffsetPosition(fetchInfo.fetchOffsetMetadata.messageOffset, fetchInfo.fetchOffsetMetadata.relativePositionInSegment); OffsetIndex offsetIndex = indexCache.getIndexEntry(segmentMetadata).offsetIndex(); - long upperBoundOffset = offsetIndex.fetchUpperBoundOffset(startOffsetPosition, fetchSize + skippedBytes) + long upperBoundOffset = offsetIndex.fetchUpperBoundOffset(startOffsetPosition, fetchSize) .map(position -> position.offset).orElse(segmentMetadata.endOffset() + 1); final Set abortedTransactions = new HashSet<>(); From b6497980140d9c4315693ef4864f68badff49e98 Mon Sep 17 00:00:00 2001 From: Kamal Chandraprakash Date: Wed, 6 Nov 2024 10:00:53 +0530 Subject: [PATCH 3/5] Add unit test --- .../log/remote/RemoteLogManagerTest.java | 71 +++++++++++++++++++ 1 file changed, 71 insertions(+) diff --git a/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java b/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java index 9b78d426b9a9c..207fd3f12f31c 100644 --- a/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java +++ b/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java @@ -135,6 +135,7 @@ import scala.jdk.javaapi.CollectionConverters; import static kafka.log.remote.RemoteLogManager.isRemoteSegmentWithinLeaderEpochs; +import static org.apache.kafka.common.record.TimestampType.CREATE_TIME; import static org.apache.kafka.server.log.remote.metadata.storage.TopicBasedRemoteLogMetadataManagerConfig.REMOTE_LOG_METADATA_COMMON_CLIENT_PREFIX; import static org.apache.kafka.server.log.remote.metadata.storage.TopicBasedRemoteLogMetadataManagerConfig.REMOTE_LOG_METADATA_CONSUMER_PREFIX; import static org.apache.kafka.server.log.remote.metadata.storage.TopicBasedRemoteLogMetadataManagerConfig.REMOTE_LOG_METADATA_PRODUCER_PREFIX; @@ -150,6 +151,7 @@ import static org.apache.kafka.server.log.remote.storage.RemoteStorageMetrics.REMOTE_LOG_MANAGER_TASKS_AVG_IDLE_PERCENT_METRIC; import static org.apache.kafka.server.log.remote.storage.RemoteStorageMetrics.REMOTE_LOG_READER_FETCH_RATE_AND_TIME_METRIC; import static org.apache.kafka.server.log.remote.storage.RemoteStorageMetrics.REMOTE_STORAGE_THREAD_POOL_METRICS; +import static org.apache.kafka.test.TestUtils.tempFile; import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; @@ -3484,6 +3486,75 @@ public void testTierLagResetsToZeroOnBecomingFollower() { assertEquals(0, brokerTopicStats.topicStats(leaderTopicIdPartition.topic()).remoteCopyLagSegments()); } + @Test + public void testRemoteReadFetchDataInfo() throws RemoteStorageException, IOException { + checkpoint.write(totalEpochEntries); + LeaderEpochFileCache cache = new LeaderEpochFileCache(tp, checkpoint, scheduler); + when(mockLog.leaderEpochCache()).thenReturn(Option.apply(cache)); + when(remoteLogMetadataManager.remoteLogSegmentMetadata(eq(leaderTopicIdPartition), anyInt(), anyLong())) + .thenAnswer(ans -> { + long offset = ans.getArgument(2); + RemoteLogSegmentId segmentId = new RemoteLogSegmentId(leaderTopicIdPartition, Uuid.randomUuid()); + RemoteLogSegmentMetadata segmentMetadata = createRemoteLogSegmentMetadata(segmentId, + offset - 10, offset + 99, 1024, totalEpochEntries, RemoteLogSegmentState.COPY_SEGMENT_FINISHED); + return Optional.of(segmentMetadata); + }); + + File segmentFile = tempFile(); + appendRecordsToFile(segmentFile, 100, 3); + FileInputStream fileInputStream = new FileInputStream(segmentFile); + when(remoteStorageManager.fetchLogSegment(any(RemoteLogSegmentMetadata.class), anyInt())) + .thenReturn(fileInputStream); + + RemoteLogManager remoteLogManager = new RemoteLogManager(config.remoteLogManagerConfig(), brokerId, logDir, clusterId, time, + tp -> Optional.of(mockLog), + (topicPartition, offset) -> currentLogStartOffset.set(offset), + brokerTopicStats, metrics) { + public RemoteStorageManager createRemoteStorageManager() { + return remoteStorageManager; + } + public RemoteLogMetadataManager createRemoteLogMetadataManager() { + return remoteLogMetadataManager; + } + int lookupPositionForOffset(RemoteLogSegmentMetadata remoteLogSegmentMetadata, long offset) { + return 0; + } + }; + remoteLogManager.startup(); + remoteLogManager.onLeadershipChange( + Collections.singleton(mockPartition(leaderTopicIdPartition)), Collections.emptySet(), topicIds); + + long fetchOffset = 10; + FetchRequest.PartitionData partitionData = new FetchRequest.PartitionData( + Uuid.randomUuid(), fetchOffset, 0, 100, Optional.empty()); + RemoteStorageFetchInfo remoteStorageFetchInfo = new RemoteStorageFetchInfo( + 1048576, true, leaderTopicIdPartition.topicPartition(), + partitionData, FetchIsolation.HIGH_WATERMARK, false); + FetchDataInfo fetchDataInfo = remoteLogManager.read(remoteStorageFetchInfo); + // firstBatch baseOffset may not be equal to the fetchOffset + assertEquals(9, fetchDataInfo.fetchOffsetMetadata.messageOffset); + assertEquals(273, fetchDataInfo.fetchOffsetMetadata.relativePositionInSegment); + } + + private void appendRecordsToFile(File file, int nRecords, int nRecordsPerBatch) throws IOException { + byte magic = RecordBatch.CURRENT_MAGIC_VALUE; + Compression compression = Compression.NONE; + long offset = 0; + List records = new ArrayList<>(); + try (FileRecords fileRecords = FileRecords.open(file)) { + for (long counter = 1; counter < nRecords + 1; counter++) { + records.add(new SimpleRecord("foo".getBytes())); + if (counter % nRecordsPerBatch == 0) { + fileRecords.append(MemoryRecords.withRecords(magic, offset, compression, CREATE_TIME, + records.toArray(new SimpleRecord[0]))); + offset += records.size(); + records.clear(); + } + } + fileRecords.flush(); + } + } + private Partition mockPartition(TopicIdPartition topicIdPartition) { TopicPartition tp = topicIdPartition.topicPartition(); Partition partition = mock(Partition.class); From 582f7aaade0c851b4a3e0df37948a0920ce9fb40 Mon Sep 17 00:00:00 2001 From: Kamal Chandraprakash Date: Thu, 7 Nov 2024 07:59:57 +0530 Subject: [PATCH 4/5] Address the review comments. --- .../kafka/log/remote/RemoteLogManager.java | 21 +++++-------------- 1 file changed, 5 insertions(+), 16 deletions(-) diff --git a/core/src/main/java/kafka/log/remote/RemoteLogManager.java b/core/src/main/java/kafka/log/remote/RemoteLogManager.java index 1ec0248649820..7b973bbac2d1c 100644 --- a/core/src/main/java/kafka/log/remote/RemoteLogManager.java +++ b/core/src/main/java/kafka/log/remote/RemoteLogManager.java @@ -1654,22 +1654,23 @@ public FetchDataInfo read(RemoteStorageFetchInfo remoteStorageFetchInfo) throws InputStream remoteSegInputStream = null; try { int startPos = 0; - EnrichedRecordBatch enrichedRecordBatch = new EnrichedRecordBatch(); + EnrichedRecordBatch enrichedRecordBatch = new EnrichedRecordBatch(null, 0); // Iteration over multiple RemoteSegmentMetadata is required in case of log compaction. // It may be possible the offset is log compacted in the current RemoteLogSegmentMetadata // And we need to iterate over the next segment metadata to fetch messages higher than the given offset. - while (enrichedRecordBatch.getBatch() == null && rlsMetadataOptional.isPresent()) { + while (enrichedRecordBatch.batch == null && rlsMetadataOptional.isPresent()) { remoteLogSegmentMetadata = rlsMetadataOptional.get(); // Search forward for the position of the last offset that is greater than or equal to the target offset startPos = lookupPositionForOffset(remoteLogSegmentMetadata, offset); remoteSegInputStream = remoteLogStorageManager.fetchLogSegment(remoteLogSegmentMetadata, startPos); RemoteLogInputStream remoteLogInputStream = getRemoteLogInputStream(remoteSegInputStream); enrichedRecordBatch = findFirstBatch(remoteLogInputStream, offset); - if (enrichedRecordBatch.getBatch() == null) { + if (enrichedRecordBatch.batch == null) { + Utils.closeQuietly(remoteSegInputStream, "RemoteLogSegmentInputStream"); rlsMetadataOptional = findNextSegmentMetadata(rlsMetadataOptional.get(), logOptional.get().leaderEpochCache()); } } - RecordBatch firstBatch = enrichedRecordBatch.getBatch(); + RecordBatch firstBatch = enrichedRecordBatch.batch; if (firstBatch == null) return new FetchDataInfo(new LogOffsetMetadata(offset), MemoryRecords.EMPTY, false, includeAbortedTxns ? Optional.of(Collections.emptyList()) : Optional.empty()); @@ -2172,21 +2173,9 @@ static class EnrichedRecordBatch { private final RecordBatch batch; private final int skippedBytes; - public EnrichedRecordBatch() { - this(null, 0); - } - public EnrichedRecordBatch(RecordBatch batch, int skippedBytes) { this.batch = batch; this.skippedBytes = skippedBytes; } - - public RecordBatch getBatch() { - return batch; - } - - public int getSkippedBytes() { - return skippedBytes; - } } } From 7e90c28f4c84c13844ac912f48dca78b950c554b Mon Sep 17 00:00:00 2001 From: Kamal Chandraprakash Date: Fri, 8 Nov 2024 09:06:23 +0530 Subject: [PATCH 5/5] Close the input stream once --- core/src/main/java/kafka/log/remote/RemoteLogManager.java | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/core/src/main/java/kafka/log/remote/RemoteLogManager.java b/core/src/main/java/kafka/log/remote/RemoteLogManager.java index 7b973bbac2d1c..38cd6b9ac2cf8 100644 --- a/core/src/main/java/kafka/log/remote/RemoteLogManager.java +++ b/core/src/main/java/kafka/log/remote/RemoteLogManager.java @@ -1651,10 +1651,10 @@ public FetchDataInfo read(RemoteStorageFetchInfo remoteStorageFetchInfo) throws } RemoteLogSegmentMetadata remoteLogSegmentMetadata = rlsMetadataOptional.get(); + EnrichedRecordBatch enrichedRecordBatch = new EnrichedRecordBatch(null, 0); InputStream remoteSegInputStream = null; try { int startPos = 0; - EnrichedRecordBatch enrichedRecordBatch = new EnrichedRecordBatch(null, 0); // Iteration over multiple RemoteSegmentMetadata is required in case of log compaction. // It may be possible the offset is log compacted in the current RemoteLogSegmentMetadata // And we need to iterate over the next segment metadata to fetch messages higher than the given offset. @@ -1711,7 +1711,9 @@ public FetchDataInfo read(RemoteStorageFetchInfo remoteStorageFetchInfo) throws return fetchDataInfo; } finally { - Utils.closeQuietly(remoteSegInputStream, "RemoteLogSegmentInputStream"); + if (enrichedRecordBatch.batch != null) { + Utils.closeQuietly(remoteSegInputStream, "RemoteLogSegmentInputStream"); + } } } // for testing