This is an automated email from the ASF dual-hosted git repository.
satishd pushed a commit to branch 3.6
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/3.6 by this push:
new 946ab8f4100 KAFKA-15410: Delete records with tiered storage
integration test (4/4) (#14330)
946ab8f4100 is described below
commit 946ab8f4100e4f923648d1d38eca2d1640b53f79
Author: Kamal Chandraprakash <kchandraprak...@uber.com>
AuthorDate: Thu Sep 7 21:02:39 2023 +0530
KAFKA-15410: Delete records with tiered storage integration test (4/4)
(#14330)
* Added the integration test for DELETE_RECORDS API for tiered storage
enabled topic
* Added validation checks before removing remote log segments for
log-start-offset breach
Reviewers: Satish Duggana <sati...@apache.org>, Luke Chen
<show...@gmail.com>, Christo Lolov <lol...@amazon.com>
---
.../java/kafka/log/remote/RemoteLogManager.java | 108 +++++++++++++--------
.../tiered/storage/TieredStorageTestBuilder.java | 8 +-
.../storage/actions/DeleteRecordsAction.java | 39 +++++++-
...eleteSegmentsDueToLogStartOffsetBreachTest.java | 92 ++++++++++++++++++
4 files changed, 202 insertions(+), 45 deletions(-)
diff --git a/core/src/main/java/kafka/log/remote/RemoteLogManager.java
b/core/src/main/java/kafka/log/remote/RemoteLogManager.java
index 9b70cfbe568..4a35abf6a11 100644
--- a/core/src/main/java/kafka/log/remote/RemoteLogManager.java
+++ b/core/src/main/java/kafka/log/remote/RemoteLogManager.java
@@ -837,10 +837,10 @@ public class RemoteLogManager implements Closeable {
return false;
}
- boolean isSegmentDeleted = deleteRemoteLogSegment(metadata, x
-> {
+ boolean isSegmentDeleted = deleteRemoteLogSegment(metadata,
ignored -> {
// Assumption that segments contain size >= 0
if (remainingBreachedSize > 0) {
- long remainingBytes = remainingBreachedSize -
x.segmentSizeInBytes();
+ long remainingBytes = remainingBreachedSize -
metadata.segmentSizeInBytes();
if (remainingBytes >= 0) {
remainingBreachedSize = remainingBytes;
return true;
@@ -864,7 +864,7 @@ public class RemoteLogManager implements Closeable {
}
boolean isSegmentDeleted = deleteRemoteLogSegment(metadata,
- x -> x.maxTimestampMs() <=
retentionTimeData.get().cleanupUntilMs);
+ ignored -> metadata.maxTimestampMs() <=
retentionTimeData.get().cleanupUntilMs);
if (isSegmentDeleted) {
remainingBreachedSize = Math.max(0, remainingBreachedSize
- metadata.segmentSizeInBytes());
// It is fine to have logStartOffset as
`metadata.endOffset() + 1` as the segment offset intervals
@@ -876,27 +876,40 @@ public class RemoteLogManager implements Closeable {
return isSegmentDeleted;
}
- private boolean
deleteLogStartOffsetBreachedSegments(RemoteLogSegmentMetadata metadata, long
startOffset)
+ private boolean
deleteLogStartOffsetBreachedSegments(RemoteLogSegmentMetadata metadata,
+ long
logStartOffset,
+
NavigableMap<Integer, Long> leaderEpochEntries)
throws RemoteStorageException, ExecutionException,
InterruptedException {
- boolean isSegmentDeleted = deleteRemoteLogSegment(metadata, x
-> startOffset > x.endOffset());
- if (isSegmentDeleted && retentionSizeData.isPresent()) {
- remainingBreachedSize = Math.max(0, remainingBreachedSize
- metadata.segmentSizeInBytes());
- logger.info("Deleted remote log segment {} due to log
start offset {} breach", metadata.remoteLogSegmentId(), startOffset);
+ boolean isSegmentDeleted = deleteRemoteLogSegment(metadata,
ignored -> {
+ if (!leaderEpochEntries.isEmpty()) {
+ // Note that `logStartOffset` and
`leaderEpochEntries.firstEntry().getValue()` should be same
+ Integer firstEpoch = leaderEpochEntries.firstKey();
+ return
metadata.segmentLeaderEpochs().keySet().stream().allMatch(epoch -> epoch <=
firstEpoch)
+ && metadata.endOffset() < logStartOffset;
+ }
+ return false;
+ });
+ if (isSegmentDeleted) {
+ logger.info("Deleted remote log segment {} due to
log-start-offset {} breach. " +
+ "Current earliest-epoch-entry: {},
segment-end-offset: {} and segment-epochs: {}",
+ metadata.remoteLogSegmentId(), logStartOffset,
leaderEpochEntries.firstEntry(),
+ metadata.endOffset(),
metadata.segmentLeaderEpochs());
}
-
return isSegmentDeleted;
}
// It removes the segments beyond the current leader's earliest
epoch. Those segments are considered as
// unreferenced because they are not part of the current leader
epoch lineage.
- private boolean
deleteLogSegmentsDueToLeaderEpochCacheTruncation(EpochEntry earliestEpochEntry,
RemoteLogSegmentMetadata metadata) throws RemoteStorageException,
ExecutionException, InterruptedException {
- boolean isSegmentDeleted = deleteRemoteLogSegment(metadata, x
->
-
x.segmentLeaderEpochs().keySet().stream().allMatch(epoch -> epoch <
earliestEpochEntry.epoch));
+ private boolean
deleteLogSegmentsDueToLeaderEpochCacheTruncation(EpochEntry earliestEpochEntry,
+
RemoteLogSegmentMetadata metadata)
+ throws RemoteStorageException, ExecutionException,
InterruptedException {
+ boolean isSegmentDeleted = deleteRemoteLogSegment(metadata,
ignored ->
+
metadata.segmentLeaderEpochs().keySet().stream().allMatch(epoch -> epoch <
earliestEpochEntry.epoch));
if (isSegmentDeleted) {
- logger.info("Deleted remote log segment {} due to leader
epoch cache truncation. Current earliest epoch: {}, segmentEndOffset: {} and
segmentEpochs: {}",
+ logger.info("Deleted remote log segment {} due to
leader-epoch-cache truncation. " +
+ "Current earliest-epoch-entry: {},
segment-end-offset: {} and segment-epochs: {}",
metadata.remoteLogSegmentId(), earliestEpochEntry,
metadata.endOffset(), metadata.segmentLeaderEpochs().keySet());
}
-
// No need to update the log-start-offset as these
epochs/offsets are earlier to that value.
return isSegmentDeleted;
}
@@ -904,7 +917,7 @@ public class RemoteLogManager implements Closeable {
private boolean deleteRemoteLogSegment(RemoteLogSegmentMetadata
segmentMetadata, Predicate<RemoteLogSegmentMetadata> predicate)
throws RemoteStorageException, ExecutionException,
InterruptedException {
if (predicate.test(segmentMetadata)) {
- logger.info("Deleting remote log segment {}",
segmentMetadata.remoteLogSegmentId());
+ logger.debug("Deleting remote log segment {}",
segmentMetadata.remoteLogSegmentId());
// Publish delete segment started event.
remoteLogMetadataManager.updateRemoteLogSegmentMetadata(
new
RemoteLogSegmentMetadataUpdate(segmentMetadata.remoteLogSegmentId(),
time.milliseconds(),
@@ -917,10 +930,9 @@ public class RemoteLogManager implements Closeable {
remoteLogMetadataManager.updateRemoteLogSegmentMetadata(
new
RemoteLogSegmentMetadataUpdate(segmentMetadata.remoteLogSegmentId(),
time.milliseconds(),
segmentMetadata.customMetadata(),
RemoteLogSegmentState.DELETE_SEGMENT_FINISHED, brokerId)).get();
- logger.info("Deleted remote log segment {}",
segmentMetadata.remoteLogSegmentId());
+ logger.debug("Deleted remote log segment {}",
segmentMetadata.remoteLogSegmentId());
return true;
}
-
return false;
}
@@ -967,7 +979,6 @@ public class RemoteLogManager implements Closeable {
LeaderEpochFileCache leaderEpochCache =
leaderEpochCacheOption.get();
// Build the leader epoch map by filtering the epochs that do not
have any records.
NavigableMap<Integer, Long> epochWithOffsets =
buildFilteredLeaderEpochMap(leaderEpochCache.epochWithOffsets());
- Optional<EpochEntry> earliestEpochEntryOptional =
leaderEpochCache.earliestEntry();
long logStartOffset = log.logStartOffset();
long logEndOffset = log.logEndOffset();
@@ -977,24 +988,35 @@ public class RemoteLogManager implements Closeable {
RemoteLogRetentionHandler remoteLogRetentionHandler = new
RemoteLogRetentionHandler(retentionSizeData, retentionTimeData);
Iterator<Integer> epochIterator =
epochWithOffsets.navigableKeySet().iterator();
- boolean isSegmentDeleted = true;
- while (isSegmentDeleted && epochIterator.hasNext()) {
+ boolean canProcess = true;
+ while (canProcess && epochIterator.hasNext()) {
Integer epoch = epochIterator.next();
Iterator<RemoteLogSegmentMetadata> segmentsIterator =
remoteLogMetadataManager.listRemoteLogSegments(topicIdPartition, epoch);
- while (isSegmentDeleted && segmentsIterator.hasNext()) {
+ while (canProcess && segmentsIterator.hasNext()) {
if (isCancelled() || !isLeader()) {
logger.info("Returning from remote log segments
cleanup for the remaining segments as the task state is changed.");
return;
}
RemoteLogSegmentMetadata metadata =
segmentsIterator.next();
- // check whether the segment contains the required epoch
range with in the current leader epoch lineage.
- if (isRemoteSegmentWithinLeaderEpochs(metadata,
logEndOffset, epochWithOffsets)) {
- isSegmentDeleted =
-
remoteLogRetentionHandler.deleteRetentionTimeBreachedSegments(metadata) ||
-
remoteLogRetentionHandler.deleteRetentionSizeBreachedSegments(metadata) ||
-
remoteLogRetentionHandler.deleteLogStartOffsetBreachedSegments(metadata,
logStartOffset);
+ // When the log-start-offset is moved by the user, the
leader-epoch-checkpoint file gets truncated
+ // as per the log-start-offset. Until the
rlm-cleaner-thread runs in the next iteration, those
+ // remote log segments won't be removed. The
`isRemoteSegmentWithinLeaderEpoch` validates whether
+ // the epochs present in the segment lies in the
checkpoint file. It will always return false
+ // since the checkpoint file was already truncated.
+ boolean isSegmentDeleted =
remoteLogRetentionHandler.deleteLogStartOffsetBreachedSegments(
+ metadata, logStartOffset, epochWithOffsets);
+ boolean isValidSegment = false;
+ if (!isSegmentDeleted) {
+ // check whether the segment contains the required
epoch range with in the current leader epoch lineage.
+ isValidSegment =
isRemoteSegmentWithinLeaderEpochs(metadata, logEndOffset, epochWithOffsets);
+ if (isValidSegment) {
+ isSegmentDeleted =
+
remoteLogRetentionHandler.deleteRetentionTimeBreachedSegments(metadata) ||
+
remoteLogRetentionHandler.deleteRetentionSizeBreachedSegments(metadata);
+ }
}
+ canProcess = isSegmentDeleted || !isValidSegment;
}
}
@@ -1002,9 +1024,12 @@ public class RemoteLogManager implements Closeable {
// to the leader. This will remove the unreferenced segments in
the remote storage. This is needed for
// unclean leader election scenarios as the remote storage can
have epochs earlier to the current leader's
// earliest leader epoch.
+ Optional<EpochEntry> earliestEpochEntryOptional =
leaderEpochCache.earliestEntry();
if (earliestEpochEntryOptional.isPresent()) {
EpochEntry earliestEpochEntry =
earliestEpochEntryOptional.get();
- Iterator<Integer> epochsToClean =
remoteLeaderEpochs.stream().filter(x -> x <
earliestEpochEntry.epoch).iterator();
+ Iterator<Integer> epochsToClean = remoteLeaderEpochs.stream()
+ .filter(remoteEpoch -> remoteEpoch <
earliestEpochEntry.epoch)
+ .iterator();
while (epochsToClean.hasNext()) {
int epoch = epochsToClean.next();
Iterator<RemoteLogSegmentMetadata> segmentsToBeCleaned =
remoteLogMetadataManager.listRemoteLogSegments(topicIdPartition, epoch);
@@ -1093,8 +1118,9 @@ public class RemoteLogManager implements Closeable {
Integer segmentFirstEpoch = segmentLeaderEpochs.firstKey();
Integer segmentLastEpoch = segmentLeaderEpochs.lastKey();
if (segmentFirstEpoch < leaderEpochs.firstKey() || segmentLastEpoch >
leaderEpochs.lastKey()) {
- LOGGER.debug("[{}] Remote segment {} is not within the partition
leader epoch lineage. Remote segment epochs: {} and partition leader epochs:
{}",
- segmentMetadata.topicIdPartition(),
segmentMetadata.remoteLogSegmentId(), segmentLeaderEpochs, leaderEpochs);
+ LOGGER.debug("Segment {} is not within the partition leader epoch
lineage. " +
+ "Remote segment epochs: {} and partition leader
epochs: {}",
+ segmentMetadata.remoteLogSegmentId(), segmentLeaderEpochs,
leaderEpochs);
return false;
}
@@ -1104,15 +1130,16 @@ public class RemoteLogManager implements Closeable {
// 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("[{}] Remote segment {}'s epoch {} is not within
the leader epoch lineage. Remote segment epochs: {} and partition leader
epochs: {}",
- segmentMetadata.topicIdPartition(),
segmentMetadata.remoteLogSegmentId(), epoch, segmentLeaderEpochs, leaderEpochs);
+ LOGGER.debug("Segment {} epoch {} is not within the leader
epoch lineage. " +
+ "Remote segment epochs: {} and partition
leader epochs: {}",
+ 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("[{}] Remote segment {}'s first epoch {}'s
offset is less than leader epoch's offset {}.",
- segmentMetadata.topicIdPartition(),
segmentMetadata.remoteLogSegmentId(), epoch, leaderEpochs.get(epoch));
+ LOGGER.debug("Segment {} first epoch {} offset is less than
leader epoch offset {}.",
+ segmentMetadata.remoteLogSegmentId(), epoch,
leaderEpochs.get(epoch));
return false;
}
@@ -1120,8 +1147,8 @@ public class RemoteLogManager implements Closeable {
if (epoch == segmentLastEpoch) {
Map.Entry<Integer, Long> nextEntry =
leaderEpochs.higherEntry(epoch);
if (nextEntry != null && segmentEndOffset >
nextEntry.getValue() - 1) {
- LOGGER.debug("[{}] Remote segment {}'s end offset {} is
more than leader epoch's offset {}.",
- segmentMetadata.topicIdPartition(),
segmentMetadata.remoteLogSegmentId(), segmentEndOffset, nextEntry.getValue() -
1);
+ LOGGER.debug("Segment {} end offset {} is more than leader
epoch offset {}.",
+ segmentMetadata.remoteLogSegmentId(),
segmentEndOffset, nextEntry.getValue() - 1);
return false;
}
}
@@ -1129,13 +1156,12 @@ public class RemoteLogManager implements Closeable {
// 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)))
{
- LOGGER.debug("[{}] Remote segment {}'s epoch {} is not within
the leader epoch lineage. Remote segment epochs: {} and partition leader
epochs: {}",
- segmentMetadata.topicIdPartition(),
segmentMetadata.remoteLogSegmentId(), epoch, segmentLeaderEpochs, leaderEpochs);
+ LOGGER.debug("Segment {} epoch {} is not within the leader
epoch lineage. " +
+ "Remote segment epochs: {} and partition
leader epochs: {}",
+ segmentMetadata.remoteLogSegmentId(), epoch,
segmentLeaderEpochs, leaderEpochs);
return false;
}
-
}
-
// segment end offset should be with in the log end offset.
return segmentEndOffset < logEndOffset;
}
@@ -1286,7 +1312,7 @@ public class RemoteLogManager implements Closeable {
OffsetIndex offsetIndex =
indexCache.getIndexEntry(segmentMetadata).offsetIndex();
long upperBoundOffset =
offsetIndex.fetchUpperBoundOffset(startOffsetPosition, fetchSize)
- .map(x -> x.offset).orElse(segmentMetadata.endOffset() + 1);
+ .map(position ->
position.offset).orElse(segmentMetadata.endOffset() + 1);
final Set<FetchResponseData.AbortedTransaction> abortedTransactions =
new HashSet<>();
diff --git
a/storage/src/test/java/org/apache/kafka/tiered/storage/TieredStorageTestBuilder.java
b/storage/src/test/java/org/apache/kafka/tiered/storage/TieredStorageTestBuilder.java
index ff489002b38..5c28c33183f 100644
---
a/storage/src/test/java/org/apache/kafka/tiered/storage/TieredStorageTestBuilder.java
+++
b/storage/src/test/java/org/apache/kafka/tiered/storage/TieredStorageTestBuilder.java
@@ -323,7 +323,7 @@ public final class TieredStorageTestBuilder {
Integer partition,
Long beforeOffset) {
TopicPartition topicPartition = new TopicPartition(topic, partition);
- actions.add(new DeleteRecordsAction(topicPartition, beforeOffset));
+ actions.add(new DeleteRecordsAction(topicPartition, beforeOffset,
buildDeleteSegmentSpecList(topic)));
return this;
}
@@ -377,6 +377,10 @@ public final class TieredStorageTestBuilder {
private DeleteTopicAction buildDeleteTopicAction(String topic,
Boolean shouldDelete) {
+ return new DeleteTopicAction(topic, buildDeleteSegmentSpecList(topic),
shouldDelete);
+ }
+
+ private List<RemoteDeleteSegmentSpec> buildDeleteSegmentSpecList(String
topic) {
List<RemoteDeleteSegmentSpec> deleteSegmentSpecList =
deletables.entrySet()
.stream()
.filter(e -> e.getKey().topic().equals(topic))
@@ -389,7 +393,7 @@ public final class TieredStorageTestBuilder {
})
.collect(Collectors.toList());
deleteSegmentSpecList.forEach(spec ->
deletables.remove(spec.getTopicPartition()));
- return new DeleteTopicAction(topic, deleteSegmentSpecList,
shouldDelete);
+ return deleteSegmentSpecList;
}
}
diff --git
a/storage/src/test/java/org/apache/kafka/tiered/storage/actions/DeleteRecordsAction.java
b/storage/src/test/java/org/apache/kafka/tiered/storage/actions/DeleteRecordsAction.java
index 0f81d35a052..0f6a756c048 100644
---
a/storage/src/test/java/org/apache/kafka/tiered/storage/actions/DeleteRecordsAction.java
+++
b/storage/src/test/java/org/apache/kafka/tiered/storage/actions/DeleteRecordsAction.java
@@ -16,36 +16,71 @@
*/
package org.apache.kafka.tiered.storage.actions;
+import org.apache.kafka.server.log.remote.storage.LocalTieredStorage;
+import org.apache.kafka.server.log.remote.storage.LocalTieredStorageCondition;
import org.apache.kafka.tiered.storage.TieredStorageTestAction;
import org.apache.kafka.tiered.storage.TieredStorageTestContext;
import org.apache.kafka.clients.admin.RecordsToDelete;
import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.tiered.storage.specs.RemoteDeleteSegmentSpec;
import java.io.PrintStream;
import java.util.Collections;
+import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.stream.Collectors;
+
+import static
org.apache.kafka.server.log.remote.storage.LocalTieredStorageCondition.expectEvent;
+import static
org.apache.kafka.server.log.remote.storage.LocalTieredStorageEvent.EventType.DELETE_SEGMENT;
public final class DeleteRecordsAction implements TieredStorageTestAction {
+ private static final int DELETE_WAIT_TIMEOUT_SEC = 10;
private final TopicPartition partition;
private final Long beforeOffset;
+ private final List<RemoteDeleteSegmentSpec> deleteSegmentSpecs;
public DeleteRecordsAction(TopicPartition partition,
- Long beforeOffset) {
+ Long beforeOffset,
+ List<RemoteDeleteSegmentSpec>
deleteSegmentSpecs) {
this.partition = partition;
this.beforeOffset = beforeOffset;
+ this.deleteSegmentSpecs = deleteSegmentSpecs;
}
@Override
- public void doExecute(TieredStorageTestContext context) throws
InterruptedException, ExecutionException {
+ public void doExecute(TieredStorageTestContext context)
+ throws InterruptedException, ExecutionException, TimeoutException {
+ List<LocalTieredStorage> tieredStorages =
context.remoteStorageManagers();
+ List<LocalTieredStorageCondition> tieredStorageConditions =
deleteSegmentSpecs.stream()
+ .filter(spec -> spec.getEventType() == DELETE_SEGMENT)
+ .map(spec -> expectEvent(
+ tieredStorages,
+ spec.getEventType(),
+ spec.getSourceBrokerId(),
+ spec.getTopicPartition(),
+ false,
+ spec.getEventCount()))
+ .collect(Collectors.toList());
+
Map<TopicPartition, RecordsToDelete> recordsToDeleteMap =
Collections.singletonMap(partition,
RecordsToDelete.beforeOffset(beforeOffset));
context.admin().deleteRecords(recordsToDeleteMap).all().get();
+
+ if (!tieredStorageConditions.isEmpty()) {
+ tieredStorageConditions.stream()
+ .reduce(LocalTieredStorageCondition::and)
+ .get()
+ .waitUntilTrue(DELETE_WAIT_TIMEOUT_SEC, TimeUnit.SECONDS);
+ }
}
@Override
public void describe(PrintStream output) {
output.printf("delete-records partition: %s, before-offset: %d%n",
partition, beforeOffset);
+ deleteSegmentSpecs.forEach(spec -> output.println(" " + spec));
}
}
diff --git
a/storage/src/test/java/org/apache/kafka/tiered/storage/integration/DeleteSegmentsDueToLogStartOffsetBreachTest.java
b/storage/src/test/java/org/apache/kafka/tiered/storage/integration/DeleteSegmentsDueToLogStartOffsetBreachTest.java
new file mode 100644
index 00000000000..5f10df5c68c
--- /dev/null
+++
b/storage/src/test/java/org/apache/kafka/tiered/storage/integration/DeleteSegmentsDueToLogStartOffsetBreachTest.java
@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.tiered.storage.integration;
+
+import org.apache.kafka.tiered.storage.TieredStorageTestBuilder;
+import org.apache.kafka.tiered.storage.TieredStorageTestHarness;
+import org.apache.kafka.tiered.storage.specs.KeyValueSpec;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+
+import static
org.apache.kafka.server.log.remote.storage.LocalTieredStorageEvent.EventType.DELETE_SEGMENT;
+import static org.apache.kafka.common.utils.Utils.mkEntry;
+import static org.apache.kafka.common.utils.Utils.mkMap;
+
+public final class DeleteSegmentsDueToLogStartOffsetBreachTest extends
TieredStorageTestHarness {
+
+ @Override
+ public int brokerCount() {
+ return 2;
+ }
+
+ @Override
+ protected void writeTestSpecifications(TieredStorageTestBuilder builder) {
+ final Integer broker0 = 0;
+ final Integer broker1 = 1;
+ final String topicA = "topicA";
+ final Integer p0 = 0;
+ final Integer partitionCount = 1;
+ final Integer replicationFactor = 2;
+ final Integer maxBatchCountPerSegment = 2;
+ final Map<Integer, List<Integer>> replicaAssignment =
mkMap(mkEntry(p0, Arrays.asList(broker0, broker1)));
+ final boolean enableRemoteLogStorage = true;
+ final int beginEpoch = 0;
+ final long startOffset = 3;
+ final long beforeOffset = 3L;
+ final long beforeOffset1 = 7L;
+
+ // Create topicA with 1 partition and 2 RF
+ builder.createTopic(topicA, partitionCount, replicationFactor,
maxBatchCountPerSegment, replicaAssignment,
+ enableRemoteLogStorage)
+ // produce events to partition 0 and expect 2 segments to be
offloaded
+ .expectSegmentToBeOffloaded(broker0, topicA, p0, 0, new
KeyValueSpec("k0", "v0"),
+ new KeyValueSpec("k1", "v1"))
+ .expectSegmentToBeOffloaded(broker0, topicA, p0, 2, new
KeyValueSpec("k2", "v2"),
+ new KeyValueSpec("k3", "v3"))
+ .expectEarliestLocalOffsetInLogDirectory(topicA, p0, 4L)
+ .produce(topicA, p0, new KeyValueSpec("k0", "v0"), new
KeyValueSpec("k1", "v1"),
+ new KeyValueSpec("k2", "v2"), new KeyValueSpec("k3",
"v3"), new KeyValueSpec("k4", "v4"))
+ // Use DELETE_RECORDS API to delete the records upto offset 3
and expect one remote segment to be deleted
+ .expectDeletionInRemoteStorage(broker0, topicA, p0,
DELETE_SEGMENT, 1)
+ .deleteRecords(topicA, p0, beforeOffset)
+ // expect that the leader epoch checkpoint is updated
+ // Comment out this line if it's FLAKY since the leader-epoch
is not deterministic in ZK mode.
+ .expectLeaderEpochCheckpoint(broker0, topicA, p0, beginEpoch,
startOffset)
+ // consume from the offset-3 of the topic to read data from
local and remote storage
+ .expectFetchFromTieredStorage(broker0, topicA, p0, 1)
+ .consume(topicA, p0, 3L, 2, 1)
+
+ // switch leader to change the leader-epoch from 0 to 1
+ .expectLeader(topicA, p0, broker1, true)
+ // produce some more messages and move the log-start-offset
such that earliest-epoch changes from 0 to 1
+ .expectSegmentToBeOffloaded(broker1, topicA, p0, 4, new
KeyValueSpec("k4", "v4"),
+ new KeyValueSpec("k5", "v5"))
+ .expectSegmentToBeOffloaded(broker1, topicA, p0, 6, new
KeyValueSpec("k6", "v6"),
+ new KeyValueSpec("k7", "v7"))
+ .expectEarliestLocalOffsetInLogDirectory(topicA, p0, 8L)
+ .produce(topicA, p0, new KeyValueSpec("k5", "v5"), new
KeyValueSpec("k6", "v6"),
+ new KeyValueSpec("k7", "v7"), new KeyValueSpec("k8",
"v8"), new KeyValueSpec("k9", "v9"))
+ // Use DELETE_RECORDS API to delete the records upto offset 7
and expect 2 remote segments to be deleted
+ .expectDeletionInRemoteStorage(broker1, topicA, p0,
DELETE_SEGMENT, 2)
+ .deleteRecords(topicA, p0, beforeOffset1)
+ // consume from the topic with fetch-offset 7 to read data
from local and remote storage
+ .expectFetchFromTieredStorage(broker1, topicA, p0, 1)
+ .consume(topicA, p0, 7L, 3, 1);
+ }
+}
