Skip to content

KAFKA-17801: RemoteLogManager may compute inaccurate upperBoundOffset for aborted txns #17676

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
kamalcph merged 5 commits into from
Nov 9, 2024
+103 −15
Merged

KAFKA-17801: RemoteLogManager may compute inaccurate upperBoundOffset for aborted txns #17676

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
9185f8b
KAFKA-17801: RemoteLogManager may compute inaccurate upperBoundOffset…
kamalcph Nov 3, 2024
080c6c2
Addressed the review comments
kamalcph Nov 6, 2024
b649798
Add unit test
kamalcph Nov 6, 2024
582f7aa
Address the review comments.
kamalcph Nov 7, 2024
7e90c28
Close the input stream once
kamalcph Nov 8, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 28 additions & 11 deletions core/src/main/java/kafka/log/remote/RemoteLogManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1651,25 +1651,26 @@ public FetchDataInfo read(RemoteStorageFetchInfo remoteStorageFetchInfo) throws
}

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);
Comment thread
kamalcph marked this conversation as resolved.
if (enrichedRecordBatch.batch == null) {
Utils.closeQuietly(remoteSegInputStream, "RemoteLogSegmentInputStream");
Comment thread
kamalcph marked this conversation as resolved.
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());
Expand Down Expand Up @@ -1700,16 +1701,19 @@ 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());
}

return fetchDataInfo;
} finally {
Utils.closeQuietly(remoteSegInputStream, "RemoteLogSegmentInputStream");
if (enrichedRecordBatch.batch != null) {
Utils.closeQuietly(remoteSegInputStream, "RemoteLogSegmentInputStream");
}
}
}
// for testing
Expand Down Expand Up @@ -1804,15 +1808,18 @@ Optional<RemoteLogSegmentMetadata> 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 {
Expand Down Expand Up @@ -2163,4 +2170,14 @@ public String toString() {
'}';
}
}

static class EnrichedRecordBatch {
private final RecordBatch batch;
private final int skippedBytes;

public EnrichedRecordBatch(RecordBatch batch, int skippedBytes) {
this.batch = batch;
this.skippedBytes = skippedBytes;
}
}
}
79 changes: 75 additions & 4 deletions core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -3021,8 +3023,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);
Expand Down Expand Up @@ -3088,10 +3090,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);
Expand Down Expand Up @@ -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<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])));
Comment thread
kamalcph marked this conversation as resolved.
offset += records.size();
records.clear();
}
}
fileRecords.flush();
Comment thread
kamalcph marked this conversation as resolved.
}
}

private Partition mockPartition(TopicIdPartition topicIdPartition) {
TopicPartition tp = topicIdPartition.topicPartition();
Partition partition = mock(Partition.class);
Expand Down