You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by sh...@apache.org on 2024/05/08 07:21:30 UTC
(kafka) branch trunk updated: KAFKA-16511: Fix the leaking tiered segments during segment deletion (#15817)
This is an automated email from the ASF dual-hosted git repository.
showuon 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 8655094e6c9 KAFKA-16511: Fix the leaking tiered segments during segment deletion (#15817)
8655094e6c9 is described below
commit 8655094e6c991cc4b40c5e3721ae4a9d69adc843
Author: Kamal Chandraprakash <kc...@uber.com>
AuthorDate: Wed May 8 12:51:23 2024 +0530
KAFKA-16511: Fix the leaking tiered segments during segment deletion (#15817)
When there are overlapping segments in the remote storage, then the deletion may fail to remove the segments due to isRemoteSegmentWithinLeaderEpochs check. Once the deletion starts to fail for a partition, then segments won't be eligible for cleanup. The one workaround that we have is to move the log-start-offset using the kafka-delete-records script.
Reviewers: Luke Chen <sh...@gmail.com>, Satish Duggana <sa...@apache.org>
---
.../java/kafka/log/remote/RemoteLogManager.java | 77 +++++++++++++---------
.../kafka/log/remote/RemoteLogManagerTest.java | 73 ++++++++++++++++----
2 files changed, 106 insertions(+), 44 deletions(-)
diff --git a/core/src/main/java/kafka/log/remote/RemoteLogManager.java b/core/src/main/java/kafka/log/remote/RemoteLogManager.java
index 262702a75b6..b2985f9d7ca 100644
--- a/core/src/main/java/kafka/log/remote/RemoteLogManager.java
+++ b/core/src/main/java/kafka/log/remote/RemoteLogManager.java
@@ -1217,26 +1217,39 @@ public class RemoteLogManager implements Closeable {
* @return true if the remote segment's epoch/offsets are within the leader epoch lineage of the partition.
*/
// Visible for testing
- public static boolean isRemoteSegmentWithinLeaderEpochs(RemoteLogSegmentMetadata segmentMetadata,
- long logEndOffset,
- NavigableMap<Integer, Long> leaderEpochs) {
+ static boolean isRemoteSegmentWithinLeaderEpochs(RemoteLogSegmentMetadata segmentMetadata,
+ long logEndOffset,
+ NavigableMap<Integer, Long> leaderEpochs) {
long segmentEndOffset = segmentMetadata.endOffset();
// Filter epochs that does not have any messages/records associated with them.
NavigableMap<Integer, Long> segmentLeaderEpochs = buildFilteredLeaderEpochMap(segmentMetadata.segmentLeaderEpochs());
// Check for out of bound epochs between segment epochs and current leader epochs.
- Integer segmentFirstEpoch = segmentLeaderEpochs.firstKey();
Integer segmentLastEpoch = segmentLeaderEpochs.lastKey();
- if (segmentFirstEpoch < leaderEpochs.firstKey() || segmentLastEpoch > leaderEpochs.lastKey()) {
+ if (segmentLastEpoch < leaderEpochs.firstKey() || segmentLastEpoch > leaderEpochs.lastKey()) {
+ LOGGER.debug("Segment {} is not within the partition leader epoch lineage. " +
+ "Remote segment epochs: {} and partition leader epochs: {}",
+ segmentMetadata.remoteLogSegmentId(), segmentLeaderEpochs, leaderEpochs);
+ return false;
+ }
+ // There can be overlapping remote log segments in the remote storage. (eg)
+ // leader-epoch-file-cache: {(5, 10), (7, 15), (9, 100)}
+ // segment1: offset-range = 5-50, Broker = 0, epochs = {(5, 10), (7, 15)}
+ // segment2: offset-range = 14-150, Broker = 1, epochs = {(5, 14), (7, 15), (9, 100)}, after leader-election.
+ // When the segment1 gets deleted, then the log-start-offset = 51 and leader-epoch-file-cache gets updated to: {(7, 51), (9, 100)}.
+ // While validating the segment2, we should ensure the overlapping remote log segments case.
+ Integer segmentFirstEpoch = segmentLeaderEpochs.ceilingKey(leaderEpochs.firstKey());
+ if (segmentFirstEpoch == null) {
LOGGER.debug("Segment {} is not within the partition leader epoch lineage. " +
"Remote segment epochs: {} and partition leader epochs: {}",
segmentMetadata.remoteLogSegmentId(), segmentLeaderEpochs, leaderEpochs);
return false;
}
-
for (Map.Entry<Integer, Long> entry : segmentLeaderEpochs.entrySet()) {
int epoch = entry.getKey();
long offset = entry.getValue();
-
+ if (epoch < segmentFirstEpoch) {
+ continue;
+ }
// If segment's epoch does not exist in the leader epoch lineage then it is not a valid segment.
if (!leaderEpochs.containsKey(epoch)) {
LOGGER.debug("Segment {} epoch {} is not within the leader epoch lineage. " +
@@ -1244,24 +1257,29 @@ public class RemoteLogManager implements Closeable {
segmentMetadata.remoteLogSegmentId(), epoch, segmentLeaderEpochs, leaderEpochs);
return false;
}
-
- // Segment's first epoch's offset should be more than or equal to the respective leader epoch's offset.
- if (epoch == segmentFirstEpoch && offset < leaderEpochs.get(epoch)) {
- LOGGER.debug("Segment {} first epoch {} offset is less than leader epoch offset {}.",
- segmentMetadata.remoteLogSegmentId(), epoch, leaderEpochs.get(epoch));
+ // Two cases:
+ // case-1: When the segment-first-epoch equals to the first-epoch in the leader-epoch-lineage, then the
+ // offset value can lie anywhere between 0 to (next-epoch-start-offset - 1) is valid.
+ // case-2: When the segment-first-epoch is not equal to the first-epoch in the leader-epoch-lineage, then
+ // the offset value should be between (current-epoch-start-offset) to (next-epoch-start-offset - 1).
+ if (epoch == segmentFirstEpoch && leaderEpochs.lowerKey(epoch) != null && offset < leaderEpochs.get(epoch)) {
+ LOGGER.debug("Segment {} first-valid epoch {} offset is less than first leader epoch offset {}." +
+ "Remote segment epochs: {} and partition leader epochs: {}",
+ segmentMetadata.remoteLogSegmentId(), epoch, leaderEpochs.get(epoch),
+ segmentLeaderEpochs, leaderEpochs);
return false;
}
-
// Segment's end offset should be less than or equal to the respective leader epoch's offset.
if (epoch == segmentLastEpoch) {
Map.Entry<Integer, Long> nextEntry = leaderEpochs.higherEntry(epoch);
if (nextEntry != null && segmentEndOffset > nextEntry.getValue() - 1) {
- LOGGER.debug("Segment {} end offset {} is more than leader epoch offset {}.",
- segmentMetadata.remoteLogSegmentId(), segmentEndOffset, nextEntry.getValue() - 1);
+ LOGGER.debug("Segment {} end offset {} is more than leader epoch offset {}." +
+ "Remote segment epochs: {} and partition leader epochs: {}",
+ segmentMetadata.remoteLogSegmentId(), segmentEndOffset, nextEntry.getValue() - 1,
+ segmentLeaderEpochs, leaderEpochs);
return false;
}
}
-
// Next segment epoch entry and next leader epoch entry should be same to ensure that the segment's epoch
// is within the leader epoch lineage.
if (epoch != segmentLastEpoch && !leaderEpochs.higherEntry(epoch).equals(segmentLeaderEpochs.higherEntry(epoch))) {
@@ -1283,7 +1301,6 @@ public class RemoteLogManager implements Closeable {
/**
* Returns a map containing the epoch vs start-offset for the given leader epoch map by filtering the epochs that
* does not contain any messages/records associated with them.
- *
* For ex:
* <pre>
* {@code
@@ -1298,8 +1315,7 @@ public class RemoteLogManager implements Closeable {
* 7 - 70
* }
* </pre>
- *
- * When the above leaderEpochMap is passed to this method, it returns the following map:
+ * When the above leaderEpochMap is passed to this method, it returns the following map:
* <pre>
* {@code
* <epoch - start offset>
@@ -1312,27 +1328,24 @@ public class RemoteLogManager implements Closeable {
* 7 - 70
* }
* </pre>
- *
* @param leaderEpochs The leader epoch map to be refined.
*/
// Visible for testing
- public static NavigableMap<Integer, Long> buildFilteredLeaderEpochMap(NavigableMap<Integer, Long> leaderEpochs) {
- List<Integer> duplicatedEpochs = new ArrayList<>();
- Map.Entry<Integer, Long> previousEntry = null;
- for (Map.Entry<Integer, Long> entry : leaderEpochs.entrySet()) {
- if (previousEntry != null && previousEntry.getValue().equals(entry.getValue())) {
- duplicatedEpochs.add(previousEntry.getKey());
+ static NavigableMap<Integer, Long> buildFilteredLeaderEpochMap(NavigableMap<Integer, Long> leaderEpochs) {
+ List<Integer> epochsWithNoMessages = new ArrayList<>();
+ Map.Entry<Integer, Long> previousEpochAndOffset = null;
+ for (Map.Entry<Integer, Long> currentEpochAndOffset : leaderEpochs.entrySet()) {
+ if (previousEpochAndOffset != null && previousEpochAndOffset.getValue().equals(currentEpochAndOffset.getValue())) {
+ epochsWithNoMessages.add(previousEpochAndOffset.getKey());
}
- previousEntry = entry;
+ previousEpochAndOffset = currentEpochAndOffset;
}
-
- if (duplicatedEpochs.isEmpty()) {
+ if (epochsWithNoMessages.isEmpty()) {
return leaderEpochs;
}
-
TreeMap<Integer, Long> filteredLeaderEpochs = new TreeMap<>(leaderEpochs);
- for (Integer duplicatedEpoch : duplicatedEpochs) {
- filteredLeaderEpochs.remove(duplicatedEpoch);
+ for (Integer epochWithNoMessage : epochsWithNoMessages) {
+ filteredLeaderEpochs.remove(epochWithNoMessage);
}
return filteredLeaderEpochs;
}
diff --git a/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java b/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java
index 193b5c15d68..85a3a3a8fa5 100644
--- a/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java
+++ b/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java
@@ -152,6 +152,8 @@ import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoMoreInteractions;
import static org.mockito.Mockito.when;
+import static kafka.log.remote.RemoteLogManager.isRemoteSegmentWithinLeaderEpochs;
+
public class RemoteLogManagerTest {
private final Time time = new MockTime();
private final int brokerId = 0;
@@ -1475,7 +1477,7 @@ public class RemoteLogManagerTest {
segmentEpochs1.put(2, 20L);
segmentEpochs1.put(3, 30L);
- assertTrue(RemoteLogManager.isRemoteSegmentWithinLeaderEpochs(createRemoteLogSegmentMetadata(
+ assertTrue(isRemoteSegmentWithinLeaderEpochs(createRemoteLogSegmentMetadata(
15,
35,
segmentEpochs1), logEndOffset, leaderEpochToStartOffset));
@@ -1483,7 +1485,7 @@ public class RemoteLogManagerTest {
// Test whether a remote segment's epochs/offsets(single) are within the range of leader epochs
TreeMap<Integer, Long> segmentEpochs2 = new TreeMap<>();
segmentEpochs2.put(1, 15L);
- assertTrue(RemoteLogManager.isRemoteSegmentWithinLeaderEpochs(createRemoteLogSegmentMetadata(
+ assertTrue(isRemoteSegmentWithinLeaderEpochs(createRemoteLogSegmentMetadata(
15,
19,
segmentEpochs2), logEndOffset, leaderEpochToStartOffset));
@@ -1491,7 +1493,7 @@ public class RemoteLogManagerTest {
// Test whether a remote segment's start offset is same as the offset of the respective leader epoch entry.
TreeMap<Integer, Long> segmentEpochs3 = new TreeMap<>();
segmentEpochs3.put(0, 0L); // same as leader epoch's start offset
- assertTrue(RemoteLogManager.isRemoteSegmentWithinLeaderEpochs(createRemoteLogSegmentMetadata(
+ assertTrue(isRemoteSegmentWithinLeaderEpochs(createRemoteLogSegmentMetadata(
0,
5,
segmentEpochs3), logEndOffset, leaderEpochToStartOffset));
@@ -1499,18 +1501,17 @@ public class RemoteLogManagerTest {
// Test whether a remote segment's start offset is same as the offset of the respective leader epoch entry.
TreeMap<Integer, Long> segmentEpochs4 = new TreeMap<>();
segmentEpochs4.put(7, 70L); // same as leader epoch's start offset
- assertTrue(RemoteLogManager.isRemoteSegmentWithinLeaderEpochs(createRemoteLogSegmentMetadata(
+ assertTrue(isRemoteSegmentWithinLeaderEpochs(createRemoteLogSegmentMetadata(
70,
75,
segmentEpochs4), logEndOffset, leaderEpochToStartOffset));
-
// Test whether a remote segment's end offset is same as the end offset of the respective leader epoch entry.
TreeMap<Integer, Long> segmentEpochs5 = new TreeMap<>();
segmentEpochs5.put(1, 15L);
segmentEpochs5.put(2, 20L);
- assertTrue(RemoteLogManager.isRemoteSegmentWithinLeaderEpochs(createRemoteLogSegmentMetadata(
+ assertTrue(isRemoteSegmentWithinLeaderEpochs(createRemoteLogSegmentMetadata(
15,
29, // same as end offset for epoch 2 in leaderEpochToStartOffset
segmentEpochs5), logEndOffset, leaderEpochToStartOffset));
@@ -1521,7 +1522,7 @@ public class RemoteLogManagerTest {
segmentEpochs6.put(6, 60L); // epoch 6 exists here but it is missing in leaderEpochToStartOffset
segmentEpochs6.put(7, 70L);
- assertFalse(RemoteLogManager.isRemoteSegmentWithinLeaderEpochs(createRemoteLogSegmentMetadata(
+ assertFalse(isRemoteSegmentWithinLeaderEpochs(createRemoteLogSegmentMetadata(
55,
85,
segmentEpochs6), logEndOffset, leaderEpochToStartOffset));
@@ -1532,18 +1533,18 @@ public class RemoteLogManagerTest {
segmentEpochs7.put(2, 20L); // epoch 3 is missing here which exists in leaderEpochToStartOffset
segmentEpochs7.put(4, 40L);
- assertFalse(RemoteLogManager.isRemoteSegmentWithinLeaderEpochs(createRemoteLogSegmentMetadata(
+ assertFalse(isRemoteSegmentWithinLeaderEpochs(createRemoteLogSegmentMetadata(
15,
45,
segmentEpochs7), logEndOffset, leaderEpochToStartOffset));
// Test a remote segment having larger end offset than the log end offset
- assertFalse(RemoteLogManager.isRemoteSegmentWithinLeaderEpochs(createRemoteLogSegmentMetadata(
+ assertFalse(isRemoteSegmentWithinLeaderEpochs(createRemoteLogSegmentMetadata(
15,
95, // larger than log end offset
leaderEpochToStartOffset), logEndOffset, leaderEpochToStartOffset));
- assertFalse(RemoteLogManager.isRemoteSegmentWithinLeaderEpochs(createRemoteLogSegmentMetadata(
+ assertFalse(isRemoteSegmentWithinLeaderEpochs(createRemoteLogSegmentMetadata(
15,
90, // equal to the log end offset
leaderEpochToStartOffset), logEndOffset, leaderEpochToStartOffset));
@@ -1553,7 +1554,7 @@ public class RemoteLogManagerTest {
segmentEpochs9.put(1, 5L);
segmentEpochs9.put(2, 20L);
- assertFalse(RemoteLogManager.isRemoteSegmentWithinLeaderEpochs(createRemoteLogSegmentMetadata(
+ assertFalse(isRemoteSegmentWithinLeaderEpochs(createRemoteLogSegmentMetadata(
5, // earlier to epoch 1's start offset
25,
segmentEpochs9), logEndOffset, leaderEpochToStartOffset));
@@ -1562,12 +1563,60 @@ public class RemoteLogManagerTest {
TreeMap<Integer, Long> segmentEpochs10 = new TreeMap<>();
segmentEpochs10.put(1, 15L);
segmentEpochs10.put(2, 20L);
- assertFalse(RemoteLogManager.isRemoteSegmentWithinLeaderEpochs(createRemoteLogSegmentMetadata(
+ assertFalse(isRemoteSegmentWithinLeaderEpochs(createRemoteLogSegmentMetadata(
15,
35, // more than epoch 2's end offset
segmentEpochs10), logEndOffset, leaderEpochToStartOffset));
}
+ @Test
+ public void testRemoteSegmentWithinLeaderEpochsForOverlappingSegments() {
+ NavigableMap<Integer, Long> leaderEpochCache = new TreeMap<>();
+ leaderEpochCache.put(7, 51L);
+ leaderEpochCache.put(9, 100L);
+
+ TreeMap<Integer, Long> segment1Epochs = new TreeMap<>();
+ segment1Epochs.put(5, 14L);
+ segment1Epochs.put(7, 15L);
+ segment1Epochs.put(9, 100L);
+ RemoteLogSegmentMetadata segment1 = createRemoteLogSegmentMetadata(14, 150, segment1Epochs);
+ assertTrue(isRemoteSegmentWithinLeaderEpochs(segment1, 210, leaderEpochCache));
+
+ // segment2Epochs are not within the leaderEpochCache
+ TreeMap<Integer, Long> segment2Epochs = new TreeMap<>();
+ segment2Epochs.put(2, 5L);
+ segment2Epochs.put(3, 6L);
+ RemoteLogSegmentMetadata segment2 = createRemoteLogSegmentMetadata(2, 7, segment2Epochs);
+ assertFalse(isRemoteSegmentWithinLeaderEpochs(segment2, 210, leaderEpochCache));
+
+ // segment3Epochs are not within the leaderEpochCache
+ TreeMap<Integer, Long> segment3Epochs = new TreeMap<>();
+ segment3Epochs.put(7, 15L);
+ segment3Epochs.put(9, 100L);
+ segment3Epochs.put(10, 200L);
+ RemoteLogSegmentMetadata segment3 = createRemoteLogSegmentMetadata(15, 250, segment3Epochs);
+ assertFalse(isRemoteSegmentWithinLeaderEpochs(segment3, 210, leaderEpochCache));
+
+ // segment4Epochs are not within the leaderEpochCache
+ TreeMap<Integer, Long> segment4Epochs = new TreeMap<>();
+ segment4Epochs.put(8, 75L);
+ RemoteLogSegmentMetadata segment4 = createRemoteLogSegmentMetadata(75, 100, segment4Epochs);
+ assertFalse(isRemoteSegmentWithinLeaderEpochs(segment4, 210, leaderEpochCache));
+
+ // segment5Epochs does not match with the leaderEpochCache
+ TreeMap<Integer, Long> segment5Epochs = new TreeMap<>();
+ segment5Epochs.put(7, 15L);
+ segment5Epochs.put(9, 101L);
+ RemoteLogSegmentMetadata segment5 = createRemoteLogSegmentMetadata(15, 150, segment5Epochs);
+ assertFalse(isRemoteSegmentWithinLeaderEpochs(segment5, 210, leaderEpochCache));
+
+ // segment6Epochs does not match with the leaderEpochCache
+ TreeMap<Integer, Long> segment6Epochs = new TreeMap<>();
+ segment6Epochs.put(9, 99L);
+ RemoteLogSegmentMetadata segment6 = createRemoteLogSegmentMetadata(99, 150, segment6Epochs);
+ assertFalse(isRemoteSegmentWithinLeaderEpochs(segment6, 210, leaderEpochCache));
+ }
+
@Test
public void testCandidateLogSegmentsSkipsActiveSegment() {
UnifiedLog log = mock(UnifiedLog.class);