You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ka...@apache.org on 2024/11/09 03:22:22 UTC
(kafka) branch trunk updated: KAFKA-17801: RemoteLogManager may compute inaccurate upperBoundOffset for aborted txns (#17676)
This is an automated email from the ASF dual-hosted git repository.
kamalcph pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 324a74d4fd1 KAFKA-17801: RemoteLogManager may compute inaccurate upperBoundOffset for aborted txns (#17676)
324a74d4fd1 is described below
commit 324a74d4fd179614df648009fc294dfbf225c775
Author: Kamal Chandraprakash <ka...@gmail.com>
AuthorDate: Sat Nov 9 08:52:17 2024 +0530
KAFKA-17801: RemoteLogManager may compute inaccurate upperBoundOffset for aborted txns (#17676)
Reviewers: Jun Rao <ju...@gmail.com>
---
.../java/kafka/log/remote/RemoteLogManager.java | 39 ++++++++---
.../kafka/log/remote/RemoteLogManagerTest.java | 79 ++++++++++++++++++++--
2 files changed, 103 insertions(+), 15 deletions(-)
diff --git a/core/src/main/java/kafka/log/remote/RemoteLogManager.java b/core/src/main/java/kafka/log/remote/RemoteLogManager.java
index 4317f0c5bc3..477ef1c47c8 100644
--- a/core/src/main/java/kafka/log/remote/RemoteLogManager.java
+++ b/core/src/main/java/kafka/log/remote/RemoteLogManager.java
@@ -1672,25 +1672,26 @@ public class RemoteLogManager implements Closeable {
}
RemoteLogSegmentMetadata remoteLogSegmentMetadata = rlsMetadataOptional.get();
+ EnrichedRecordBatch enrichedRecordBatch = new EnrichedRecordBatch(null, 0);
InputStream remoteSegInputStream = null;
try {
int startPos = 0;
- RecordBatch firstBatch = null;
-
// 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.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);
- firstBatch = findFirstBatch(remoteLogInputStream, offset);
- if (firstBatch == null) {
+ enrichedRecordBatch = findFirstBatch(remoteLogInputStream, offset);
+ if (enrichedRecordBatch.batch == null) {
+ Utils.closeQuietly(remoteSegInputStream, "RemoteLogSegmentInputStream");
rlsMetadataOptional = findNextSegmentMetadata(rlsMetadataOptional.get(), logOptional.get().leaderEpochCache());
}
}
+ RecordBatch firstBatch = enrichedRecordBatch.batch;
if (firstBatch == null)
return new FetchDataInfo(new LogOffsetMetadata(offset), MemoryRecords.EMPTY, false,
includeAbortedTxns ? Optional.of(Collections.emptyList()) : Optional.empty());
@@ -1721,8 +1722,9 @@ public class RemoteLogManager implements Closeable {
}
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());
@@ -1730,7 +1732,9 @@ public class RemoteLogManager implements Closeable {
return fetchDataInfo;
} finally {
- Utils.closeQuietly(remoteSegInputStream, "RemoteLogSegmentInputStream");
+ if (enrichedRecordBatch.batch != null) {
+ Utils.closeQuietly(remoteSegInputStream, "RemoteLogSegmentInputStream");
+ }
}
}
// for testing
@@ -1888,15 +1892,18 @@ public class RemoteLogManager implements Closeable {
}
// 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 {
@@ -2247,4 +2254,14 @@ public class RemoteLogManager implements Closeable {
'}';
}
}
+
+ static class EnrichedRecordBatch {
+ private final RecordBatch batch;
+ private final int skippedBytes;
+
+ public EnrichedRecordBatch(RecordBatch batch, int skippedBytes) {
+ this.batch = batch;
+ this.skippedBytes = skippedBytes;
+ }
+ }
}
diff --git a/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java b/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java
index e75a6ca85d4..b25e439d06d 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.Option;
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.RemoteLogManagerConfig.
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;
@@ -3107,8 +3109,8 @@ public class RemoteLogManagerTest {
}
// 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);
@@ -3174,10 +3176,10 @@ public class RemoteLogManagerTest {
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);
@@ -3570,6 +3572,75 @@ public class RemoteLogManagerTest {
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<SimpleRecord> 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);