diff --git a/.github/workflows/pr.yml b/.github/workflows/pr.yml
new file mode 100644
index 0000000000000..895d415ee97a7
--- /dev/null
+++ b/.github/workflows/pr.yml
@@ -0,0 +1,25 @@
+# The workflow to check pull requests into main.
+# This checks the source in the state as if after the merge.
+name: Pull request checks
+on:
+ pull_request:
+ branches: [ '**' ]
+jobs:
+ build:
+ strategy:
+ matrix:
+ java-version: [ 11, 17 ]
+ runs-on: [ ubuntu-latest ]
+ name: Build on ${{ matrix.runs-on }} with jdk ${{ matrix.java-version }}
+ runs-on: ${{ matrix.runs-on }}
+ steps:
+ - name: Checkout code
+ uses: actions/checkout@v3
+ - name: Set up JDK ${{ matrix.java-version }}
+ uses: actions/setup-java@v3
+ with:
+ java-version: ${{ matrix.java-version }}
+ distribution: temurin
+
+ - name: Build Docker image
+ run: make BRANCH=${GITHUB_HEAD_REF} docker_image
diff --git a/Dockerfile b/Dockerfile
new file mode 100644
index 0000000000000..5bf5d969112a7
--- /dev/null
+++ b/Dockerfile
@@ -0,0 +1,38 @@
+##
+# Copyright 2023 Aiven Oy
+#
+# Licensed 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.
+##
+# Kafka 3.4.x
+FROM confluentinc/cp-kafka:7.4.0
+
+ARG _SCALA_VERSION
+ARG _KAFKA_VERSION
+ENV _KAFKA_FULL_VERSION "kafka_${_SCALA_VERSION}-${_KAFKA_VERSION}"
+
+USER root
+COPY core/build/distributions/${_KAFKA_FULL_VERSION}.tgz /
+RUN cd / \
+ && tar -xf ${_KAFKA_FULL_VERSION}.tgz \
+ && rm -r /usr/share/java/kafka/* \
+ && cp /${_KAFKA_FULL_VERSION}/libs/* /usr/share/java/kafka/ \
+ && ln -s /usr/share/java/kafka/${_KAFKA_FULL_VERSION}.jar /usr/share/java/kafka/kafka.jar \
+ && rm -r /${_KAFKA_FULL_VERSION}.tgz /${_KAFKA_FULL_VERSION}
+
+# Add test jars with local implementations.
+COPY clients/build/libs/kafka-clients-${_KAFKA_VERSION}-test.jar /usr/share/java/kafka/
+COPY storage/build/libs/kafka-storage-${_KAFKA_VERSION}-test.jar /usr/share/java/kafka/
+COPY storage/api/build/libs/kafka-storage-api-${_KAFKA_VERSION}-test.jar /usr/share/java/kafka/
+
+# Restore the user.
+USER appuser
diff --git a/Makefile b/Makefile
new file mode 100644
index 0000000000000..a3cae94dc0a23
--- /dev/null
+++ b/Makefile
@@ -0,0 +1,39 @@
+##
+# Copyright 2023 Aiven Oy
+#
+# Licensed 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.
+##
+SCALA_VERSION=2.13
+KAFKA_VERSION=3.6.0-SNAPSHOT
+IMAGE_NAME=aivenoy/kafka
+
+BRANCH := $(shell git rev-parse --abbrev-ref HEAD)
+IMAGE_TAG := $(subst /,_,$(BRANCH))
+
+.PHONY: clean
+clean:
+ ./gradlew clean
+
+core/build/distributions/kafka_$(SCALA_VERSION)-$(KAFKA_VERSION).tgz:
+ ./gradlew -PscalaVersion=$(SCALA_VERSION) testJar releaseTarGz
+
+.PHONY: docker_image
+docker_image: core/build/distributions/kafka_$(SCALA_VERSION)-$(KAFKA_VERSION).tgz
+ docker build . \
+ --build-arg _SCALA_VERSION=$(SCALA_VERSION) \
+ --build-arg _KAFKA_VERSION=$(KAFKA_VERSION) \
+ -t $(IMAGE_NAME):$(IMAGE_TAG)
+
+.PHONY: docker_push
+docker_push:
+ docker push $(IMAGE_NAME):$(IMAGE_TAG)
diff --git a/checkstyle/suppressions.xml b/checkstyle/suppressions.xml
index 124b028eb41c3..2ded53387107a 100644
--- a/checkstyle/suppressions.xml
+++ b/checkstyle/suppressions.xml
@@ -336,9 +336,9 @@
+ files="(LogValidator|RemoteLogManagerConfig|RemoteLogManager).java"/>
+ files="(LogValidator|RemoteLogManager|RemoteIndexCache).java"/>
diff --git a/core/src/main/java/kafka/log/remote/RemoteLogManager.java b/core/src/main/java/kafka/log/remote/RemoteLogManager.java
index a9372a80fa0b3..493926ae6b4e5 100644
--- a/core/src/main/java/kafka/log/remote/RemoteLogManager.java
+++ b/core/src/main/java/kafka/log/remote/RemoteLogManager.java
@@ -92,10 +92,12 @@
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.NavigableMap;
import java.util.Optional;
import java.util.OptionalInt;
import java.util.OptionalLong;
import java.util.Set;
+import java.util.TreeMap;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@@ -110,6 +112,7 @@
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.Function;
+import java.util.function.Predicate;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -121,7 +124,8 @@
* - initializing `RemoteStorageManager` and `RemoteLogMetadataManager` instances
* - receives any leader and follower replica events and partition stop events and act on them
* - also provides APIs to fetch indexes, metadata about remote log segments
- * - copying log segments to remote storage
+ * - copying log segments to the remote storage
+ * - cleaning up segments that are expired based on retention size or retention time
*/
public class RemoteLogManager implements Closeable {
@@ -132,6 +136,7 @@ public class RemoteLogManager implements Closeable {
private final String logDir;
private final Time time;
private final Function> fetchLog;
+ private final BiConsumer updateRemoteLogStartOffset;
private final BrokerTopicStats brokerTopicStats;
private final RemoteStorageManager remoteLogStorageManager;
@@ -165,6 +170,8 @@ public class RemoteLogManager implements Closeable {
* @param time Time instance.
* @param clusterId The cluster id.
* @param fetchLog function to get UnifiedLog instance for a given topic.
+ * @param updateRemoteLogStartOffset function to update the log-start-offset for a given topic partition.
+ * @param brokerTopicStats BrokerTopicStats instance to update the respective metrics.
*/
public RemoteLogManager(RemoteLogManagerConfig rlmConfig,
int brokerId,
@@ -172,6 +179,7 @@ public RemoteLogManager(RemoteLogManagerConfig rlmConfig,
String clusterId,
Time time,
Function> fetchLog,
+ BiConsumer updateRemoteLogStartOffset,
BrokerTopicStats brokerTopicStats) throws IOException {
this.rlmConfig = rlmConfig;
this.brokerId = brokerId;
@@ -179,6 +187,7 @@ public RemoteLogManager(RemoteLogManagerConfig rlmConfig,
this.clusterId = clusterId;
this.time = time;
this.fetchLog = fetchLog;
+ this.updateRemoteLogStartOffset = updateRemoteLogStartOffset;
this.brokerTopicStats = brokerTopicStats;
remoteLogStorageManager = createRemoteStorageManager();
@@ -298,11 +307,6 @@ private void cacheTopicPartitionIds(TopicIdPartition topicIdPartition) {
}
}
- // for testing
- public RLMScheduledThreadPool rlmScheduledThreadPool() {
- return rlmScheduledThreadPool;
- }
-
/**
* Callback to receive any leadership changes for the topic partitions assigned to this broker. If there are no
* existing tasks for a given topic partition then it will assign new leader or follower task else it will convert the
@@ -409,12 +413,18 @@ private void deleteRemoteLogPartition(TopicIdPartition partition) throws RemoteS
publishEvents(deleteSegmentFinishedEvents).get();
}
- private CompletableFuture publishEvents(List events) throws RemoteStorageException {
- List> result = new ArrayList<>();
+ private CompletableFuture publishEvents(List events) {
+ CompletableFuture result = CompletableFuture.completedFuture(null);
for (RemoteLogSegmentMetadataUpdate event : events) {
- result.add(remoteLogMetadataManager.updateRemoteLogSegmentMetadata(event));
+ result = result.thenAcceptAsync(unused -> {
+ try {
+ remoteLogMetadataManager.updateRemoteLogSegmentMetadata(event);
+ } catch (RemoteStorageException e) {
+ throw new KafkaException(e);
+ }
+ });
}
- return CompletableFuture.allOf(result.toArray(new CompletableFuture[0]));
+ return result;
}
public Optional fetchRemoteLogSegmentMetadata(TopicPartition topicPartition,
@@ -486,15 +496,25 @@ public Optional findOffsetByTimestamp(TopicParti
throw new KafkaException("Topic id does not exist for topic partition: " + tp);
}
+ Optional unifiedLogOptional = fetchLog.apply(tp);
+ if (!unifiedLogOptional.isPresent()) {
+ throw new KafkaException("UnifiedLog does not exist for topic partition: " + tp);
+ }
+
+ UnifiedLog unifiedLog = unifiedLogOptional.get();
+
// Get the respective epoch in which the starting-offset exists.
OptionalInt maybeEpoch = leaderEpochCache.epochForOffset(startingOffset);
+ TopicIdPartition topicIdPartition = new TopicIdPartition(topicId, tp);
+ NavigableMap epochWithOffsets = buildFilteredLeaderEpochMap(leaderEpochCache.epochWithOffsets());
while (maybeEpoch.isPresent()) {
int epoch = maybeEpoch.getAsInt();
- Iterator iterator = remoteLogMetadataManager.listRemoteLogSegments(new TopicIdPartition(topicId, tp), epoch);
+ Iterator iterator = remoteLogMetadataManager.listRemoteLogSegments(topicIdPartition, epoch);
while (iterator.hasNext()) {
RemoteLogSegmentMetadata rlsMetadata = iterator.next();
- if (rlsMetadata.maxTimestampMs() >= timestamp && rlsMetadata.endOffset() >= startingOffset) {
+ if (rlsMetadata.maxTimestampMs() >= timestamp && rlsMetadata.endOffset() >= startingOffset &&
+ isRemoteSegmentWithinLeaderEpochs(rlsMetadata, unifiedLog.logEndOffset(), epochWithOffsets)) {
return lookupTimestamp(rlsMetadata, timestamp, startingOffset);
}
}
@@ -725,6 +745,8 @@ private void copyLogSegment(UnifiedLog log, LogSegment segment, long nextSegment
.remoteCopyBytesRate().mark(copySegmentStartedRlsm.segmentSizeInBytes());
brokerTopicStats.allTopicsStats().remoteCopyBytesRate().mark(copySegmentStartedRlsm.segmentSizeInBytes());
copiedOffsetOption = OptionalLong.of(endOffset);
+ // Update the highest offset in remote storage for this partition's log so that the local log segments
+ // are not deleted before they are copied to remote storage.
log.updateHighestOffsetInRemoteStorage(endOffset);
logger.info("Copied {} to remote storage with segment-id: {}", logFileName, copySegmentFinishedRlsm.remoteLogSegmentId());
}
@@ -744,14 +766,21 @@ public void run() {
return;
}
+ UnifiedLog log = unifiedLogOptional.get();
if (isLeader()) {
// Copy log segments to remote storage
- copyLogSegmentsToRemote(unifiedLogOptional.get());
+ copyLogSegmentsToRemote(log);
+ // Cleanup/delete expired remote log segments
+ cleanupExpiredRemoteLogSegments();
+ } else {
+ long offset = findHighestRemoteOffset(topicIdPartition, log);
+ // Update the highest offset in remote storage for this partition's log so that the local log segments
+ // are not deleted before they are copied to remote storage.
+ log.updateHighestOffsetInRemoteStorage(offset);
}
} catch (InterruptedException ex) {
if (!isCancelled()) {
- logger.warn("Current thread for topic-partition-id {} is interrupted, this task won't be rescheduled. " +
- "Reason: {}", topicIdPartition, ex.getMessage());
+ logger.warn("Current thread for topic-partition-id {} is interrupted. Reason: {}", topicIdPartition, ex.getMessage());
}
} catch (Exception ex) {
if (!isCancelled()) {
@@ -761,11 +790,385 @@ public void run() {
}
}
+ public void handleLogStartOffsetUpdate(TopicPartition topicPartition, long remoteLogStartOffset) {
+ if (isLeader()) {
+ logger.debug("Updating {} with remoteLogStartOffset: {}", topicPartition, remoteLogStartOffset);
+ updateRemoteLogStartOffset.accept(topicPartition, remoteLogStartOffset);
+ }
+ }
+
+ class RemoteLogRetentionHandler {
+
+ private final Optional retentionSizeData;
+ private final Optional retentionTimeData;
+
+ private long remainingBreachedSize;
+
+ private OptionalLong logStartOffset = OptionalLong.empty();
+
+ public RemoteLogRetentionHandler(Optional retentionSizeData, Optional retentionTimeData) {
+ this.retentionSizeData = retentionSizeData;
+ this.retentionTimeData = retentionTimeData;
+ remainingBreachedSize = retentionSizeData.map(sizeData -> sizeData.remainingBreachedSize).orElse(0L);
+ }
+
+ private boolean deleteRetentionSizeBreachedSegments(RemoteLogSegmentMetadata metadata) throws RemoteStorageException, ExecutionException, InterruptedException {
+ if (!retentionSizeData.isPresent()) {
+ return false;
+ }
+
+ boolean isSegmentDeleted = deleteRemoteLogSegment(metadata, x -> {
+ // Assumption that segments contain size >= 0
+ if (remainingBreachedSize > 0) {
+ long remainingBytes = remainingBreachedSize - x.segmentSizeInBytes();
+ if (remainingBytes >= 0) {
+ remainingBreachedSize = remainingBytes;
+ return true;
+ }
+ }
+
+ return false;
+ });
+ if (isSegmentDeleted) {
+ logStartOffset = OptionalLong.of(metadata.endOffset() + 1);
+ logger.info("Deleted remote log segment {} due to retention size {} breach. Log size after deletion will be {}.",
+ metadata.remoteLogSegmentId(), retentionSizeData.get().retentionSize, remainingBreachedSize + retentionSizeData.get().retentionSize);
+ }
+ return isSegmentDeleted;
+ }
+
+ public boolean deleteRetentionTimeBreachedSegments(RemoteLogSegmentMetadata metadata)
+ throws RemoteStorageException, ExecutionException, InterruptedException {
+ if (!retentionTimeData.isPresent()) {
+ return false;
+ }
+
+ boolean isSegmentDeleted = deleteRemoteLogSegment(metadata,
+ x -> x.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
+ // are ascending with in an epoch.
+ logStartOffset = OptionalLong.of(metadata.endOffset() + 1);
+ logger.info("Deleted remote log segment {} due to retention time {}ms breach based on the largest record timestamp in the segment",
+ metadata.remoteLogSegmentId(), retentionTimeData.get().retentionMs);
+ }
+ return isSegmentDeleted;
+ }
+
+ private boolean deleteLogStartOffsetBreachedSegments(RemoteLogSegmentMetadata metadata, long startOffset)
+ 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);
+ }
+
+ 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));
+ if (isSegmentDeleted) {
+ logger.info("Deleted remote log segment {} due to leader epoch cache truncation. Current earliest epoch: {}, segmentEndOffset: {} and segmentEpochs: {}",
+ 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;
+ }
+
+ private boolean deleteRemoteLogSegment(RemoteLogSegmentMetadata segmentMetadata, Predicate predicate)
+ throws RemoteStorageException, ExecutionException, InterruptedException {
+ if (predicate.test(segmentMetadata)) {
+ logger.info("Deleting remote log segment {}", segmentMetadata.remoteLogSegmentId());
+ // Publish delete segment started event.
+ remoteLogMetadataManager.updateRemoteLogSegmentMetadata(
+ new RemoteLogSegmentMetadataUpdate(segmentMetadata.remoteLogSegmentId(), time.milliseconds(),
+ segmentMetadata.customMetadata(), RemoteLogSegmentState.DELETE_SEGMENT_STARTED, brokerId)).get();
+
+ // Delete the segment in remote storage.
+ remoteLogStorageManager.deleteLogSegmentData(segmentMetadata);
+
+ // Publish delete segment finished event.
+ remoteLogMetadataManager.updateRemoteLogSegmentMetadata(
+ new RemoteLogSegmentMetadataUpdate(segmentMetadata.remoteLogSegmentId(), time.milliseconds(),
+ segmentMetadata.customMetadata(), RemoteLogSegmentState.DELETE_SEGMENT_FINISHED, brokerId)).get();
+ logger.info("Deleted remote log segment {}", segmentMetadata.remoteLogSegmentId());
+ return true;
+ }
+
+ return false;
+ }
+
+ }
+
+ private void cleanupExpiredRemoteLogSegments() throws RemoteStorageException, ExecutionException, InterruptedException {
+ if (isCancelled() || !isLeader()) {
+ logger.info("Returning from remote log segments cleanup as the task state is changed");
+ return;
+ }
+
+ // Cleanup remote log segments and update the log start offset if applicable.
+ final Iterator segmentMetadataIter = remoteLogMetadataManager.listRemoteLogSegments(topicIdPartition);
+ if (!segmentMetadataIter.hasNext()) {
+ logger.debug("No remote log segments available on remote storage for partition: {}", topicIdPartition);
+ return;
+ }
+
+ final Optional logOptional = fetchLog.apply(topicIdPartition.topicPartition());
+ if (!logOptional.isPresent()) {
+ logger.debug("No UnifiedLog instance available for partition: {}", topicIdPartition);
+ return;
+ }
+
+ final UnifiedLog log = logOptional.get();
+ final Option leaderEpochCacheOption = log.leaderEpochCache();
+ if (leaderEpochCacheOption.isEmpty()) {
+ logger.debug("No leader epoch cache available for partition: {}", topicIdPartition);
+ return;
+ }
+
+ final Set epochsSet = new HashSet<>();
+ // Good to have an API from RLMM to get all the remote leader epochs of all the segments of a partition
+ // instead of going through all the segments and building it here.
+ while (segmentMetadataIter.hasNext()) {
+ RemoteLogSegmentMetadata segmentMetadata = segmentMetadataIter.next();
+ epochsSet.addAll(segmentMetadata.segmentLeaderEpochs().keySet());
+ }
+
+ // All the leader epochs in sorted order that exists in remote storage
+ final List remoteLeaderEpochs = new ArrayList<>(epochsSet);
+ Collections.sort(remoteLeaderEpochs);
+
+ LeaderEpochFileCache leaderEpochCache = leaderEpochCacheOption.get();
+ // Build the leader epoch map by filtering the epochs that do not have any records.
+ NavigableMap epochWithOffsets = buildFilteredLeaderEpochMap(leaderEpochCache.epochWithOffsets());
+ Optional earliestEpochEntryOptional = leaderEpochCache.earliestEntry();
+
+ long logStartOffset = log.logStartOffset();
+ long logEndOffset = log.logEndOffset();
+ Optional retentionSizeData = buildRetentionSizeData(log.config().retentionSize,
+ log.onlyLocalLogSegmentsSize(), logEndOffset, epochWithOffsets);
+ Optional retentionTimeData = buildRetentionTimeData(log.config().retentionMs);
+
+ RemoteLogRetentionHandler remoteLogRetentionHandler = new RemoteLogRetentionHandler(retentionSizeData, retentionTimeData);
+ Iterator epochIterator = epochWithOffsets.navigableKeySet().iterator();
+ boolean isSegmentDeleted = true;
+ while (isSegmentDeleted && epochIterator.hasNext()) {
+ Integer epoch = epochIterator.next();
+ Iterator segmentsIterator = remoteLogMetadataManager.listRemoteLogSegments(topicIdPartition, epoch);
+ while (isSegmentDeleted && 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);
+ }
+ }
+ }
+
+ // Remove the remote log segments whose segment-leader-epochs are less than the earliest-epoch known
+ // 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.
+ if (earliestEpochEntryOptional.isPresent()) {
+ EpochEntry earliestEpochEntry = earliestEpochEntryOptional.get();
+ Iterator epochsToClean = remoteLeaderEpochs.stream().filter(x -> x < earliestEpochEntry.epoch).iterator();
+ while (epochsToClean.hasNext()) {
+ int epoch = epochsToClean.next();
+ Iterator segmentsToBeCleaned = remoteLogMetadataManager.listRemoteLogSegments(topicIdPartition, epoch);
+ while (segmentsToBeCleaned.hasNext()) {
+ if (isCancelled() || !isLeader()) {
+ return;
+ }
+ // No need to update the log-start-offset even though the segment is deleted as these epochs/offsets are earlier to that value.
+ remoteLogRetentionHandler.deleteLogSegmentsDueToLeaderEpochCacheTruncation(earliestEpochEntry, segmentsToBeCleaned.next());
+ }
+ }
+ }
+
+ // Update log start offset with the computed value after retention cleanup is done
+ remoteLogRetentionHandler.logStartOffset.ifPresent(offset -> handleLogStartOffsetUpdate(topicIdPartition.topicPartition(), offset));
+ }
+
+ private Optional buildRetentionTimeData(long retentionMs) {
+ return retentionMs > -1
+ ? Optional.of(new RetentionTimeData(retentionMs, time.milliseconds() - retentionMs))
+ : Optional.empty();
+ }
+
+ private Optional buildRetentionSizeData(long retentionSize,
+ long onlyLocalLogSegmentsSize,
+ long logEndOffset,
+ NavigableMap epochEntries) throws RemoteStorageException {
+ if (retentionSize > -1) {
+ long remoteLogSizeBytes = 0L;
+ Set visitedSegmentIds = new HashSet<>();
+ for (Integer epoch : epochEntries.navigableKeySet()) {
+ // remoteLogSize(topicIdPartition, epochEntry.epoch) may not be completely accurate as the remote
+ // log size may be computed for all the segments but not for segments with in the current
+ // partition's leader epoch lineage. Better to revisit this API.
+ // remoteLogSizeBytes += remoteLogMetadataManager.remoteLogSize(topicIdPartition, epochEntry.epoch);
+ Iterator segmentsIterator = remoteLogMetadataManager.listRemoteLogSegments(topicIdPartition, epoch);
+ while (segmentsIterator.hasNext()) {
+ RemoteLogSegmentMetadata segmentMetadata = segmentsIterator.next();
+ RemoteLogSegmentId segmentId = segmentMetadata.remoteLogSegmentId();
+ if (!visitedSegmentIds.contains(segmentId) && isRemoteSegmentWithinLeaderEpochs(segmentMetadata, logEndOffset, epochEntries)) {
+ remoteLogSizeBytes += segmentMetadata.segmentSizeInBytes();
+ visitedSegmentIds.add(segmentId);
+ }
+ }
+ }
+
+ // This is the total size of segments in local log that have their base-offset > local-log-start-offset
+ // and size of the segments in remote storage which have their end-offset < local-log-start-offset.
+ long totalSize = onlyLocalLogSegmentsSize + remoteLogSizeBytes;
+ if (totalSize > retentionSize) {
+ long remainingBreachedSize = totalSize - retentionSize;
+ RetentionSizeData retentionSizeData = new RetentionSizeData(retentionSize, remainingBreachedSize);
+ return Optional.of(retentionSizeData);
+ }
+ }
+
+ return Optional.empty();
+ }
+
public String toString() {
return this.getClass().toString() + "[" + topicIdPartition + "]";
}
}
+ /**
+ * Returns true if the remote segment's epoch/offsets are within the leader epoch lineage of the partition.
+ * The constraints here are as follows:
+ * - The segment's first epoch's offset should be more than or equal to the respective leader epoch's offset in the partition leader epoch lineage.
+ * - The segment's end offset should be less than or equal to the respective leader epoch's offset in the partition leader epoch lineage.
+ * - The segment's epoch lineage(epoch and offset) should be same as leader epoch lineage((epoch and offset)) except
+ * for the first and the last epochs in the segment.
+ *
+ * @param segmentMetadata The remote segment metadata to be validated.
+ * @param logEndOffset The log end offset of the partition.
+ * @param leaderEpochs The leader epoch lineage of the partition by filtering the epochs containing no data.
+ * @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 leaderEpochs) {
+ long segmentEndOffset = segmentMetadata.endOffset();
+ // Filter epochs that does not have any messages/records associated with them.
+ NavigableMap 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()) {
+ 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);
+ return false;
+ }
+
+ for (Map.Entry entry : segmentLeaderEpochs.entrySet()) {
+ int epoch = entry.getKey();
+ long offset = entry.getValue();
+
+ // 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);
+ 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));
+ return false;
+ }
+
+ // Segment's end offset should be less than or equal to the respective leader epoch's offset.
+ if (epoch == segmentLastEpoch) {
+ Map.Entry 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);
+ 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))) {
+ 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);
+ return false;
+ }
+
+ }
+
+ // segment end offset should be with in the log end offset.
+ return segmentEndOffset < logEndOffset;
+ }
+
+ /**
+ * 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:
+ *
+ * 0 - 0
+ * 1 - 10
+ * 2 - 20
+ * 3 - 30
+ * 4 - 40
+ * 5 - 60 // epoch 5 does not have records or messages associated with it
+ * 6 - 60
+ * 7 - 70
+ *
+ * When the above leaderEpochMap is passed to this method, it returns the following map:
+ *
+ * 0 - 0
+ * 1 - 10
+ * 2 - 20
+ * 3 - 30
+ * 4 - 40
+ * 6 - 60
+ * 7 - 70
+ *
+ * @param leaderEpochs The leader epoch map to be refined.
+ */
+ // Visible for testing
+ public static NavigableMap buildFilteredLeaderEpochMap(NavigableMap leaderEpochs) {
+ List duplicatedEpochs = new ArrayList<>();
+ Map.Entry previousEntry = null;
+ for (Map.Entry entry : leaderEpochs.entrySet()) {
+ if (previousEntry != null && previousEntry.getValue().equals(entry.getValue())) {
+ duplicatedEpochs.add(previousEntry.getKey());
+ }
+ previousEntry = entry;
+ }
+
+ if (duplicatedEpochs.isEmpty()) {
+ return leaderEpochs;
+ }
+
+ TreeMap filteredLeaderEpochs = new TreeMap<>(leaderEpochs);
+ for (Integer duplicatedEpoch : duplicatedEpochs) {
+ filteredLeaderEpochs.remove(duplicatedEpoch);
+ }
+ return filteredLeaderEpochs;
+ }
+
public FetchDataInfo read(RemoteStorageFetchInfo remoteStorageFetchInfo) throws RemoteStorageException, IOException {
int fetchMaxBytes = remoteStorageFetchInfo.fetchMaxBytes;
TopicPartition tp = remoteStorageFetchInfo.topicPartition;
@@ -887,7 +1290,7 @@ private void collectAbortedTransactions(long startOffset,
// Search in remote segments first.
Optional nextSegmentMetadataOpt = Optional.of(segmentMetadata);
while (nextSegmentMetadataOpt.isPresent()) {
- Optional txnIndexOpt = nextSegmentMetadataOpt.map(metadata -> indexCache.getIndexEntry(metadata).txnIndex());
+ Optional txnIndexOpt = nextSegmentMetadataOpt.flatMap(metadata -> indexCache.getIndexEntry(metadata).txnIndex());
if (txnIndexOpt.isPresent()) {
TxnIndexSearchResult searchResult = txnIndexOpt.get().collectAbortedTxns(startOffset, upperBoundOffset);
accumulator.accept(searchResult.abortedTransactions);
@@ -1096,6 +1499,43 @@ public void close() {
}
}
+ // Visible for testing
+ public static class RetentionSizeData {
+ private final long retentionSize;
+ private final long remainingBreachedSize;
+
+ public RetentionSizeData(long retentionSize, long remainingBreachedSize) {
+ if (retentionSize < 0)
+ throw new IllegalArgumentException("retentionSize should be non negative, but it is " + retentionSize);
+
+ if (remainingBreachedSize <= 0) {
+ throw new IllegalArgumentException("remainingBreachedSize should be more than zero, but it is " + remainingBreachedSize);
+ }
+
+ this.retentionSize = retentionSize;
+ this.remainingBreachedSize = remainingBreachedSize;
+ }
+ }
+
+ // Visible for testing
+ public static class RetentionTimeData {
+
+ private final long retentionMs;
+ private final long cleanupUntilMs;
+
+ public RetentionTimeData(long retentionMs, long cleanupUntilMs) {
+ if (retentionMs < 0)
+ throw new IllegalArgumentException("retentionMs should be non negative, but it is " + retentionMs);
+
+ if (cleanupUntilMs < retentionMs) {
+ throw new IllegalArgumentException("cleanupUntilMs [" + cleanupUntilMs + "] must be greater than retentionMs [" + retentionMs + "]");
+ }
+
+ this.retentionMs = retentionMs;
+ this.cleanupUntilMs = cleanupUntilMs;
+ }
+ }
+
// Visible for testing
static class EnrichedLogSegment {
private final LogSegment logSegment;
diff --git a/core/src/main/scala/kafka/log/LogManager.scala b/core/src/main/scala/kafka/log/LogManager.scala
index bed9f0dfa03b0..78e48010e5240 100755
--- a/core/src/main/scala/kafka/log/LogManager.scala
+++ b/core/src/main/scala/kafka/log/LogManager.scala
@@ -430,7 +430,8 @@ class LogManager(logDirs: Seq[File],
val remainingLogs = decNumRemainingLogs(numRemainingLogs, dir.getAbsolutePath)
val currentNumLoaded = logsToLoad.length - remainingLogs
log match {
- case Some(loadedLog) => info(s"Completed load of $loadedLog with ${loadedLog.numberOfSegments} segments in ${logLoadDurationMs}ms " +
+ case Some(loadedLog) => info(s"Completed load of $loadedLog with ${loadedLog.numberOfSegments} segments, " +
+ s"local-log-start-offset ${loadedLog.localLogStartOffset()} and log-end-offset ${loadedLog.logEndOffset} in ${logLoadDurationMs}ms " +
s"($currentNumLoaded/${logsToLoad.length} completed in $logDirAbsolutePath)")
case None => info(s"Error while loading logs in $logDir in ${logLoadDurationMs}ms ($currentNumLoaded/${logsToLoad.length} completed in $logDirAbsolutePath)")
}
diff --git a/core/src/main/scala/kafka/log/UnifiedLog.scala b/core/src/main/scala/kafka/log/UnifiedLog.scala
index 1b191844e7714..73b8ec003187e 100644
--- a/core/src/main/scala/kafka/log/UnifiedLog.scala
+++ b/core/src/main/scala/kafka/log/UnifiedLog.scala
@@ -147,11 +147,27 @@ class UnifiedLog(@volatile var logStartOffset: Long,
def localLogStartOffset(): Long = _localLogStartOffset
+ // This is the offset(inclusive) until which segments are copied to the remote storage.
@volatile private var highestOffsetInRemoteStorage: Long = -1L
locally {
+ def updateLocalLogStartOffset(offset: Long): Unit = {
+ _localLogStartOffset = offset
+
+ if (highWatermark < offset) {
+ updateHighWatermark(offset)
+ }
+
+ if (this.recoveryPoint < offset) {
+ localLog.updateRecoveryPoint(offset)
+ }
+ }
+
initializePartitionMetadata()
updateLogStartOffset(logStartOffset)
+ updateLocalLogStartOffset(math.max(logStartOffset, localLog.segments.firstSegmentBaseOffset.getOrElse(0L)))
+ if (!remoteLogEnabled())
+ logStartOffset = localLogStartOffset()
maybeIncrementFirstUnstableOffset()
initializeTopicId()
@@ -162,6 +178,14 @@ class UnifiedLog(@volatile var logStartOffset: Long,
logOffsetsListener = listener
}
+ def updateLogStartOffsetFromRemoteTier(remoteLogStartOffset: Long): Unit = {
+ if (!remoteLogEnabled()) {
+ error("Ignoring the call as the remote log storage is disabled")
+ return;
+ }
+ maybeIncrementLogStartOffset(remoteLogStartOffset, LogStartOffsetIncrementReason.SegmentDeletion)
+ }
+
def remoteLogEnabled(): Boolean = {
// Remote log is enabled only for non-compact and non-internal topics
remoteStorageSystemEnable &&
@@ -520,6 +544,7 @@ class UnifiedLog(@volatile var logStartOffset: Long,
localLog.updateRecoveryPoint(offset)
}
}
+
def updateHighestOffsetInRemoteStorage(offset: Long): Unit = {
if (!remoteLogEnabled())
warn(s"Unable to update the highest offset in remote storage with offset $offset since remote storage is not enabled. The existing highest offset is $highestOffsetInRemoteStorage.")
@@ -956,6 +981,15 @@ class UnifiedLog(@volatile var logStartOffset: Long,
}
}
+ private def maybeIncrementLocalLogStartOffset(newLocalLogStartOffset: Long, reason: LogStartOffsetIncrementReason): Unit = {
+ lock synchronized {
+ if (newLocalLogStartOffset > localLogStartOffset()) {
+ _localLogStartOffset = newLocalLogStartOffset
+ info(s"Incremented local log start offset to ${localLogStartOffset()} due to reason $reason")
+ }
+ }
+ }
+
/**
* Increment the log start offset if the provided offset is larger.
*
@@ -966,7 +1000,8 @@ class UnifiedLog(@volatile var logStartOffset: Long,
* @throws OffsetOutOfRangeException if the log start offset is greater than the high watermark
* @return true if the log start offset was updated; otherwise false
*/
- def maybeIncrementLogStartOffset(newLogStartOffset: Long, reason: LogStartOffsetIncrementReason): Boolean = {
+ def maybeIncrementLogStartOffset(newLogStartOffset: Long,
+ reason: LogStartOffsetIncrementReason): Boolean = {
// We don't have to write the log start offset to log-start-offset-checkpoint immediately.
// The deleteRecordsOffset may be lost only if all in-sync replicas of this broker are shutdown
// in an unclean manner within log.flush.start.offset.checkpoint.interval.ms. The chance of this happening is low.
@@ -977,11 +1012,15 @@ class UnifiedLog(@volatile var logStartOffset: Long,
throw new OffsetOutOfRangeException(s"Cannot increment the log start offset to $newLogStartOffset of partition $topicPartition " +
s"since it is larger than the high watermark $highWatermark")
+ if (remoteLogEnabled()) {
+ // This should be set log-start-offset is set more than the current local-log-start-offset
+ _localLogStartOffset = math.max(newLogStartOffset, localLogStartOffset())
+ }
+
localLog.checkIfMemoryMappedBufferClosed()
if (newLogStartOffset > logStartOffset) {
updatedLogStartOffset = true
updateLogStartOffset(newLogStartOffset)
- _localLogStartOffset = newLogStartOffset
info(s"Incremented log start offset to $newLogStartOffset due to $reason")
leaderEpochCache.foreach(_.truncateFromStart(logStartOffset))
producerStateManager.onLogStartOffsetIncremented(newLogStartOffset)
@@ -1292,7 +1331,7 @@ class UnifiedLog(@volatile var logStartOffset: Long,
latestEpochAsOptional(leaderEpochCache)))
} else {
// We need to search the first segment whose largest timestamp is >= the target timestamp if there is one.
- val remoteOffset = if (remoteLogEnabled()) {
+ if (remoteLogEnabled()) {
if (remoteLogManager.isEmpty) {
throw new KafkaException("RemoteLogManager is empty even though the remote log storage is enabled.")
}
@@ -1300,25 +1339,28 @@ class UnifiedLog(@volatile var logStartOffset: Long,
throw new KafkaException("Tiered storage is supported only with versions supporting leader epochs, that means RecordVersion must be >= 2.")
}
- remoteLogManager.get.findOffsetByTimestamp(topicPartition, targetTimestamp, logStartOffset, leaderEpochCache.get)
- } else Optional.empty()
-
- if (remoteOffset.isPresent) {
- remoteOffset.asScala
+ val remoteOffset = remoteLogManager.get.findOffsetByTimestamp(topicPartition, targetTimestamp, logStartOffset, leaderEpochCache.get)
+ if (remoteOffset.isPresent) {
+ remoteOffset.asScala
+ } else {
+ // If it is not found in remote log storage, search in the local log storage from local log start offset.
+ searchOffsetInLocalLog(targetTimestamp, localLogStartOffset())
+ }
} else {
- // If it is not found in remote storage, search in the local storage starting with local log start offset.
-
- // Cache to avoid race conditions. `toBuffer` is faster than most alternatives and provides
- // constant time access while being safe to use with concurrent collections unlike `toArray`.
- val segmentsCopy = logSegments.toBuffer
-
- val targetSeg = segmentsCopy.find(_.largestTimestamp >= targetTimestamp)
- targetSeg.flatMap(_.findOffsetByTimestamp(targetTimestamp, _localLogStartOffset))
+ searchOffsetInLocalLog(targetTimestamp, logStartOffset)
}
}
}
}
+ private def searchOffsetInLocalLog(targetTimestamp: Long, startOffset: Long): Option[TimestampAndOffset] = {
+ // Cache to avoid race conditions. `toBuffer` is faster than most alternatives and provides
+ // constant time access while being safe to use with concurrent collections unlike `toArray`.
+ val segmentsCopy = logSegments.toBuffer
+ val targetSeg = segmentsCopy.find(_.largestTimestamp >= targetTimestamp)
+ targetSeg.flatMap(_.findOffsetByTimestamp(targetTimestamp, startOffset))
+ }
+
def legacyFetchOffsetsBefore(timestamp: Long, maxNumOffsets: Int): Seq[Long] = {
// Cache to avoid race conditions. `toBuffer` is faster than most alternatives and provides
// constant time access while being safe to use with concurrent collections unlike `toArray`.
@@ -1390,7 +1432,13 @@ class UnifiedLog(@volatile var logStartOffset: Long,
private def deleteOldSegments(predicate: (LogSegment, Option[LogSegment]) => Boolean,
reason: SegmentDeletionReason): Int = {
def shouldDelete(segment: LogSegment, nextSegmentOpt: Option[LogSegment]): Boolean = {
- highWatermark >= nextSegmentOpt.map(_.baseOffset).getOrElse(localLog.logEndOffset) &&
+ val upperBoundOffset = nextSegmentOpt.map(_.baseOffset).getOrElse(localLog.logEndOffset)
+
+ // Check not to delete segments which are not yet copied to tiered storage if remote log is enabled.
+ (!remoteLogEnabled() || (upperBoundOffset > 0 && upperBoundOffset - 1 <= highestOffsetInRemoteStorage)) &&
+ // We don't delete segments with offsets at or beyond the high watermark to ensure that the log start
+ // offset can never exceed it.
+ highWatermark >= upperBoundOffset &&
predicate(segment, nextSegmentOpt)
}
lock synchronized {
@@ -1402,6 +1450,11 @@ class UnifiedLog(@volatile var logStartOffset: Long,
}
}
+ private def incrementStartOffset(startOffset: Long, reason: LogStartOffsetIncrementReason): Unit = {
+ if (remoteLogEnabled()) maybeIncrementLocalLogStartOffset(startOffset, reason)
+ else maybeIncrementLogStartOffset(startOffset, reason)
+ }
+
private def deleteSegments(deletable: Iterable[LogSegment], reason: SegmentDeletionReason): Int = {
maybeHandleIOException(s"Error while deleting segments for $topicPartition in dir ${dir.getParent}") {
val numToDelete = deletable.size
@@ -1419,7 +1472,7 @@ class UnifiedLog(@volatile var logStartOffset: Long,
// remove the segments for lookups
localLog.removeAndDeleteSegments(segmentsToDelete, asyncDelete = true, reason)
deleteProducerSnapshots(deletable, asyncDelete = true)
- maybeIncrementLogStartOffset(localLog.segments.firstSegmentBaseOffset.get, LogStartOffsetIncrementReason.SegmentDeletion)
+ incrementStartOffset(localLog.segments.firstSegmentBaseOffset.get, LogStartOffsetIncrementReason.SegmentDeletion)
}
numToDelete
}
@@ -1442,19 +1495,21 @@ class UnifiedLog(@volatile var logStartOffset: Long,
}
private def deleteRetentionMsBreachedSegments(): Int = {
- if (config.retentionMs < 0) return 0
+ val retentionMs = localRetentionMs(config, remoteLogEnabled())
+ if (retentionMs < 0) return 0
val startMs = time.milliseconds
def shouldDelete(segment: LogSegment, nextSegmentOpt: Option[LogSegment]): Boolean = {
- startMs - segment.largestTimestamp > config.retentionMs
+ startMs - segment.largestTimestamp > retentionMs
}
- deleteOldSegments(shouldDelete, RetentionMsBreach(this))
+ deleteOldSegments(shouldDelete, RetentionMsBreach(this, remoteLogEnabled()))
}
private def deleteRetentionSizeBreachedSegments(): Int = {
- if (config.retentionSize < 0 || size < config.retentionSize) return 0
- var diff = size - config.retentionSize
+ val retentionSize: Long = localRetentionSize(config, remoteLogEnabled())
+ if (retentionSize < 0 || size < retentionSize) return 0
+ var diff = size - retentionSize
def shouldDelete(segment: LogSegment, nextSegmentOpt: Option[LogSegment]): Boolean = {
if (diff - segment.size >= 0) {
diff -= segment.size
@@ -1464,15 +1519,15 @@ class UnifiedLog(@volatile var logStartOffset: Long,
}
}
- deleteOldSegments(shouldDelete, RetentionSizeBreach(this))
+ deleteOldSegments(shouldDelete, RetentionSizeBreach(this, remoteLogEnabled()))
}
private def deleteLogStartOffsetBreachedSegments(): Int = {
def shouldDelete(segment: LogSegment, nextSegmentOpt: Option[LogSegment]): Boolean = {
- nextSegmentOpt.exists(_.baseOffset <= logStartOffset)
+ nextSegmentOpt.exists(_.baseOffset <= (if (remoteLogEnabled()) localLogStartOffset() else logStartOffset))
}
- deleteOldSegments(shouldDelete, StartOffsetBreach(this))
+ deleteOldSegments(shouldDelete, StartOffsetBreach(this, remoteLogEnabled()))
}
def isFuture: Boolean = localLog.isFuture
@@ -1482,6 +1537,11 @@ class UnifiedLog(@volatile var logStartOffset: Long,
*/
def size: Long = localLog.segments.sizeInBytes
+ /**
+ * The log size in bytes for all segments that are only in local log but not yet in remote log.
+ */
+ def onlyLocalLogSegmentsSize: Long = UnifiedLog.sizeInBytes(logSegments.filter(_.baseOffset >= highestOffsetInRemoteStorage))
+
/**
* The offset of the next message that will be appended to the log
*/
@@ -2173,6 +2233,14 @@ object UnifiedLog extends Logging {
}
}
+ private[log] def localRetentionMs(config: LogConfig, remoteLogEnabled: Boolean): Long = {
+ if (remoteLogEnabled) config.remoteLogConfig.localRetentionMs else config.retentionMs
+ }
+
+ private[log] def localRetentionSize(config: LogConfig, remoteLogEnabled: Boolean): Long = {
+ if (remoteLogEnabled) config.remoteLogConfig.localRetentionBytes else config.retentionSize
+ }
+
}
object LogMetricNames {
@@ -2186,35 +2254,48 @@ object LogMetricNames {
}
}
-case class RetentionMsBreach(log: UnifiedLog) extends SegmentDeletionReason {
+case class RetentionMsBreach(log: UnifiedLog, remoteLogEnabled: Boolean) extends SegmentDeletionReason {
override def logReason(toDelete: List[LogSegment]): Unit = {
- val retentionMs = log.config.retentionMs
+ val retentionMs = UnifiedLog.localRetentionMs(log.config, remoteLogEnabled)
toDelete.foreach { segment =>
segment.largestRecordTimestamp match {
case Some(_) =>
- log.info(s"Deleting segment $segment due to retention time ${retentionMs}ms breach based on the largest " +
- s"record timestamp in the segment")
+ if (remoteLogEnabled)
+ log.info(s"Deleting segment $segment due to local log retention time ${retentionMs}ms breach based on the largest " +
+ s"record timestamp in the segment")
+ else
+ log.info(s"Deleting segment $segment due to log retention time ${retentionMs}ms breach based on the largest " +
+ s"record timestamp in the segment")
case None =>
- log.info(s"Deleting segment $segment due to retention time ${retentionMs}ms breach based on the " +
- s"last modified time of the segment")
+ if (remoteLogEnabled)
+ log.info(s"Deleting segment $segment due to local log retention time ${retentionMs}ms breach based on the " +
+ s"last modified time of the segment")
+ else
+ log.info(s"Deleting segment $segment due to log retention time ${retentionMs}ms breach based on the " +
+ s"last modified time of the segment")
}
}
}
}
-case class RetentionSizeBreach(log: UnifiedLog) extends SegmentDeletionReason {
+case class RetentionSizeBreach(log: UnifiedLog, remoteLogEnabled: Boolean) extends SegmentDeletionReason {
override def logReason(toDelete: List[LogSegment]): Unit = {
var size = log.size
toDelete.foreach { segment =>
size -= segment.size
- log.info(s"Deleting segment $segment due to retention size ${log.config.retentionSize} breach. Log size " +
+ if (remoteLogEnabled) log.info(s"Deleting segment $segment due to local log retention size ${UnifiedLog.localRetentionSize(log.config, remoteLogEnabled)} breach. " +
+ s"Local log size after deletion will be $size.")
+ else log.info(s"Deleting segment $segment due to log retention size ${log.config.retentionSize} breach. Log size " +
s"after deletion will be $size.")
}
}
}
-case class StartOffsetBreach(log: UnifiedLog) extends SegmentDeletionReason {
+case class StartOffsetBreach(log: UnifiedLog, remoteLogEnabled: Boolean) extends SegmentDeletionReason {
override def logReason(toDelete: List[LogSegment]): Unit = {
- log.info(s"Deleting segments due to log start offset ${log.logStartOffset} breach: ${toDelete.mkString(",")}")
+ if (remoteLogEnabled)
+ log.info(s"Deleting segments due to local log start offset ${log.localLogStartOffset()} breach: ${toDelete.mkString(",")}")
+ else
+ log.info(s"Deleting segments due to log start offset ${log.logStartOffset} breach: ${toDelete.mkString(",")}")
}
}
diff --git a/core/src/main/scala/kafka/server/BrokerServer.scala b/core/src/main/scala/kafka/server/BrokerServer.scala
index bd55cc2571a86..997f9d8d98958 100644
--- a/core/src/main/scala/kafka/server/BrokerServer.scala
+++ b/core/src/main/scala/kafka/server/BrokerServer.scala
@@ -575,7 +575,13 @@ class BrokerServer(
}
Some(new RemoteLogManager(config.remoteLogManagerConfig, config.brokerId, config.logDirs.head, clusterId, time,
- (tp: TopicPartition) => logManager.getLog(tp).asJava, brokerTopicStats));
+ (tp: TopicPartition) => logManager.getLog(tp).asJava,
+ (tp: TopicPartition, remoteLogStartOffset: java.lang.Long) => {
+ logManager.getLog(tp).foreach { log =>
+ log.updateLogStartOffsetFromRemoteTier(remoteLogStartOffset)
+ }
+ },
+ brokerTopicStats))
} else {
None
}
diff --git a/core/src/main/scala/kafka/server/KafkaServer.scala b/core/src/main/scala/kafka/server/KafkaServer.scala
index 645ad40f9e6bc..9c30719f737a1 100755
--- a/core/src/main/scala/kafka/server/KafkaServer.scala
+++ b/core/src/main/scala/kafka/server/KafkaServer.scala
@@ -614,7 +614,13 @@ class KafkaServer(
}
Some(new RemoteLogManager(config.remoteLogManagerConfig, config.brokerId, config.logDirs.head, clusterId, time,
- (tp: TopicPartition) => logManager.getLog(tp).asJava, brokerTopicStats));
+ (tp: TopicPartition) => logManager.getLog(tp).asJava,
+ (tp: TopicPartition, remoteLogStartOffset: java.lang.Long) => {
+ logManager.getLog(tp).foreach { log =>
+ log.updateLogStartOffsetFromRemoteTier(remoteLogStartOffset)
+ }
+ },
+ brokerTopicStats));
} else {
None
}
diff --git a/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java b/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java
index f0b89aca85578..5fd6e41a2ecb7 100644
--- a/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java
+++ b/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java
@@ -73,7 +73,6 @@
import java.io.ByteArrayInputStream;
import java.io.File;
-import java.io.FileInputStream;
import java.io.IOException;
import java.nio.file.Files;
import java.util.ArrayList;
@@ -93,6 +92,7 @@
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.function.BiConsumer;
+import java.util.function.Supplier;
import java.util.stream.Collectors;
import static org.apache.kafka.server.log.remote.metadata.storage.TopicBasedRemoteLogMetadataManagerConfig.REMOTE_LOG_METADATA_COMMON_CLIENT_PREFIX;
@@ -127,39 +127,40 @@
import static org.mockito.Mockito.when;
public class RemoteLogManagerTest {
- Time time = new MockTime();
- int brokerId = 0;
- String logDir = TestUtils.tempDirectory("kafka-").toString();
- String clusterId = "dummyId";
- String remoteLogStorageTestProp = "remote.log.storage.test";
- String remoteLogStorageTestVal = "storage.test";
- String remoteLogMetadataTestProp = "remote.log.metadata.test";
- String remoteLogMetadataTestVal = "metadata.test";
- String remoteLogMetadataCommonClientTestProp = REMOTE_LOG_METADATA_COMMON_CLIENT_PREFIX + "common.client.test";
- String remoteLogMetadataCommonClientTestVal = "common.test";
- String remoteLogMetadataProducerTestProp = REMOTE_LOG_METADATA_PRODUCER_PREFIX + "producer.test";
- String remoteLogMetadataProducerTestVal = "producer.test";
- String remoteLogMetadataConsumerTestProp = REMOTE_LOG_METADATA_CONSUMER_PREFIX + "consumer.test";
- String remoteLogMetadataConsumerTestVal = "consumer.test";
- String remoteLogMetadataTopicPartitionsNum = "1";
-
- RemoteStorageManager remoteStorageManager = mock(RemoteStorageManager.class);
- RemoteLogMetadataManager remoteLogMetadataManager = mock(RemoteLogMetadataManager.class);
- RemoteLogManagerConfig remoteLogManagerConfig = null;
-
- BrokerTopicStats brokerTopicStats = null;
- RemoteLogManager remoteLogManager = null;
-
- TopicIdPartition leaderTopicIdPartition = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("Leader", 0));
- TopicIdPartition followerTopicIdPartition = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("Follower", 0));
- Map topicIds = new HashMap<>();
- TopicPartition tp = new TopicPartition("TestTopic", 5);
- EpochEntry epochEntry0 = new EpochEntry(0, 0);
- EpochEntry epochEntry1 = new EpochEntry(1, 100);
- EpochEntry epochEntry2 = new EpochEntry(2, 200);
- List totalEpochEntries = Arrays.asList(epochEntry0, epochEntry1, epochEntry2);
- LeaderEpochCheckpoint checkpoint = new LeaderEpochCheckpoint() {
+ private final Time time = new MockTime();
+ private final int brokerId = 0;
+ private final String logDir = TestUtils.tempDirectory("kafka-").toString();
+ private final String clusterId = "dummyId";
+ private final String remoteLogStorageTestProp = "remote.log.storage.test";
+ private final String remoteLogStorageTestVal = "storage.test";
+ private final String remoteLogMetadataTestProp = "remote.log.metadata.test";
+ private final String remoteLogMetadataTestVal = "metadata.test";
+ private final String remoteLogMetadataCommonClientTestProp = REMOTE_LOG_METADATA_COMMON_CLIENT_PREFIX + "common.client.test";
+ private final String remoteLogMetadataCommonClientTestVal = "common.test";
+ private final String remoteLogMetadataProducerTestProp = REMOTE_LOG_METADATA_PRODUCER_PREFIX + "producer.test";
+ private final String remoteLogMetadataProducerTestVal = "producer.test";
+ private final String remoteLogMetadataConsumerTestProp = REMOTE_LOG_METADATA_CONSUMER_PREFIX + "consumer.test";
+ private final String remoteLogMetadataConsumerTestVal = "consumer.test";
+ private final String remoteLogMetadataTopicPartitionsNum = "1";
+
+ private final RemoteStorageManager remoteStorageManager = mock(RemoteStorageManager.class);
+ private final RemoteLogMetadataManager remoteLogMetadataManager = mock(RemoteLogMetadataManager.class);
+ private RemoteLogManagerConfig remoteLogManagerConfig = null;
+
+ private BrokerTopicStats brokerTopicStats = null;
+ private RemoteLogManager remoteLogManager = null;
+
+ private final TopicIdPartition leaderTopicIdPartition = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("Leader", 0));
+ private final TopicIdPartition followerTopicIdPartition = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("Follower", 0));
+ private final Map topicIds = new HashMap<>();
+ private final TopicPartition tp = new TopicPartition("TestTopic", 5);
+ private final EpochEntry epochEntry0 = new EpochEntry(0, 0);
+ private final EpochEntry epochEntry1 = new EpochEntry(1, 100);
+ private final EpochEntry epochEntry2 = new EpochEntry(2, 200);
+ private final List totalEpochEntries = Arrays.asList(epochEntry0, epochEntry1, epochEntry2);
+ private final LeaderEpochCheckpoint checkpoint = new LeaderEpochCheckpoint() {
List epochs = Collections.emptyList();
+
@Override
public void write(Collection epochs) {
this.epochs = new ArrayList<>(epochs);
@@ -171,7 +172,7 @@ public List read() {
}
};
- UnifiedLog mockLog = mock(UnifiedLog.class);
+ private final UnifiedLog mockLog = mock(UnifiedLog.class);
@BeforeEach
void setUp() throws Exception {
@@ -183,8 +184,10 @@ void setUp() throws Exception {
brokerTopicStats = new BrokerTopicStats(Optional.of(KafkaConfig.fromProps(props)));
kafka.utils.TestUtils.clearYammerMetrics();
-
- remoteLogManager = new RemoteLogManager(remoteLogManagerConfig, brokerId, logDir, clusterId, time, tp -> Optional.of(mockLog), brokerTopicStats) {
+ remoteLogManager = new RemoteLogManager(remoteLogManagerConfig, brokerId, logDir, clusterId, time,
+ tp -> Optional.of(mockLog),
+ (topicPartition, offset) -> { },
+ brokerTopicStats) {
public RemoteStorageManager createRemoteStorageManager() {
return remoteStorageManager;
}
@@ -273,7 +276,15 @@ void testRemoteLogMetadataManagerWithEndpointConfigOverridden() throws IOExcepti
Properties props = new Properties();
// override common security.protocol by adding "RLMM prefix" and "remote log metadata common client prefix"
props.put(DEFAULT_REMOTE_LOG_METADATA_MANAGER_CONFIG_PREFIX + REMOTE_LOG_METADATA_COMMON_CLIENT_PREFIX + "security.protocol", "SSL");
- try (RemoteLogManager remoteLogManager = new RemoteLogManager(createRLMConfig(props), brokerId, logDir, clusterId, time, tp -> Optional.of(mockLog), brokerTopicStats) {
+ try (RemoteLogManager remoteLogManager = new RemoteLogManager(
+ createRLMConfig(props),
+ brokerId,
+ logDir,
+ clusterId,
+ time,
+ tp -> Optional.of(mockLog),
+ (topicPartition, offset) -> { },
+ brokerTopicStats) {
public RemoteStorageManager createRemoteStorageManager() {
return remoteStorageManager;
}
@@ -790,7 +801,10 @@ private void verifyLogSegmentData(LogSegmentData logSegmentData,
void testGetClassLoaderAwareRemoteStorageManager() throws Exception {
ClassLoaderAwareRemoteStorageManager rsmManager = mock(ClassLoaderAwareRemoteStorageManager.class);
try (RemoteLogManager remoteLogManager =
- new RemoteLogManager(remoteLogManagerConfig, brokerId, logDir, clusterId, time, t -> Optional.empty(), brokerTopicStats) {
+ new RemoteLogManager(remoteLogManagerConfig, brokerId, logDir, clusterId, time,
+ t -> Optional.empty(),
+ (topicPartition, offset) -> { },
+ brokerTopicStats) {
public RemoteStorageManager createRemoteStorageManager() {
return rsmManager;
}
@@ -887,77 +901,122 @@ void testRLMTaskShouldSetLeaderEpochCorrectly() {
@Test
void testFindOffsetByTimestamp() throws IOException, RemoteStorageException {
TopicPartition tp = leaderTopicIdPartition.topicPartition();
- RemoteLogSegmentId remoteLogSegmentId = new RemoteLogSegmentId(leaderTopicIdPartition, Uuid.randomUuid());
+
+ long ts = time.milliseconds();
+ long startOffset = 120;
+ int targetLeaderEpoch = 10;
+
+ TreeMap validSegmentEpochs = new TreeMap<>();
+ validSegmentEpochs.put(targetLeaderEpoch, startOffset);
+
+ LeaderEpochFileCache leaderEpochFileCache = new LeaderEpochFileCache(tp, checkpoint);
+ leaderEpochFileCache.assign(4, 99L);
+ leaderEpochFileCache.assign(5, 99L);
+ leaderEpochFileCache.assign(targetLeaderEpoch, startOffset);
+ leaderEpochFileCache.assign(12, 500L);
+
+ doTestFindOffsetByTimestamp(ts, startOffset, targetLeaderEpoch, validSegmentEpochs);
+
+ // Fetching message for timestamp `ts` will return the message with startOffset+1, and `ts+1` as there are no
+ // messages starting with the startOffset and with `ts`.
+ Optional maybeTimestampAndOffset1 = remoteLogManager.findOffsetByTimestamp(tp, ts, startOffset, leaderEpochFileCache);
+ assertEquals(Optional.of(new FileRecords.TimestampAndOffset(ts + 1, startOffset + 1, Optional.of(targetLeaderEpoch))), maybeTimestampAndOffset1);
+
+ // Fetching message for `ts+2` will return the message with startOffset+2 and its timestamp value is `ts+2`.
+ Optional maybeTimestampAndOffset2 = remoteLogManager.findOffsetByTimestamp(tp, ts + 2, startOffset, leaderEpochFileCache);
+ assertEquals(Optional.of(new FileRecords.TimestampAndOffset(ts + 2, startOffset + 2, Optional.of(targetLeaderEpoch))), maybeTimestampAndOffset2);
+
+ // Fetching message for `ts+3` will return None as there are no records with timestamp >= ts+3.
+ Optional maybeTimestampAndOffset3 = remoteLogManager.findOffsetByTimestamp(tp, ts + 3, startOffset, leaderEpochFileCache);
+ assertEquals(Optional.empty(), maybeTimestampAndOffset3);
+ }
+
+ @Test
+ void testFindOffsetByTimestampWithInvalidEpochSegments() throws IOException, RemoteStorageException {
+ TopicPartition tp = leaderTopicIdPartition.topicPartition();
+
long ts = time.milliseconds();
long startOffset = 120;
int targetLeaderEpoch = 10;
+ TreeMap validSegmentEpochs = new TreeMap<>();
+ validSegmentEpochs.put(targetLeaderEpoch - 1, startOffset - 1); // invalid epochs not aligning with leader epoch cache
+ validSegmentEpochs.put(targetLeaderEpoch, startOffset);
+
+ LeaderEpochFileCache leaderEpochFileCache = new LeaderEpochFileCache(tp, checkpoint);
+ leaderEpochFileCache.assign(4, 99L);
+ leaderEpochFileCache.assign(5, 99L);
+ leaderEpochFileCache.assign(targetLeaderEpoch, startOffset);
+ leaderEpochFileCache.assign(12, 500L);
+
+ doTestFindOffsetByTimestamp(ts, startOffset, targetLeaderEpoch, validSegmentEpochs);
+
+ // Fetch offsets for this segment returns empty as the segment epochs are not with in the leader epoch cache.
+ Optional maybeTimestampAndOffset1 = remoteLogManager.findOffsetByTimestamp(tp, ts, startOffset, leaderEpochFileCache);
+ assertEquals(Optional.empty(), maybeTimestampAndOffset1);
+
+ Optional maybeTimestampAndOffset2 = remoteLogManager.findOffsetByTimestamp(tp, ts + 2, startOffset, leaderEpochFileCache);
+ assertEquals(Optional.empty(), maybeTimestampAndOffset2);
+
+ Optional maybeTimestampAndOffset3 = remoteLogManager.findOffsetByTimestamp(tp, ts + 3, startOffset, leaderEpochFileCache);
+ assertEquals(Optional.empty(), maybeTimestampAndOffset3);
+ }
+
+ private void doTestFindOffsetByTimestamp(long ts, long startOffset, int targetLeaderEpoch,
+ TreeMap validSegmentEpochs) throws IOException, RemoteStorageException {
+ TopicPartition tp = leaderTopicIdPartition.topicPartition();
+ RemoteLogSegmentId remoteLogSegmentId = new RemoteLogSegmentId(leaderTopicIdPartition, Uuid.randomUuid());
+
RemoteLogSegmentMetadata segmentMetadata = mock(RemoteLogSegmentMetadata.class);
when(segmentMetadata.remoteLogSegmentId()).thenReturn(remoteLogSegmentId);
when(segmentMetadata.maxTimestampMs()).thenReturn(ts + 2);
when(segmentMetadata.startOffset()).thenReturn(startOffset);
when(segmentMetadata.endOffset()).thenReturn(startOffset + 2);
+ when(segmentMetadata.segmentLeaderEpochs()).thenReturn(validSegmentEpochs);
File tpDir = new File(logDir, tp.toString());
Files.createDirectory(tpDir.toPath());
File txnIdxFile = new File(tpDir, "txn-index" + UnifiedLog.TxnIndexFileSuffix());
txnIdxFile.createNewFile();
when(remoteStorageManager.fetchIndex(any(RemoteLogSegmentMetadata.class), any(IndexType.class)))
- .thenAnswer(ans -> {
- RemoteLogSegmentMetadata metadata = ans.getArgument(0);
- IndexType indexType = ans.getArgument(1);
- int maxEntries = (int) (metadata.endOffset() - metadata.startOffset());
- OffsetIndex offsetIdx = new OffsetIndex(new File(tpDir, String.valueOf(metadata.startOffset()) + UnifiedLog.IndexFileSuffix()),
- metadata.startOffset(), maxEntries * 8);
- TimeIndex timeIdx = new TimeIndex(new File(tpDir, String.valueOf(metadata.startOffset()) + UnifiedLog.TimeIndexFileSuffix()),
- metadata.startOffset(), maxEntries * 12);
- switch (indexType) {
- case OFFSET:
- return new FileInputStream(offsetIdx.file());
- case TIMESTAMP:
- return new FileInputStream(timeIdx.file());
- case TRANSACTION:
- return new FileInputStream(txnIdxFile);
- }
- return null;
- });
+ .thenAnswer(ans -> {
+ RemoteLogSegmentMetadata metadata = ans.getArgument(0);
+ IndexType indexType = ans.getArgument(1);
+ int maxEntries = (int) (metadata.endOffset() - metadata.startOffset());
+ OffsetIndex offsetIdx = new OffsetIndex(new File(tpDir, String.valueOf(metadata.startOffset()) + UnifiedLog.IndexFileSuffix()),
+ metadata.startOffset(), maxEntries * 8);
+ TimeIndex timeIdx = new TimeIndex(new File(tpDir, String.valueOf(metadata.startOffset()) + UnifiedLog.TimeIndexFileSuffix()),
+ metadata.startOffset(), maxEntries * 12);
+ switch (indexType) {
+ case OFFSET:
+ return Files.newInputStream(offsetIdx.file().toPath());
+ case TIMESTAMP:
+ return Files.newInputStream(timeIdx.file().toPath());
+ case TRANSACTION:
+ return Files.newInputStream(txnIdxFile.toPath());
+ }
+ return null;
+ });
when(remoteLogMetadataManager.listRemoteLogSegments(eq(leaderTopicIdPartition), anyInt()))
- .thenAnswer(ans -> {
- int leaderEpoch = ans.getArgument(1);
- if (leaderEpoch == targetLeaderEpoch)
- return Collections.singleton(segmentMetadata).iterator();
- else
- return Collections.emptyList().iterator();
- });
-
-
+ .thenAnswer(ans -> {
+ int leaderEpoch = ans.getArgument(1);
+ if (leaderEpoch == targetLeaderEpoch)
+ return Collections.singleton(segmentMetadata).iterator();
+ else
+ return Collections.emptyIterator();
+ });
// 3 messages are added with offset, and timestamp as below
// startOffset , ts-1
// startOffset+1 , ts+1
// startOffset+2 , ts+2
when(remoteStorageManager.fetchLogSegment(segmentMetadata, 0))
- .thenAnswer(a -> new ByteArrayInputStream(records(ts, startOffset, targetLeaderEpoch).buffer().array()));
+ .thenAnswer(a -> new ByteArrayInputStream(records(ts, startOffset, targetLeaderEpoch).buffer().array()));
- LeaderEpochFileCache leaderEpochFileCache = new LeaderEpochFileCache(tp, checkpoint);
- leaderEpochFileCache.assign(5, 99L);
- leaderEpochFileCache.assign(targetLeaderEpoch, startOffset);
- leaderEpochFileCache.assign(12, 500L);
+ when(mockLog.logEndOffset()).thenReturn(600L);
remoteLogManager.onLeadershipChange(Collections.singleton(mockPartition(leaderTopicIdPartition)), Collections.emptySet(), topicIds);
- // Fetching message for timestamp `ts` will return the message with startOffset+1, and `ts+1` as there are no
- // messages starting with the startOffset and with `ts`.
- Optional maybeTimestampAndOffset1 = remoteLogManager.findOffsetByTimestamp(tp, ts, startOffset, leaderEpochFileCache);
- assertEquals(Optional.of(new FileRecords.TimestampAndOffset(ts + 1, startOffset + 1, Optional.of(targetLeaderEpoch))), maybeTimestampAndOffset1);
-
- // Fetching message for `ts+2` will return the message with startOffset+2 and its timestamp value is `ts+2`.
- Optional maybeTimestampAndOffset2 = remoteLogManager.findOffsetByTimestamp(tp, ts + 2, startOffset, leaderEpochFileCache);
- assertEquals(Optional.of(new FileRecords.TimestampAndOffset(ts + 2, startOffset + 2, Optional.of(targetLeaderEpoch))), maybeTimestampAndOffset2);
-
- // Fetching message for `ts+3` will return None as there are no records with timestamp >= ts+3.
- Optional maybeTimestampAndOffset3 = remoteLogManager.findOffsetByTimestamp(tp, ts + 3, startOffset, leaderEpochFileCache);
- assertEquals(Optional.empty(), maybeTimestampAndOffset3);
}
@Test
@@ -974,7 +1033,7 @@ public void testRemoveMetricsOnClose() throws IOException {
MockedConstruction mockMetricsGroupCtor = mockConstruction(KafkaMetricsGroup.class);
try {
RemoteLogManager remoteLogManager = new RemoteLogManager(remoteLogManagerConfig, brokerId, logDir, clusterId,
- time, tp -> Optional.of(mockLog), brokerTopicStats) {
+ time, tp -> Optional.of(mockLog), (topicPartition, offset) -> { }, brokerTopicStats) {
public RemoteStorageManager createRemoteStorageManager() {
return remoteStorageManager;
}
@@ -1007,6 +1066,155 @@ public RemoteLogMetadataManager createRemoteLogMetadataManager() {
}
}
+ private static RemoteLogSegmentMetadata createRemoteLogSegmentMetadata(long startOffset, long endOffset, Map segmentEpochs) {
+ return new RemoteLogSegmentMetadata(
+ new RemoteLogSegmentId(new TopicIdPartition(Uuid.randomUuid(),
+ new TopicPartition("topic", 0)), Uuid.randomUuid()),
+ startOffset, endOffset,
+ 100000L,
+ 1,
+ 100000L,
+ 1000,
+ Optional.empty(),
+ RemoteLogSegmentState.COPY_SEGMENT_FINISHED, segmentEpochs);
+ }
+
+ @Test
+ public void testBuildFilteredLeaderEpochMap() {
+ TreeMap leaderEpochToStartOffset = new TreeMap<>();
+ leaderEpochToStartOffset.put(0, 0L);
+ leaderEpochToStartOffset.put(1, 0L);
+ leaderEpochToStartOffset.put(2, 0L);
+ leaderEpochToStartOffset.put(3, 30L);
+ leaderEpochToStartOffset.put(4, 40L);
+ leaderEpochToStartOffset.put(5, 60L);
+ leaderEpochToStartOffset.put(6, 60L);
+ leaderEpochToStartOffset.put(7, 70L);
+ leaderEpochToStartOffset.put(8, 70L);
+
+ TreeMap expectedLeaderEpochs = new TreeMap<>();
+ expectedLeaderEpochs.put(2, 0L);
+ expectedLeaderEpochs.put(3, 30L);
+ expectedLeaderEpochs.put(4, 40L);
+ expectedLeaderEpochs.put(6, 60L);
+ expectedLeaderEpochs.put(8, 70L);
+
+ NavigableMap refinedLeaderEpochMap = RemoteLogManager.buildFilteredLeaderEpochMap(leaderEpochToStartOffset);
+ assertEquals(expectedLeaderEpochs, refinedLeaderEpochMap);
+ }
+
+ @Test
+ public void testRemoteSegmentWithinLeaderEpochs() {
+ // Test whether a remote segment is within the leader epochs
+ final long logEndOffset = 90L;
+
+ TreeMap leaderEpochToStartOffset = new TreeMap<>();
+ leaderEpochToStartOffset.put(0, 0L);
+ leaderEpochToStartOffset.put(1, 10L);
+ leaderEpochToStartOffset.put(2, 20L);
+ leaderEpochToStartOffset.put(3, 30L);
+ leaderEpochToStartOffset.put(4, 40L);
+ leaderEpochToStartOffset.put(5, 50L);
+ leaderEpochToStartOffset.put(7, 70L);
+
+ // Test whether a remote segment's epochs/offsets(multiple) are within the range of leader epochs
+ TreeMap segmentEpochs1 = new TreeMap<>();
+ segmentEpochs1.put(1, 15L);
+ segmentEpochs1.put(2, 20L);
+ segmentEpochs1.put(3, 30L);
+
+ assertTrue(RemoteLogManager.isRemoteSegmentWithinLeaderEpochs(createRemoteLogSegmentMetadata(
+ 15,
+ 35,
+ segmentEpochs1), logEndOffset, leaderEpochToStartOffset));
+
+ // Test whether a remote segment's epochs/offsets(single) are within the range of leader epochs
+ TreeMap segmentEpochs2 = new TreeMap<>();
+ segmentEpochs2.put(1, 15L);
+ assertTrue(RemoteLogManager.isRemoteSegmentWithinLeaderEpochs(createRemoteLogSegmentMetadata(
+ 15,
+ 19,
+ segmentEpochs2), logEndOffset, leaderEpochToStartOffset));
+
+ // Test whether a remote segment's start offset is same as the offset of the respective leader epoch entry.
+ TreeMap segmentEpochs3 = new TreeMap<>();
+ segmentEpochs3.put(0, 0L); // same as leader epoch's start offset
+ assertTrue(RemoteLogManager.isRemoteSegmentWithinLeaderEpochs(createRemoteLogSegmentMetadata(
+ 0,
+ 5,
+ segmentEpochs3), logEndOffset, leaderEpochToStartOffset));
+
+ // Test whether a remote segment's start offset is same as the offset of the respective leader epoch entry.
+ TreeMap segmentEpochs4 = new TreeMap<>();
+ segmentEpochs4.put(7, 70L); // same as leader epoch's start offset
+ assertTrue(RemoteLogManager.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 segmentEpochs5 = new TreeMap<>();
+ segmentEpochs5.put(1, 15L);
+ segmentEpochs5.put(2, 20L);
+
+ assertTrue(RemoteLogManager.isRemoteSegmentWithinLeaderEpochs(createRemoteLogSegmentMetadata(
+ 15,
+ 29, // same as end offset for epoch 2 in leaderEpochToStartOffset
+ segmentEpochs5), logEndOffset, leaderEpochToStartOffset));
+
+ // Test whether any of the epoch's is not with in the leader epoch chain.
+ TreeMap segmentEpochs6 = new TreeMap();
+ segmentEpochs6.put(5, 55L);
+ segmentEpochs6.put(6, 60L); // epoch 6 exists here but it is missing in leaderEpochToStartOffset
+ segmentEpochs6.put(7, 70L);
+
+ assertFalse(RemoteLogManager.isRemoteSegmentWithinLeaderEpochs(createRemoteLogSegmentMetadata(
+ 55,
+ 85,
+ segmentEpochs6), logEndOffset, leaderEpochToStartOffset));
+
+ // Test whether an epoch existing in remote segment does not exist in leader epoch chain.
+ TreeMap segmentEpochs7 = new TreeMap<>();
+ segmentEpochs7.put(1, 15L);
+ segmentEpochs7.put(2, 20L); // epoch 3 is missing here which exists in leaderEpochToStartOffset
+ segmentEpochs7.put(4, 40L);
+
+ assertFalse(RemoteLogManager.isRemoteSegmentWithinLeaderEpochs(createRemoteLogSegmentMetadata(
+ 15,
+ 45,
+ segmentEpochs7), logEndOffset, leaderEpochToStartOffset));
+
+ // Test a remote segment having larger end offset than the log end offset
+ TreeMap segmentEpochs8 = new TreeMap<>();
+ segmentEpochs8.put(1, 15L);
+ segmentEpochs8.put(2, 20L);
+
+ assertFalse(RemoteLogManager.isRemoteSegmentWithinLeaderEpochs(createRemoteLogSegmentMetadata(
+ 15,
+ 95, // larger than log end offset
+ segmentEpochs8), logEndOffset, leaderEpochToStartOffset));
+
+ // Test whether a segment's first offset is earlier to the respective epoch's start offset
+ TreeMap segmentEpochs9 = new TreeMap<>();
+ segmentEpochs9.put(1, 5L);
+ segmentEpochs9.put(2, 20L);
+
+ assertFalse(RemoteLogManager.isRemoteSegmentWithinLeaderEpochs(createRemoteLogSegmentMetadata(
+ 5, // earlier to epoch 1's start offset
+ 25,
+ segmentEpochs9), logEndOffset, leaderEpochToStartOffset));
+
+ // Test whether a segment's last offset is more than the respective epoch's end offset
+ TreeMap segmentEpochs10 = new TreeMap<>();
+ segmentEpochs10.put(1, 15L);
+ segmentEpochs10.put(2, 20L);
+ assertFalse(RemoteLogManager.isRemoteSegmentWithinLeaderEpochs(createRemoteLogSegmentMetadata(
+ 15,
+ 35, // more than epoch 2's end offset
+ segmentEpochs10), logEndOffset, leaderEpochToStartOffset));
+ }
+
@Test
public void testCandidateLogSegmentsSkipsActiveSegment() {
UnifiedLog log = mock(UnifiedLog.class);
@@ -1057,6 +1265,35 @@ public void testCandidateLogSegmentsSkipsSegmentsAfterLastStableOffset() {
assertEquals(expected, actual);
}
+ @Test
+ public void testRemoteSizeData() {
+ Supplier[] invalidRetentionSizeData =
+ new Supplier[]{
+ () -> new RemoteLogManager.RetentionSizeData(10, 0),
+ () -> new RemoteLogManager.RetentionSizeData(10, -1),
+ () -> new RemoteLogManager.RetentionSizeData(-1, 10),
+ () -> new RemoteLogManager.RetentionSizeData(-1, -1),
+ () -> new RemoteLogManager.RetentionSizeData(-1, 0)
+ };
+
+ for (Supplier invalidRetentionSizeDataEntry : invalidRetentionSizeData) {
+ assertThrows(IllegalArgumentException.class, invalidRetentionSizeDataEntry::get);
+ }
+ }
+
+ @Test
+ public void testRemoteSizeTime() {
+ Supplier[] invalidRetentionTimeData =
+ new Supplier[] {
+ () -> new RemoteLogManager.RetentionTimeData(-1, 10),
+ () -> new RemoteLogManager.RetentionTimeData(1000, 10),
+ };
+
+ for (Supplier invalidRetentionTimeDataEntry : invalidRetentionTimeData) {
+ assertThrows(IllegalArgumentException.class, invalidRetentionTimeDataEntry::get);
+ }
+ }
+
@Test
public void testStopPartitionsWithoutDeletion() throws RemoteStorageException {
BiConsumer errorHandler = (topicPartition, throwable) -> fail("shouldn't be called");
diff --git a/core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala b/core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala
index 8c98bfb927899..0104c55e4f2a2 100755
--- a/core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala
+++ b/core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala
@@ -3585,9 +3585,25 @@ class UnifiedLogTest {
log.updateHighWatermark(90L)
log.maybeIncrementLogStartOffset(20L, LogStartOffsetIncrementReason.SegmentDeletion)
assertEquals(20, log.logStartOffset)
- assertEquals(log.logStartOffset, log.localLogStartOffset())
}
+ @Test
+ def testStartOffsetsRemoteLogStorageIsEnabled(): Unit = {
+ val logConfig = LogTestUtils.createLogConfig(remoteLogStorageEnable = true)
+ val log = createLog(logDir, logConfig, remoteStorageSystemEnable = true)
+
+ for (i <- 0 until 100) {
+ val records = TestUtils.singletonRecords(value = s"test$i".getBytes)
+ log.appendAsLeader(records, leaderEpoch = 0)
+ }
+
+ log.updateHighWatermark(80L)
+ val newLogStartOffset = 40L;
+ log.maybeIncrementLogStartOffset(newLogStartOffset, LogStartOffsetIncrementReason.SegmentDeletion)
+ assertEquals(newLogStartOffset, log.logStartOffset)
+ assertEquals(log.logStartOffset, log.localLogStartOffset())
+ }
+
private class MockLogOffsetsListener extends LogOffsetsListener {
private var highWatermark: Long = -1L
diff --git a/core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala b/core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala
index 4b92da4007e84..4abe993cb9480 100644
--- a/core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala
+++ b/core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala
@@ -35,7 +35,7 @@ import org.slf4j.{Logger, LoggerFactory}
import java.io.{File, FileInputStream}
import java.nio.file.Files
import java.util
-import java.util.Collections
+import java.util.{Collections, Optional}
import java.util.concurrent.{CountDownLatch, Executors, TimeUnit}
import scala.collection.mutable
@@ -101,7 +101,7 @@ class RemoteIndexCacheTest {
def testIndexFileNameAndLocationOnDisk(): Unit = {
val entry = cache.getIndexEntry(rlsMetadata)
val offsetIndexFile = entry.offsetIndex.file().toPath
- val txnIndexFile = entry.txnIndex.file().toPath
+ val txnIndexFile = entry.txnIndex.get().file().toPath
val timeIndexFile = entry.timeIndex.file().toPath
val expectedOffsetIndexFileName: String = remoteOffsetIndexFileName(rlsMetadata)
@@ -269,7 +269,7 @@ class RemoteIndexCacheTest {
// verify that index(s) rename is only called 1 time
verify(cacheEntry.timeIndex).renameTo(any(classOf[File]))
verify(cacheEntry.offsetIndex).renameTo(any(classOf[File]))
- verify(cacheEntry.txnIndex).renameTo(any(classOf[File]))
+ verify(cacheEntry.txnIndex.get()).renameTo(any(classOf[File]))
// verify no index files on disk
assertFalse(getIndexFileFromDisk(LogFileUtils.INDEX_FILE_SUFFIX).isPresent,
@@ -327,12 +327,12 @@ class RemoteIndexCacheTest {
verify(spyEntry).close()
// close for all index entries must be invoked
- verify(spyEntry.txnIndex).close()
+ verify(spyEntry.txnIndex.get()).close()
verify(spyEntry.offsetIndex).close()
verify(spyEntry.timeIndex).close()
// index files must not be deleted
- verify(spyEntry.txnIndex, times(0)).deleteIfExists()
+ verify(spyEntry.txnIndex.get(), times(0)).deleteIfExists()
verify(spyEntry.offsetIndex, times(0)).deleteIfExists()
verify(spyEntry.timeIndex, times(0)).deleteIfExists()
@@ -500,9 +500,9 @@ class RemoteIndexCacheTest {
val rlsMetadata = new RemoteLogSegmentMetadata(remoteLogSegmentId, baseOffset, lastOffset,
time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata))
- val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata))
+ val txnIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata))
val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata))
- spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, Optional.of(txnIndex)))
}
private def assertAtLeastOnePresent(cache: RemoteIndexCache, uuids: Uuid*): Unit = {
diff --git a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
index 868f3c76ecec2..d9cd8cb0ab769 100644
--- a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
@@ -3631,6 +3631,7 @@ class ReplicaManagerTest {
"clusterId",
time,
_ => Optional.of(mockLog),
+ (TopicPartition, Long) => {},
brokerTopicStats)
val spyRLM = spy(remoteLogManager)
diff --git a/storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/ConsumerManager.java b/storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/ConsumerManager.java
index 14ec707a2ebb4..186cbb17c56f1 100644
--- a/storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/ConsumerManager.java
+++ b/storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/ConsumerManager.java
@@ -27,9 +27,7 @@
import org.slf4j.LoggerFactory;
import java.io.Closeable;
-import java.io.File;
import java.io.IOException;
-import java.nio.file.Path;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.TimeoutException;
@@ -41,8 +39,6 @@
*/
public class ConsumerManager implements Closeable {
- public static final String COMMITTED_OFFSETS_FILE_NAME = "_rlmm_committed_offsets";
-
private static final Logger log = LoggerFactory.getLogger(ConsumerManager.class);
private static final long CONSUME_RECHECK_INTERVAL_MS = 50L;
@@ -60,15 +56,13 @@ public ConsumerManager(TopicBasedRemoteLogMetadataManagerConfig rlmmConfig,
//Create a task to consume messages and submit the respective events to RemotePartitionMetadataEventHandler.
KafkaConsumer consumer = new KafkaConsumer<>(rlmmConfig.consumerProperties());
- Path committedOffsetsPath = new File(rlmmConfig.logDir(), COMMITTED_OFFSETS_FILE_NAME).toPath();
consumerTask = new ConsumerTask(
- consumer,
- rlmmConfig.remoteLogMetadataTopicName(),
- remotePartitionMetadataEventHandler,
- topicPartitioner,
- committedOffsetsPath,
- time,
- 60_000L
+ remotePartitionMetadataEventHandler,
+ topicPartitioner,
+ consumer,
+ 100L,
+ 300_000L,
+ time
);
consumerTaskThread = KafkaThread.nonDaemon("RLMMConsumerTask", consumerTask);
}
@@ -110,7 +104,7 @@ public void waitTillConsumptionCatchesUp(RecordMetadata recordMetadata,
log.info("Waiting until consumer is caught up with the target partition: [{}]", partition);
// If the current assignment does not have the subscription for this partition then return immediately.
- if (!consumerTask.isPartitionAssigned(partition)) {
+ if (!consumerTask.isMetadataPartitionAssigned(partition)) {
throw new KafkaException("This consumer is not assigned to the target partition " + partition + ". " +
"Partitions currently assigned: " + consumerTask.metadataPartitionsAssigned());
}
@@ -119,17 +113,17 @@ public void waitTillConsumptionCatchesUp(RecordMetadata recordMetadata,
long startTimeMs = time.milliseconds();
while (true) {
log.debug("Checking if partition [{}] is up to date with offset [{}]", partition, offset);
- long receivedOffset = consumerTask.receivedOffsetForPartition(partition).orElse(-1L);
- if (receivedOffset >= offset) {
+ long readOffset = consumerTask.readOffsetForMetadataPartition(partition).orElse(-1L);
+ if (readOffset >= offset) {
return;
}
log.debug("Expected offset [{}] for partition [{}], but the committed offset: [{}], Sleeping for [{}] to retry again",
- offset, partition, receivedOffset, consumeCheckIntervalMs);
+ offset, partition, readOffset, consumeCheckIntervalMs);
if (time.milliseconds() - startTimeMs > timeoutMs) {
log.warn("Expected offset for partition:[{}] is : [{}], but the committed offset: [{}] ",
- partition, receivedOffset, offset);
+ partition, readOffset, offset);
throw new TimeoutException("Timed out in catching up with the expected offset by consumer.");
}
@@ -158,7 +152,7 @@ public void removeAssignmentsForPartitions(Set partitions) {
consumerTask.removeAssignmentsForPartitions(partitions);
}
- public Optional receivedOffsetForPartition(int metadataPartition) {
- return consumerTask.receivedOffsetForPartition(metadataPartition);
+ public Optional readOffsetForPartition(int metadataPartition) {
+ return consumerTask.readOffsetForMetadataPartition(metadataPartition);
}
}
diff --git a/storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/ConsumerTask.java b/storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/ConsumerTask.java
index 2c95bf399a52d..b580775d1c816 100644
--- a/storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/ConsumerTask.java
+++ b/storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/ConsumerTask.java
@@ -16,12 +16,13 @@
*/
package org.apache.kafka.server.log.remote.metadata.storage;
+import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
-import org.apache.kafka.clients.consumer.KafkaConsumer;
-import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.TopicIdPartition;
import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.WakeupException;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.server.log.remote.metadata.storage.serialization.RemoteLogMetadataSerde;
import org.apache.kafka.server.log.remote.storage.RemoteLogMetadata;
@@ -30,8 +31,6 @@
import org.slf4j.LoggerFactory;
import java.io.Closeable;
-import java.io.IOException;
-import java.nio.file.Path;
import java.time.Duration;
import java.util.Collections;
import java.util.HashMap;
@@ -64,302 +63,403 @@
class ConsumerTask implements Runnable, Closeable {
private static final Logger log = LoggerFactory.getLogger(ConsumerTask.class);
- private static final long POLL_INTERVAL_MS = 100L;
-
private final RemoteLogMetadataSerde serde = new RemoteLogMetadataSerde();
- private final KafkaConsumer consumer;
- private final String metadataTopicName;
+ private final Consumer consumer;
private final RemotePartitionMetadataEventHandler remotePartitionMetadataEventHandler;
private final RemoteLogMetadataTopicPartitioner topicPartitioner;
+ // The timeout for the consumer to poll records from the remote log metadata topic.
+ private final long pollTimeoutMs;
private final Time time;
- // It indicates whether the closing process has been started or not. If it is set as true,
- // consumer will stop consuming messages, and it will not allow partition assignments to be updated.
- private volatile boolean closing = false;
-
- // It indicates whether the consumer needs to assign the partitions or not. This is set when it is
- // determined that the consumer needs to be assigned with the updated partitions.
- private volatile boolean assignPartitions = false;
+ // It indicates whether the ConsumerTask is closed or not.
+ private volatile boolean isClosed = false;
+ // It indicates whether the user topic partition assignment to the consumer has changed or not. If the assignment
+ // has changed, the consumer will eventually start tracking the newly assigned partitions and stop tracking the
+ // ones it is no longer assigned to.
+ // The initial value is set to true to wait for partition assignment on the first execution; otherwise thread will
+ // be busy without actually doing anything
+ private volatile boolean hasAssignmentChanged = true;
// It represents a lock for any operations related to the assignedTopicPartitions.
private final Object assignPartitionsLock = new Object();
// Remote log metadata topic partitions that consumer is assigned to.
- private volatile Set assignedMetaPartitions = Collections.emptySet();
+ private volatile Set assignedMetadataPartitions = Collections.emptySet();
// User topic partitions that this broker is a leader/follower for.
- private Set assignedTopicPartitions = Collections.emptySet();
+ private volatile Map assignedUserTopicIdPartitions = Collections.emptyMap();
+ private volatile Set processedAssignmentOfUserTopicIdPartitions = Collections.emptySet();
- // Map of remote log metadata topic partition to consumed offsets. Received consumer records
- // may or may not have been processed based on the assigned topic partitions.
- private final Map partitionToConsumedOffsets = new ConcurrentHashMap<>();
+ private long uninitializedAt;
+ private boolean isAllUserTopicPartitionsInitialized;
- // Map of remote log metadata topic partition to processed offsets that were synced in committedOffsetsFile.
- private Map lastSyncedPartitionToConsumedOffsets = Collections.emptyMap();
+ // Map of remote log metadata topic partition to consumed offsets.
+ private final Map readOffsetsByMetadataPartition = new ConcurrentHashMap<>();
+ private final Map readOffsetsByUserTopicPartition = new HashMap<>();
- private final long committedOffsetSyncIntervalMs;
- private CommittedOffsetsFile committedOffsetsFile;
- private long lastSyncedTimeMs;
+ private Map offsetHolderByMetadataPartition = new HashMap<>();
+ private boolean hasLastOffsetsFetchFailed = false;
+ private long lastFailedFetchOffsetsTimestamp;
+ // The interval between retries to fetch the start and end offsets for the metadata partitions after a failed fetch.
+ private final long offsetFetchRetryIntervalMs;
- public ConsumerTask(KafkaConsumer consumer,
- String metadataTopicName,
- RemotePartitionMetadataEventHandler remotePartitionMetadataEventHandler,
+ public ConsumerTask(RemotePartitionMetadataEventHandler remotePartitionMetadataEventHandler,
RemoteLogMetadataTopicPartitioner topicPartitioner,
- Path committedOffsetsPath,
- Time time,
- long committedOffsetSyncIntervalMs) {
- this.consumer = Objects.requireNonNull(consumer);
- this.metadataTopicName = Objects.requireNonNull(metadataTopicName);
+ Consumer consumer,
+ long pollTimeoutMs,
+ long offsetFetchRetryIntervalMs,
+ Time time) {
+ this.consumer = consumer;
this.remotePartitionMetadataEventHandler = Objects.requireNonNull(remotePartitionMetadataEventHandler);
this.topicPartitioner = Objects.requireNonNull(topicPartitioner);
+ this.pollTimeoutMs = pollTimeoutMs;
+ this.offsetFetchRetryIntervalMs = offsetFetchRetryIntervalMs;
this.time = Objects.requireNonNull(time);
- this.committedOffsetSyncIntervalMs = committedOffsetSyncIntervalMs;
-
- initializeConsumerAssignment(committedOffsetsPath);
- }
-
- private void initializeConsumerAssignment(Path committedOffsetsPath) {
- try {
- committedOffsetsFile = new CommittedOffsetsFile(committedOffsetsPath.toFile());
- } catch (IOException e) {
- throw new KafkaException(e);
- }
-
- Map committedOffsets = Collections.emptyMap();
- try {
- // Load committed offset and assign them in the consumer.
- committedOffsets = committedOffsetsFile.readEntries();
- } catch (IOException e) {
- // Ignore the error and consumer consumes from the earliest offset.
- log.error("Encountered error while building committed offsets from the file. " +
- "Consumer will consume from the earliest offset for the assigned partitions.", e);
- }
-
- if (!committedOffsets.isEmpty()) {
- // Assign topic partitions from the earlier committed offsets file.
- Set earlierAssignedPartitions = committedOffsets.keySet();
- assignedMetaPartitions = Collections.unmodifiableSet(earlierAssignedPartitions);
- Set metadataTopicPartitions = earlierAssignedPartitions.stream()
- .map(x -> new TopicPartition(REMOTE_LOG_METADATA_TOPIC_NAME, x))
- .collect(Collectors.toSet());
- consumer.assign(metadataTopicPartitions);
-
- // Seek to the committed offsets
- for (Map.Entry entry : committedOffsets.entrySet()) {
- log.debug("Updating consumed offset: [{}] for partition [{}]", entry.getValue(), entry.getKey());
- partitionToConsumedOffsets.put(entry.getKey(), entry.getValue());
- consumer.seek(new TopicPartition(REMOTE_LOG_METADATA_TOPIC_NAME, entry.getKey()), entry.getValue());
- }
-
- lastSyncedPartitionToConsumedOffsets = Collections.unmodifiableMap(committedOffsets);
- }
+ this.uninitializedAt = time.milliseconds();
}
@Override
public void run() {
- log.info("Started Consumer task thread.");
- lastSyncedTimeMs = time.milliseconds();
- try {
- while (!closing) {
- maybeWaitForPartitionsAssignment();
+ log.info("Starting consumer task thread.");
+ while (!isClosed) {
+ try {
+ if (hasAssignmentChanged) {
+ maybeWaitForPartitionAssignments();
+ }
log.trace("Polling consumer to receive remote log metadata topic records");
- ConsumerRecords consumerRecords = consumer.poll(Duration.ofMillis(POLL_INTERVAL_MS));
+ final ConsumerRecords consumerRecords = consumer.poll(Duration.ofMillis(pollTimeoutMs));
for (ConsumerRecord record : consumerRecords) {
processConsumerRecord(record);
}
-
- maybeSyncCommittedDataAndOffsets(false);
+ maybeMarkUserPartitionsAsReady();
+ } catch (final WakeupException ex) {
+ // ignore logging the error
+ isClosed = true;
+ } catch (final RetriableException ex) {
+ log.warn("Retriable error occurred while processing the records. Retrying...", ex);
+ } catch (final Exception ex) {
+ isClosed = true;
+ log.error("Error occurred while processing the records", ex);
}
- } catch (Exception e) {
- log.error("Error occurred in consumer task, close:[{}]", closing, e);
- } finally {
- maybeSyncCommittedDataAndOffsets(true);
- closeConsumer();
- log.info("Exiting from consumer task thread");
}
+ try {
+ consumer.close(Duration.ofSeconds(30));
+ } catch (final Exception e) {
+ log.error("Error encountered while closing the consumer", e);
+ }
+ log.info("Exited from consumer task thread");
}
private void processConsumerRecord(ConsumerRecord record) {
- // Taking assignPartitionsLock here as updateAssignmentsForPartitions changes assignedTopicPartitions
- // and also calls remotePartitionMetadataEventHandler.clearTopicPartition(removedPartition) for the removed
- // partitions.
- RemoteLogMetadata remoteLogMetadata = serde.deserialize(record.value());
- synchronized (assignPartitionsLock) {
- if (assignedTopicPartitions.contains(remoteLogMetadata.topicIdPartition())) {
- remotePartitionMetadataEventHandler.handleRemoteLogMetadata(remoteLogMetadata);
- } else {
- log.debug("This event {} is skipped as the topic partition is not assigned for this instance.", remoteLogMetadata);
- }
- log.debug("Updating consumed offset: [{}] for partition [{}]", record.offset(), record.partition());
- partitionToConsumedOffsets.put(record.partition(), record.offset());
+ final RemoteLogMetadata remoteLogMetadata = serde.deserialize(record.value());
+ if (shouldProcess(remoteLogMetadata, record.offset())) {
+ remotePartitionMetadataEventHandler.handleRemoteLogMetadata(remoteLogMetadata);
+ readOffsetsByUserTopicPartition.put(remoteLogMetadata.topicIdPartition(), record.offset());
+ } else {
+ log.debug("The event {} is skipped because it is either already processed or not assigned to this consumer", remoteLogMetadata);
}
+ log.debug("Updating consumed offset: [{}] for partition [{}]", record.offset(), record.partition());
+ readOffsetsByMetadataPartition.put(record.partition(), record.offset());
}
- private void maybeSyncCommittedDataAndOffsets(boolean forceSync) {
- // Return immediately if there is no consumption from last time.
- boolean noConsumedOffsetUpdates = partitionToConsumedOffsets.equals(lastSyncedPartitionToConsumedOffsets);
- if (noConsumedOffsetUpdates || !forceSync && time.milliseconds() - lastSyncedTimeMs < committedOffsetSyncIntervalMs) {
- log.debug("Skip syncing committed offsets, noConsumedOffsetUpdates: {}, forceSync: {}", noConsumedOffsetUpdates, forceSync);
+ private boolean shouldProcess(final RemoteLogMetadata metadata, final long recordOffset) {
+ final TopicIdPartition tpId = metadata.topicIdPartition();
+ final Long readOffset = readOffsetsByUserTopicPartition.get(tpId);
+ return processedAssignmentOfUserTopicIdPartitions.contains(tpId) && (readOffset == null || readOffset < recordOffset);
+ }
+
+ private void maybeMarkUserPartitionsAsReady() {
+ if (isAllUserTopicPartitionsInitialized) {
return;
}
-
- try {
- // Need to take lock on assignPartitionsLock as assignedTopicPartitions might
- // get updated by other threads.
- synchronized (assignPartitionsLock) {
- for (TopicIdPartition topicIdPartition : assignedTopicPartitions) {
- int metadataPartition = topicPartitioner.metadataPartition(topicIdPartition);
- Long offset = partitionToConsumedOffsets.get(metadataPartition);
- if (offset != null) {
- remotePartitionMetadataEventHandler.syncLogMetadataSnapshot(topicIdPartition, metadataPartition, offset);
+ maybeFetchStartAndEndOffsets();
+ boolean isAllInitialized = true;
+ for (final UserTopicIdPartition utp : assignedUserTopicIdPartitions.values()) {
+ if (utp.isAssigned && !utp.isInitialized) {
+ final Integer metadataPartition = utp.metadataPartition;
+ final StartAndEndOffsetHolder holder = offsetHolderByMetadataPartition.get(toRemoteLogPartition(metadataPartition));
+ // The offset-holder can be null, when the recent assignment wasn't picked up by the consumer.
+ if (holder != null) {
+ final Long readOffset = readOffsetsByMetadataPartition.getOrDefault(metadataPartition, -1L);
+ // 1) The end-offset was fetched only once during reassignment. The metadata-partition can receive
+ // new stream of records, so the consumer can read records more than the last-fetched end-offset.
+ // 2) When the internal topic becomes empty due to breach by size/time/start-offset, then there
+ // are no records to read.
+ if (readOffset + 1 >= holder.endOffset || holder.endOffset.equals(holder.startOffset)) {
+ markInitialized(utp);
} else {
- log.debug("Skipping sync-up of the remote-log-metadata-file for partition: [{}] , with remote log metadata partition{}, and no offset",
- topicIdPartition, metadataPartition);
+ log.debug("The user-topic-partition {} could not be marked initialized since the read-offset is {} " +
+ "but the end-offset is {} for the metadata-partition {}", utp, readOffset, holder.endOffset,
+ metadataPartition);
}
+ } else {
+ log.debug("The offset-holder is null for the metadata-partition {}. The consumer may not have picked" +
+ " up the recent assignment", metadataPartition);
}
-
- // Write partitionToConsumedOffsets into committed offsets file as we do not want to process them again
- // in case of restarts.
- committedOffsetsFile.writeEntries(partitionToConsumedOffsets);
- lastSyncedPartitionToConsumedOffsets = new HashMap<>(partitionToConsumedOffsets);
}
-
- lastSyncedTimeMs = time.milliseconds();
- } catch (IOException e) {
- throw new KafkaException("Error encountered while writing committed offsets to a local file", e);
+ isAllInitialized = isAllInitialized && utp.isInitialized;
}
- }
-
- private void closeConsumer() {
- log.info("Closing the consumer instance");
- try {
- consumer.close(Duration.ofSeconds(30));
- } catch (Exception e) {
- log.error("Error encountered while closing the consumer", e);
+ if (isAllInitialized) {
+ log.info("Initialized for all the {} assigned user-partitions mapped to the {} meta-partitions in {} ms",
+ assignedUserTopicIdPartitions.size(), assignedMetadataPartitions.size(),
+ time.milliseconds() - uninitializedAt);
}
+ isAllUserTopicPartitionsInitialized = isAllInitialized;
}
- private void maybeWaitForPartitionsAssignment() {
- Set assignedMetaPartitionsSnapshot = Collections.emptySet();
+ void maybeWaitForPartitionAssignments() throws InterruptedException {
+ // Snapshots of the metadata-partition and user-topic-partition are used to reduce the scope of the
+ // synchronization block.
+ // 1) LEADER_AND_ISR and STOP_REPLICA requests adds / removes the user-topic-partitions from the request
+ // handler threads. Those threads should not be blocked for a long time, therefore scope of the
+ // synchronization block is reduced to bare minimum.
+ // 2) Note that the consumer#position, consumer#seekToBeginning, consumer#seekToEnd and the other consumer APIs
+ // response times are un-predictable. Those should not be kept in the synchronization block.
+ final Set metadataPartitionSnapshot = new HashSet<>();
+ final Set assignedUserTopicIdPartitionsSnapshot = new HashSet<>();
synchronized (assignPartitionsLock) {
- // If it is closing, return immediately. This should be inside the assignPartitionsLock as the closing is updated
- // in close() method with in the same lock to avoid any race conditions.
- if (closing) {
- return;
+ while (!isClosed && assignedUserTopicIdPartitions.isEmpty()) {
+ log.debug("Waiting for remote log metadata partitions to be assigned");
+ assignPartitionsLock.wait();
}
-
- while (assignedMetaPartitions.isEmpty()) {
- // If no partitions are assigned, wait until they are assigned.
- log.debug("Waiting for assigned remote log metadata partitions..");
- try {
- // No timeout is set here, as it is always notified. Even when it is closed, the race can happen
- // between the thread calling this method and the thread calling close(). We should have a check
- // for closing as that might have been set and notified with assignPartitionsLock by `close`
- // method.
- assignPartitionsLock.wait();
-
- if (closing) {
- return;
- }
- } catch (InterruptedException e) {
- throw new KafkaException(e);
- }
- }
-
- if (assignPartitions) {
- assignedMetaPartitionsSnapshot = new HashSet<>(assignedMetaPartitions);
- // Removing unassigned meta partitions from partitionToConsumedOffsets and partitionToCommittedOffsets
- partitionToConsumedOffsets.entrySet().removeIf(entry -> !assignedMetaPartitions.contains(entry.getKey()));
-
- assignPartitions = false;
+ if (!isClosed && hasAssignmentChanged) {
+ assignedUserTopicIdPartitions.values().forEach(utp -> {
+ metadataPartitionSnapshot.add(utp.metadataPartition);
+ assignedUserTopicIdPartitionsSnapshot.add(utp);
+ });
+ hasAssignmentChanged = false;
}
}
-
- if (!assignedMetaPartitionsSnapshot.isEmpty()) {
- executeReassignment(assignedMetaPartitionsSnapshot);
+ if (!metadataPartitionSnapshot.isEmpty()) {
+ final Set remoteLogPartitions = toRemoteLogPartitions(metadataPartitionSnapshot);
+ consumer.assign(remoteLogPartitions);
+ this.assignedMetadataPartitions = Collections.unmodifiableSet(metadataPartitionSnapshot);
+ // for newly assigned user-partitions, read from the beginning of the corresponding metadata partition
+ final Set seekToBeginOffsetPartitions = assignedUserTopicIdPartitionsSnapshot
+ .stream()
+ .filter(utp -> !utp.isAssigned)
+ .map(utp -> toRemoteLogPartition(utp.metadataPartition))
+ .collect(Collectors.toSet());
+ consumer.seekToBeginning(seekToBeginOffsetPartitions);
+ // for other metadata partitions, read from the offset where the processing left last time.
+ remoteLogPartitions.stream()
+ .filter(tp -> !seekToBeginOffsetPartitions.contains(tp) &&
+ readOffsetsByMetadataPartition.containsKey(tp.partition()))
+ .forEach(tp -> consumer.seek(tp, readOffsetsByMetadataPartition.get(tp.partition())));
+ Set processedAssignmentPartitions = new HashSet<>();
+ // mark all the user-topic-partitions as assigned to the consumer.
+ assignedUserTopicIdPartitionsSnapshot.forEach(utp -> {
+ if (!utp.isAssigned) {
+ // Note that there can be a race between `remove` and `add` partition assignment. Calling the
+ // `maybeLoadPartition` here again to be sure that the partition gets loaded on the handler.
+ remotePartitionMetadataEventHandler.maybeLoadPartition(utp.topicIdPartition);
+ utp.isAssigned = true;
+ }
+ processedAssignmentPartitions.add(utp.topicIdPartition);
+ });
+ processedAssignmentOfUserTopicIdPartitions = new HashSet<>(processedAssignmentPartitions);
+ clearResourcesForUnassignedUserTopicPartitions(processedAssignmentPartitions);
+ isAllUserTopicPartitionsInitialized = false;
+ uninitializedAt = time.milliseconds();
+ fetchStartAndEndOffsets();
}
}
- private void executeReassignment(Set assignedMetaPartitionsSnapshot) {
- Set assignedMetaTopicPartitions =
- assignedMetaPartitionsSnapshot.stream()
- .map(partitionNum -> new TopicPartition(REMOTE_LOG_METADATA_TOPIC_NAME, partitionNum))
- .collect(Collectors.toSet());
- log.info("Reassigning partitions to consumer task [{}]", assignedMetaTopicPartitions);
- consumer.assign(assignedMetaTopicPartitions);
+ private void clearResourcesForUnassignedUserTopicPartitions(Set assignedPartitions) {
+ // Note that there can be previously assigned user-topic-partitions where no records are there to read
+ // (eg) none of the segments for a partition were uploaded. Those partition resources won't be cleared.
+ // It can be fixed later when required since they are empty resources.
+ Set unassignedPartitions = readOffsetsByUserTopicPartition.keySet()
+ .stream()
+ .filter(e -> !assignedPartitions.contains(e))
+ .collect(Collectors.toSet());
+ unassignedPartitions.forEach(unassignedPartition -> {
+ remotePartitionMetadataEventHandler.clearTopicPartition(unassignedPartition);
+ readOffsetsByUserTopicPartition.remove(unassignedPartition);
+ });
+ log.info("Unassigned user-topic-partitions: {}", unassignedPartitions.size());
+ }
+
+ public void addAssignmentsForPartitions(final Set partitions) {
+ updateAssignments(Objects.requireNonNull(partitions), Collections.emptySet());
+ }
+
+ public void removeAssignmentsForPartitions(final Set partitions) {
+ updateAssignments(Collections.emptySet(), Objects.requireNonNull(partitions));
}
- public void addAssignmentsForPartitions(Set partitions) {
- updateAssignmentsForPartitions(partitions, Collections.emptySet());
+ private void updateAssignments(final Set addedPartitions,
+ final Set removedPartitions) {
+ log.info("Updating assignments for partitions added: {} and removed: {}", addedPartitions, removedPartitions);
+ if (!addedPartitions.isEmpty() || !removedPartitions.isEmpty()) {
+ synchronized (assignPartitionsLock) {
+ // Make a copy of the existing assignments and update the copy.
+ final Map updatedUserPartitions = new HashMap<>(assignedUserTopicIdPartitions);
+ addedPartitions.forEach(tpId -> updatedUserPartitions.putIfAbsent(tpId, newUserTopicIdPartition(tpId)));
+ removedPartitions.forEach(updatedUserPartitions::remove);
+ if (!updatedUserPartitions.equals(assignedUserTopicIdPartitions)) {
+ assignedUserTopicIdPartitions = Collections.unmodifiableMap(updatedUserPartitions);
+ hasAssignmentChanged = true;
+ log.debug("Assigned user-topic-partitions: {}", assignedUserTopicIdPartitions);
+ assignPartitionsLock.notifyAll();
+ }
+ }
+ }
}
- public void removeAssignmentsForPartitions(Set partitions) {
- updateAssignmentsForPartitions(Collections.emptySet(), partitions);
+ public Optional readOffsetForMetadataPartition(final int partition) {
+ return Optional.ofNullable(readOffsetsByMetadataPartition.get(partition));
}
- private void updateAssignmentsForPartitions(Set addedPartitions,
- Set removedPartitions) {
- log.info("Updating assignments for addedPartitions: {} and removedPartition: {}", addedPartitions, removedPartitions);
+ public boolean isMetadataPartitionAssigned(final int partition) {
+ return assignedMetadataPartitions.contains(partition);
+ }
- Objects.requireNonNull(addedPartitions, "addedPartitions must not be null");
- Objects.requireNonNull(removedPartitions, "removedPartitions must not be null");
+ public boolean isUserPartitionAssigned(final TopicIdPartition partition) {
+ final UserTopicIdPartition utp = assignedUserTopicIdPartitions.get(partition);
+ return utp != null && utp.isAssigned;
+ }
- if (addedPartitions.isEmpty() && removedPartitions.isEmpty()) {
- return;
+ @Override
+ public void close() {
+ if (!isClosed) {
+ log.info("Closing the instance");
+ synchronized (assignPartitionsLock) {
+ isClosed = true;
+ assignedUserTopicIdPartitions.values().forEach(this::markInitialized);
+ consumer.wakeup();
+ assignPartitionsLock.notifyAll();
+ }
}
+ }
- synchronized (assignPartitionsLock) {
- Set updatedReassignedPartitions = new HashSet<>(assignedTopicPartitions);
- updatedReassignedPartitions.addAll(addedPartitions);
- updatedReassignedPartitions.removeAll(removedPartitions);
- Set updatedAssignedMetaPartitions = new HashSet<>();
- for (TopicIdPartition tp : updatedReassignedPartitions) {
- updatedAssignedMetaPartitions.add(topicPartitioner.metadataPartition(tp));
- }
+ public Set metadataPartitionsAssigned() {
+ return Collections.unmodifiableSet(assignedMetadataPartitions);
+ }
+
+ private void fetchStartAndEndOffsets() {
+ try {
+ final Set uninitializedPartitions = assignedUserTopicIdPartitions.values().stream()
+ .filter(utp -> utp.isAssigned && !utp.isInitialized)
+ .map(utp -> toRemoteLogPartition(utp.metadataPartition))
+ .collect(Collectors.toSet());
+ // Removing the previous offset holder if it exists. During reassignment, if the list-offset
+ // call to `earliest` and `latest` offset fails, then we should not use the previous values.
+ uninitializedPartitions.forEach(tp -> offsetHolderByMetadataPartition.remove(tp));
+ if (!uninitializedPartitions.isEmpty()) {
+ Map endOffsets = consumer.endOffsets(uninitializedPartitions);
+ Map startOffsets = consumer.beginningOffsets(uninitializedPartitions);
+ offsetHolderByMetadataPartition = endOffsets.entrySet()
+ .stream()
+ .collect(Collectors.toMap(Map.Entry::getKey,
+ e -> new StartAndEndOffsetHolder(startOffsets.get(e.getKey()), e.getValue())));
- // Clear removed topic partitions from in-memory cache.
- for (TopicIdPartition removedPartition : removedPartitions) {
- remotePartitionMetadataEventHandler.clearTopicPartition(removedPartition);
}
+ hasLastOffsetsFetchFailed = false;
+ } catch (final RetriableException ex) {
+ // ignore LEADER_NOT_AVAILABLE error, this can happen when the partition leader is not yet assigned.
+ hasLastOffsetsFetchFailed = true;
+ lastFailedFetchOffsetsTimestamp = time.milliseconds();
+ }
+ }
- assignedTopicPartitions = Collections.unmodifiableSet(updatedReassignedPartitions);
- log.debug("Assigned topic partitions: {}", assignedTopicPartitions);
+ private void maybeFetchStartAndEndOffsets() {
+ // If the leader for a `__remote_log_metadata` partition is not available, then the call to `ListOffsets`
+ // will fail after the default timeout of 1 min. Added a delay between the retries to prevent the thread from
+ // aggressively fetching the list offsets. During this time, the recently reassigned user-topic-partitions
+ // won't be marked as initialized.
+ if (hasLastOffsetsFetchFailed && lastFailedFetchOffsetsTimestamp + offsetFetchRetryIntervalMs < time.milliseconds()) {
+ fetchStartAndEndOffsets();
+ }
+ }
- if (!updatedAssignedMetaPartitions.equals(assignedMetaPartitions)) {
- assignedMetaPartitions = Collections.unmodifiableSet(updatedAssignedMetaPartitions);
- log.debug("Assigned metadata topic partitions: {}", assignedMetaPartitions);
+ private UserTopicIdPartition newUserTopicIdPartition(final TopicIdPartition tpId) {
+ return new UserTopicIdPartition(tpId, topicPartitioner.metadataPartition(tpId));
+ }
- assignPartitions = true;
- assignPartitionsLock.notifyAll();
- } else {
- log.debug("No change in assigned metadata topic partitions: {}", assignedMetaPartitions);
- }
+ private void markInitialized(final UserTopicIdPartition utp) {
+ // Silently not initialize the utp
+ if (!utp.isAssigned) {
+ log.warn("Tried to initialize a UTP: {} that was not yet assigned!", utp);
+ return;
+ }
+ if (!utp.isInitialized) {
+ remotePartitionMetadataEventHandler.markInitialized(utp.topicIdPartition);
+ utp.isInitialized = true;
}
}
- public Optional receivedOffsetForPartition(int partition) {
- return Optional.ofNullable(partitionToConsumedOffsets.get(partition));
+ static Set toRemoteLogPartitions(final Set partitions) {
+ return partitions.stream()
+ .map(ConsumerTask::toRemoteLogPartition)
+ .collect(Collectors.toSet());
}
- public boolean isPartitionAssigned(int partition) {
- return assignedMetaPartitions.contains(partition);
+ static TopicPartition toRemoteLogPartition(int partition) {
+ return new TopicPartition(REMOTE_LOG_METADATA_TOPIC_NAME, partition);
}
- public void close() {
- if (!closing) {
- synchronized (assignPartitionsLock) {
- // Closing should be updated only after acquiring the lock to avoid race in
- // maybeWaitForPartitionsAssignment() where it waits on assignPartitionsLock. It should not wait
- // if the closing is already set.
- closing = true;
- consumer.wakeup();
- assignPartitionsLock.notifyAll();
- }
+ static class UserTopicIdPartition {
+ private final TopicIdPartition topicIdPartition;
+ private final Integer metadataPartition;
+ // The `utp` will be initialized once it reads all the existing events from the remote log metadata topic.
+ boolean isInitialized;
+ // denotes whether this `utp` is assigned to the consumer
+ boolean isAssigned;
+
+ /**
+ * UserTopicIdPartition denotes the user topic-partitions for which this broker acts as a leader/follower of.
+ *
+ * @param tpId the unique topic partition identifier
+ * @param metadataPartition the remote log metadata partition mapped for this user-topic-partition.
+ */
+ public UserTopicIdPartition(final TopicIdPartition tpId, final Integer metadataPartition) {
+ this.topicIdPartition = Objects.requireNonNull(tpId);
+ this.metadataPartition = Objects.requireNonNull(metadataPartition);
+ this.isInitialized = false;
+ this.isAssigned = false;
+ }
+
+ @Override
+ public String toString() {
+ return "UserTopicIdPartition{" +
+ "topicIdPartition=" + topicIdPartition +
+ ", metadataPartition=" + metadataPartition +
+ ", isInitialized=" + isInitialized +
+ ", isAssigned=" + isAssigned +
+ '}';
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ UserTopicIdPartition that = (UserTopicIdPartition) o;
+ return topicIdPartition.equals(that.topicIdPartition) && metadataPartition.equals(that.metadataPartition);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(topicIdPartition, metadataPartition);
}
}
- public Set metadataPartitionsAssigned() {
- return Collections.unmodifiableSet(assignedMetaPartitions);
+ static class StartAndEndOffsetHolder {
+ Long startOffset;
+ Long endOffset;
+
+ public StartAndEndOffsetHolder(Long startOffset, Long endOffset) {
+ this.startOffset = startOffset;
+ this.endOffset = endOffset;
+ }
+
+ @Override
+ public String toString() {
+ return "StartAndEndOffsetHolder{" +
+ "startOffset=" + startOffset +
+ ", endOffset=" + endOffset +
+ '}';
+ }
}
-}
+}
\ No newline at end of file
diff --git a/storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/RemoteLogMetadataCache.java b/storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/RemoteLogMetadataCache.java
index 8c89df3df2c8f..d1f15325eb900 100644
--- a/storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/RemoteLogMetadataCache.java
+++ b/storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/RemoteLogMetadataCache.java
@@ -32,6 +32,8 @@
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
/**
* This class provides an in-memory cache of remote log segment metadata. This maintains the lineage of segments
@@ -104,6 +106,22 @@ public class RemoteLogMetadataCache {
// https://issues.apache.org/jira/browse/KAFKA-12641
protected final ConcurrentMap leaderEpochEntries = new ConcurrentHashMap<>();
+ private final CountDownLatch initializedLatch = new CountDownLatch(1);
+
+ public void markInitialized() {
+ initializedLatch.countDown();
+ }
+
+ public void ensureInitialized() throws InterruptedException {
+ if (!initializedLatch.await(2, TimeUnit.MINUTES)) {
+ throw new InterruptedException();
+ }
+ }
+
+ public boolean isInitialized() {
+ return initializedLatch.getCount() == 0;
+ }
+
/**
* Returns {@link RemoteLogSegmentMetadata} if it exists for the given leader-epoch containing the offset and with
* {@link RemoteLogSegmentState#COPY_SEGMENT_FINISHED} state, else returns {@link Optional#empty()}.
diff --git a/storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/RemotePartitionMetadataEventHandler.java b/storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/RemotePartitionMetadataEventHandler.java
index c92a51ecacac6..f4f43b0d88377 100644
--- a/storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/RemotePartitionMetadataEventHandler.java
+++ b/storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/RemotePartitionMetadataEventHandler.java
@@ -50,4 +50,9 @@ public abstract void syncLogMetadataSnapshot(TopicIdPartition topicIdPartition,
public abstract void clearTopicPartition(TopicIdPartition topicIdPartition);
+ public abstract void markInitialized(TopicIdPartition partition);
+
+ public abstract boolean isInitialized(TopicIdPartition partition);
+
+ public abstract void maybeLoadPartition(TopicIdPartition partition);
}
\ No newline at end of file
diff --git a/storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/RemotePartitionMetadataStore.java b/storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/RemotePartitionMetadataStore.java
index 7051d184aad5b..5cecd3023564d 100644
--- a/storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/RemotePartitionMetadataStore.java
+++ b/storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/RemotePartitionMetadataStore.java
@@ -150,6 +150,11 @@ private FileBasedRemoteLogMetadataCache getRemoteLogMetadataCache(TopicIdPartiti
if (remoteLogMetadataCache == null) {
throw new RemoteResourceNotFoundException("No resource found for partition: " + topicIdPartition);
}
+ try {
+ remoteLogMetadataCache.ensureInitialized();
+ } catch (InterruptedException e) {
+ throw new RemoteResourceNotFoundException("Couldn't initialize remote log metadata cache for partition: " + topicIdPartition);
+ }
return remoteLogMetadataCache;
}
@@ -180,9 +185,21 @@ public void close() throws IOException {
idToRemoteLogMetadataCache = Collections.emptyMap();
}
+ @Override
public void maybeLoadPartition(TopicIdPartition partition) {
idToRemoteLogMetadataCache.computeIfAbsent(partition,
topicIdPartition -> new FileBasedRemoteLogMetadataCache(topicIdPartition, partitionLogDirectory(topicIdPartition.topicPartition())));
}
+ @Override
+ public void markInitialized(TopicIdPartition partition) {
+ idToRemoteLogMetadataCache.get(partition).markInitialized();
+ log.trace("Remote log components are initialized for user-partition: {}", partition);
+ }
+
+ @Override
+ public boolean isInitialized(TopicIdPartition topicIdPartition) {
+ RemoteLogMetadataCache metadataCache = idToRemoteLogMetadataCache.get(topicIdPartition);
+ return metadataCache != null && metadataCache.isInitialized();
+ }
}
diff --git a/storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManager.java b/storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManager.java
index 4b02b9b676367..e1bf145bbd888 100644
--- a/storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManager.java
+++ b/storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManager.java
@@ -84,7 +84,7 @@ public class TopicBasedRemoteLogMetadataManager implements RemoteLogMetadataMana
private RemotePartitionMetadataStore remotePartitionMetadataStore;
private volatile TopicBasedRemoteLogMetadataManagerConfig rlmmConfig;
- private volatile RemoteLogMetadataTopicPartitioner rlmmTopicPartitioner;
+ private volatile RemoteLogMetadataTopicPartitioner rlmTopicPartitioner;
private final Set pendingAssignPartitions = Collections.synchronizedSet(new HashSet<>());
private volatile boolean initializationFailed;
@@ -260,12 +260,12 @@ public Iterator listRemoteLogSegments(TopicIdPartition
}
public int metadataPartition(TopicIdPartition topicIdPartition) {
- return rlmmTopicPartitioner.metadataPartition(topicIdPartition);
+ return rlmTopicPartitioner.metadataPartition(topicIdPartition);
}
// Visible For Testing
- public Optional receivedOffsetForPartition(int metadataPartition) {
- return consumerManager.receivedOffsetForPartition(metadataPartition);
+ public Optional readOffsetForPartition(int metadataPartition) {
+ return consumerManager.readOffsetForPartition(metadataPartition);
}
@Override
@@ -357,7 +357,7 @@ public void configure(Map configs) {
log.info("Started configuring topic-based RLMM with configs: {}", configs);
rlmmConfig = new TopicBasedRemoteLogMetadataManagerConfig(configs);
- rlmmTopicPartitioner = new RemoteLogMetadataTopicPartitioner(rlmmConfig.metadataTopicPartitionsCount());
+ rlmTopicPartitioner = new RemoteLogMetadataTopicPartitioner(rlmmConfig.metadataTopicPartitionsCount());
remotePartitionMetadataStore = new RemotePartitionMetadataStore(new File(rlmmConfig.logDir()).toPath());
configured = true;
log.info("Successfully configured topic-based RLMM with config: {}", rlmmConfig);
@@ -416,8 +416,8 @@ private void initializeResources() {
// Create producer and consumer managers.
lock.writeLock().lock();
try {
- producerManager = new ProducerManager(rlmmConfig, rlmmTopicPartitioner);
- consumerManager = new ConsumerManager(rlmmConfig, remotePartitionMetadataStore, rlmmTopicPartitioner, time);
+ producerManager = new ProducerManager(rlmmConfig, rlmTopicPartitioner);
+ consumerManager = new ConsumerManager(rlmmConfig, remotePartitionMetadataStore, rlmTopicPartitioner, time);
if (startConsumerThread) {
consumerManager.startConsumerThread();
} else {
@@ -509,10 +509,8 @@ public TopicBasedRemoteLogMetadataManagerConfig config() {
}
// Visible for testing.
- public void startConsumerThread() {
- if (consumerManager != null) {
- consumerManager.startConsumerThread();
- }
+ void setRlmTopicPartitioner(RemoteLogMetadataTopicPartitioner rlmTopicPartitioner) {
+ this.rlmTopicPartitioner = Objects.requireNonNull(rlmTopicPartitioner);
}
@Override
diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/epoch/LeaderEpochFileCache.java b/storage/src/main/java/org/apache/kafka/storage/internals/epoch/LeaderEpochFileCache.java
index 55d7be4029af4..4cb7744957960 100644
--- a/storage/src/main/java/org/apache/kafka/storage/internals/epoch/LeaderEpochFileCache.java
+++ b/storage/src/main/java/org/apache/kafka/storage/internals/epoch/LeaderEpochFileCache.java
@@ -27,6 +27,7 @@
import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.NavigableMap;
import java.util.Optional;
import java.util.OptionalInt;
import java.util.TreeMap;
@@ -398,7 +399,6 @@ public void clear() {
}
}
- // Visible for testing
public List epochEntries() {
lock.readLock().lock();
try {
@@ -408,6 +408,19 @@ public List epochEntries() {
}
}
+ public NavigableMap epochWithOffsets() {
+ lock.readLock().lock();
+ try {
+ NavigableMap epochWithOffsets = new TreeMap<>();
+ for (EpochEntry epochEntry : epochs.values()) {
+ epochWithOffsets.put(epochEntry.epoch, epochEntry.startOffset);
+ }
+ return epochWithOffsets;
+ } finally {
+ lock.readLock().unlock();
+ }
+ }
+
private void flushTo(LeaderEpochCheckpoint leaderEpochCheckpoint) {
lock.readLock().lock();
try {
diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java
index ce2880865dd38..f2540b1c30a2a 100644
--- a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java
+++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java
@@ -102,9 +102,9 @@ public String topicWarningMessage(String topicName) {
public static class RemoteLogConfig {
- private final boolean remoteStorageEnable;
- private final long localRetentionMs;
- private final long localRetentionBytes;
+ public final boolean remoteStorageEnable;
+ public final long localRetentionMs;
+ public final long localRetentionBytes;
private RemoteLogConfig(LogConfig config) {
this.remoteStorageEnable = config.getBoolean(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG);
diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java
index 30c4da4ce4440..92ba24e4d3ea5 100644
--- a/storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java
+++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java
@@ -46,6 +46,7 @@
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
+import java.util.Optional;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantReadWriteLock;
@@ -279,7 +280,7 @@ private void init() throws IOException {
TransactionIndex txnIndex = new TransactionIndex(offset, txnIndexFile);
txnIndex.sanityCheck();
- Entry entry = new Entry(offsetIndex, timeIndex, txnIndex);
+ Entry entry = new Entry(offsetIndex, timeIndex, Optional.of(txnIndex));
internalCache.put(uuid, entry);
} else {
// Delete all of them if any one of those indexes is not available for a specific segment id
@@ -319,12 +320,15 @@ private T loadIndexFile(File file, RemoteLogSegmentMetadata remoteLogSegment
if (index == null) {
File tmpIndexFile = new File(indexFile.getParentFile(), indexFile.getName() + RemoteIndexCache.TMP_FILE_SUFFIX);
- try (InputStream inputStream = fetchRemoteIndex.apply(remoteLogSegmentMetadata);) {
- Files.copy(inputStream, tmpIndexFile.toPath(), StandardCopyOption.REPLACE_EXISTING);
- }
+ final InputStream apply = fetchRemoteIndex.apply(remoteLogSegmentMetadata);
+ if (apply != null) {
+ try (InputStream inputStream = apply) {
+ Files.copy(inputStream, tmpIndexFile.toPath(), StandardCopyOption.REPLACE_EXISTING);
+ }
- Utils.atomicMoveWithFallback(tmpIndexFile.toPath(), indexFile.toPath(), false);
- index = readIndex.apply(indexFile);
+ Utils.atomicMoveWithFallback(tmpIndexFile.toPath(), indexFile.toPath(), false);
+ index = readIndex.apply(indexFile);
+ }
}
return index;
@@ -403,7 +407,7 @@ private RemoteIndexCache.Entry createCacheEntry(RemoteLogSegmentMetadata remoteL
}
});
- return new Entry(offsetIndex, timeIndex, txnIndex);
+ return new Entry(offsetIndex, timeIndex, Optional.ofNullable(txnIndex));
} catch (IOException e) {
throw new KafkaException(e);
}
@@ -459,7 +463,7 @@ public static class Entry implements AutoCloseable {
private final OffsetIndex offsetIndex;
private final TimeIndex timeIndex;
- private final TransactionIndex txnIndex;
+ private final Optional txnIndex;
// This lock is used to synchronize cleanup methods and read methods. This ensures that cleanup (which changes the
// underlying files of the index) isn't performed while a read is in-progress for the entry. This is required in
@@ -471,7 +475,7 @@ public static class Entry implements AutoCloseable {
private boolean markedForCleanup = false;
- public Entry(OffsetIndex offsetIndex, TimeIndex timeIndex, TransactionIndex txnIndex) {
+ public Entry(OffsetIndex offsetIndex, TimeIndex timeIndex, Optional txnIndex) {
this.offsetIndex = offsetIndex;
this.timeIndex = timeIndex;
this.txnIndex = txnIndex;
@@ -488,7 +492,7 @@ public TimeIndex timeIndex() {
}
// Visible for testing
- public TransactionIndex txnIndex() {
+ public Optional txnIndex() {
return txnIndex;
}
@@ -531,7 +535,10 @@ public void markForCleanup() throws IOException {
markedForCleanup = true;
offsetIndex.renameTo(new File(Utils.replaceSuffix(offsetIndex.file().getPath(), "", LogFileUtils.DELETED_FILE_SUFFIX)));
timeIndex.renameTo(new File(Utils.replaceSuffix(timeIndex.file().getPath(), "", LogFileUtils.DELETED_FILE_SUFFIX)));
- txnIndex.renameTo(new File(Utils.replaceSuffix(txnIndex.file().getPath(), "", LogFileUtils.DELETED_FILE_SUFFIX)));
+ if (txnIndex.isPresent()) {
+ final TransactionIndex transactionIndex = txnIndex.get();
+ transactionIndex.renameTo(new File(Utils.replaceSuffix(transactionIndex.file().getPath(), "", LogFileUtils.DELETED_FILE_SUFFIX)));
+ }
}
} finally {
lock.writeLock().unlock();
@@ -553,7 +560,9 @@ public void cleanup() throws IOException {
timeIndex.deleteIfExists();
return null;
}, () -> {
- txnIndex.deleteIfExists();
+ if (txnIndex.isPresent()) {
+ txnIndex.get().deleteIfExists();
+ }
return null;
});
@@ -570,7 +579,7 @@ public void close() {
try {
Utils.closeQuietly(offsetIndex, "OffsetIndex");
Utils.closeQuietly(timeIndex, "TimeIndex");
- Utils.closeQuietly(txnIndex, "TransactionIndex");
+ txnIndex.ifPresent(transactionIndex -> Utils.closeQuietly(transactionIndex, "TransactionIndex"));
} finally {
lock.writeLock().unlock();
}
@@ -579,10 +588,10 @@ public void close() {
@Override
public String toString() {
return "Entry{" +
- "offsetIndex=" + offsetIndex.file().getName() +
- ", timeIndex=" + timeIndex.file().getName() +
- ", txnIndex=" + txnIndex.file().getName() +
- '}';
+ "offsetIndex=" + offsetIndex.file().getName() +
+ ", timeIndex=" + timeIndex.file().getName() +
+ ", txnIndex=" + txnIndex.map(t -> t.file().getName()).orElse("<>") +
+ '}';
}
}
diff --git a/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/ConsumerTaskTest.java b/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/ConsumerTaskTest.java
new file mode 100644
index 0000000000000..9161ff470b7e7
--- /dev/null
+++ b/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/ConsumerTaskTest.java
@@ -0,0 +1,414 @@
+/*
+ * 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.server.log.remote.metadata.storage;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.MockConsumer;
+import org.apache.kafka.clients.consumer.OffsetResetStrategy;
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.errors.AuthorizationException;
+import org.apache.kafka.common.errors.LeaderNotAvailableException;
+import org.apache.kafka.common.errors.TimeoutException;
+import org.apache.kafka.common.utils.SystemTime;
+import org.apache.kafka.server.log.remote.metadata.storage.serialization.RemoteLogMetadataSerde;
+import org.apache.kafka.server.log.remote.storage.RemoteLogMetadata;
+import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentId;
+import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentMetadata;
+import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentMetadataUpdate;
+import org.apache.kafka.server.log.remote.storage.RemotePartitionDeleteMetadata;
+import org.apache.kafka.test.TestCondition;
+import org.apache.kafka.test.TestUtils;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.kafka.server.log.remote.metadata.storage.ConsumerTask.UserTopicIdPartition;
+import static org.apache.kafka.server.log.remote.metadata.storage.ConsumerTask.toRemoteLogPartition;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+public class ConsumerTaskTest {
+
+ private final int numMetadataTopicPartitions = 5;
+ private final RemoteLogMetadataTopicPartitioner partitioner = new RemoteLogMetadataTopicPartitioner(numMetadataTopicPartitions);
+ private final DummyEventHandler handler = new DummyEventHandler();
+ private final Set remoteLogPartitions = IntStream.range(0, numMetadataTopicPartitions).boxed()
+ .map(ConsumerTask::toRemoteLogPartition).collect(Collectors.toSet());
+ private final Uuid topicId = Uuid.randomUuid();
+ private final RemoteLogMetadataSerde serde = new RemoteLogMetadataSerde();
+
+ private ConsumerTask consumerTask;
+ private MockConsumer consumer;
+ private Thread thread;
+
+ @BeforeEach
+ public void beforeEach() {
+ final Map offsets = remoteLogPartitions.stream()
+ .collect(Collectors.toMap(Function.identity(), e -> 0L));
+ consumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST);
+ consumer.updateBeginningOffsets(offsets);
+ consumerTask = new ConsumerTask(handler, partitioner, consumer, 10L, 300_000L, new SystemTime());
+ thread = new Thread(consumerTask);
+ }
+
+ @AfterEach
+ public void afterEach() throws InterruptedException {
+ if (thread != null) {
+ consumerTask.close();
+ thread.join(10_000);
+ }
+ }
+
+ /**
+ * Tests that the consumer task shuts down gracefully when there were no assignments.
+ */
+ @Test
+ public void testCloseOnNoAssignment() throws InterruptedException {
+ thread.start();
+ Thread.sleep(10);
+ }
+
+ @Test
+ public void testIdempotentClose() {
+ thread.start();
+ consumerTask.close();
+ consumerTask.close();
+ }
+
+ @Test
+ public void testUserTopicIdPartitionEquals() {
+ final TopicIdPartition tpId = new TopicIdPartition(topicId, new TopicPartition("sample", 0));
+ final UserTopicIdPartition utp1 = new UserTopicIdPartition(tpId, partitioner.metadataPartition(tpId));
+ final UserTopicIdPartition utp2 = new UserTopicIdPartition(tpId, partitioner.metadataPartition(tpId));
+ utp1.isInitialized = true;
+ utp1.isAssigned = true;
+
+ assertFalse(utp2.isInitialized);
+ assertFalse(utp2.isAssigned);
+ assertEquals(utp1, utp2);
+ }
+
+ @Test
+ public void testAddAssignmentsForPartitions() throws InterruptedException {
+ final List idPartitions = getIdPartitions("sample", 3);
+ final Map endOffsets = idPartitions.stream()
+ .map(idp -> toRemoteLogPartition(partitioner.metadataPartition(idp)))
+ .collect(Collectors.toMap(Function.identity(), e -> 0L, (a, b) -> b));
+ consumer.updateEndOffsets(endOffsets);
+ consumerTask.addAssignmentsForPartitions(new HashSet<>(idPartitions));
+ thread.start();
+ for (final TopicIdPartition idPartition : idPartitions) {
+ TestUtils.waitForCondition(() -> consumerTask.isUserPartitionAssigned(idPartition), "Timed out waiting for " + idPartition + " to be assigned");
+ assertTrue(consumerTask.isMetadataPartitionAssigned(partitioner.metadataPartition(idPartition)));
+ assertTrue(handler.isPartitionLoaded.get(idPartition));
+ }
+ }
+
+ @Test
+ public void testRemoveAssignmentsForPartitions() throws InterruptedException {
+ final List allPartitions = getIdPartitions("sample", 3);
+ final Map endOffsets = allPartitions.stream()
+ .map(idp -> toRemoteLogPartition(partitioner.metadataPartition(idp)))
+ .collect(Collectors.toMap(Function.identity(), e -> 0L, (a, b) -> b));
+ consumer.updateEndOffsets(endOffsets);
+ consumerTask.addAssignmentsForPartitions(new HashSet<>(allPartitions));
+ thread.start();
+
+ final TopicIdPartition tpId = allPartitions.get(0);
+ TestUtils.waitForCondition(() -> consumerTask.isUserPartitionAssigned(tpId), "Timed out waiting for " + tpId + " to be assigned");
+ addRecord(consumer, partitioner.metadataPartition(tpId), tpId, 0);
+ TestUtils.waitForCondition(() -> consumerTask.readOffsetForMetadataPartition(partitioner.metadataPartition(tpId)).isPresent(),
+ "Couldn't read record");
+
+ final Set removePartitions = Collections.singleton(tpId);
+ consumerTask.removeAssignmentsForPartitions(removePartitions);
+ for (final TopicIdPartition idPartition : allPartitions) {
+ final TestCondition condition = () -> removePartitions.contains(idPartition) == !consumerTask.isUserPartitionAssigned(idPartition);
+ TestUtils.waitForCondition(condition, "Timed out waiting for " + idPartition + " to be removed");
+ }
+ for (TopicIdPartition removePartition : removePartitions) {
+ TestUtils.waitForCondition(() -> handler.isPartitionCleared.containsKey(removePartition),
+ "Timed out waiting for " + removePartition + " to be cleared");
+ }
+ }
+
+ @Test
+ public void testConcurrentPartitionAssignments() throws InterruptedException, ExecutionException {
+ final List allPartitions = getIdPartitions("sample", 100);
+ final Map endOffsets = allPartitions.stream()
+ .map(idp -> toRemoteLogPartition(partitioner.metadataPartition(idp)))
+ .collect(Collectors.toMap(Function.identity(), e -> 0L, (a, b) -> b));
+ consumer.updateEndOffsets(endOffsets);
+
+ final AtomicBoolean isAllPartitionsAssigned = new AtomicBoolean(false);
+ CountDownLatch latch = new CountDownLatch(1);
+ Thread assignor = new Thread(() -> {
+ int partitionsAssigned = 0;
+ for (TopicIdPartition partition : allPartitions) {
+ if (partitionsAssigned == 50) {
+ // Once half the topic partitions are assigned, wait for the consumer to catch up. This ensures
+ // that the consumer is already running when the rest of the partitions are assigned.
+ try {
+ latch.await(1, TimeUnit.MINUTES);
+ } catch (InterruptedException e) {
+ fail(e.getMessage());
+ }
+ }
+ consumerTask.addAssignmentsForPartitions(Collections.singleton(partition));
+ partitionsAssigned++;
+ }
+ isAllPartitionsAssigned.set(true);
+ });
+ Runnable consumerRunnable = () -> {
+ try {
+ while (!isAllPartitionsAssigned.get()) {
+ consumerTask.maybeWaitForPartitionAssignments();
+ latch.countDown();
+ }
+ } catch (Exception e) {
+ fail(e.getMessage());
+ }
+ };
+
+ ExecutorService consumerExecutor = Executors.newSingleThreadExecutor();
+ Future future = consumerExecutor.submit(consumerRunnable);
+ assignor.start();
+
+ assignor.join();
+ future.get();
+ }
+
+ @Test
+ public void testCanProcessRecord() throws InterruptedException {
+ final Uuid topicId = Uuid.fromString("Bp9TDduJRGa9Q5rlvCJOxg");
+ final TopicIdPartition tpId0 = new TopicIdPartition(topicId, new TopicPartition("sample", 0));
+ final TopicIdPartition tpId1 = new TopicIdPartition(topicId, new TopicPartition("sample", 1));
+ final TopicIdPartition tpId2 = new TopicIdPartition(topicId, new TopicPartition("sample", 2));
+ assertEquals(partitioner.metadataPartition(tpId0), partitioner.metadataPartition(tpId1));
+ assertEquals(partitioner.metadataPartition(tpId0), partitioner.metadataPartition(tpId2));
+
+ final int metadataPartition = partitioner.metadataPartition(tpId0);
+ consumer.updateEndOffsets(Collections.singletonMap(toRemoteLogPartition(metadataPartition), 0L));
+ final Set assignments = Collections.singleton(tpId0);
+ consumerTask.addAssignmentsForPartitions(assignments);
+ thread.start();
+ TestUtils.waitForCondition(() -> consumerTask.isUserPartitionAssigned(tpId0), "Timed out waiting for " + tpId0 + " to be assigned");
+
+ addRecord(consumer, metadataPartition, tpId0, 0);
+ addRecord(consumer, metadataPartition, tpId0, 1);
+ TestUtils.waitForCondition(() -> consumerTask.readOffsetForMetadataPartition(metadataPartition).equals(Optional.of(1L)), "Couldn't read record");
+ assertEquals(2, handler.metadataCounter);
+
+ // should only read the tpId1 records
+ consumerTask.addAssignmentsForPartitions(Collections.singleton(tpId1));
+ TestUtils.waitForCondition(() -> consumerTask.isUserPartitionAssigned(tpId1), "Timed out waiting for " + tpId1 + " to be assigned");
+ addRecord(consumer, metadataPartition, tpId1, 2);
+ TestUtils.waitForCondition(() -> consumerTask.readOffsetForMetadataPartition(metadataPartition).equals(Optional.of(2L)), "Couldn't read record");
+ assertEquals(3, handler.metadataCounter);
+
+ // shouldn't read tpId2 records because it's not assigned
+ addRecord(consumer, metadataPartition, tpId2, 3);
+ TestUtils.waitForCondition(() -> consumerTask.readOffsetForMetadataPartition(metadataPartition).equals(Optional.of(3L)), "Couldn't read record");
+ assertEquals(3, handler.metadataCounter);
+ }
+
+ @Test
+ public void testMaybeMarkUserPartitionsAsReady() throws InterruptedException {
+ final TopicIdPartition tpId = getIdPartitions("hello", 1).get(0);
+ final int metadataPartition = partitioner.metadataPartition(tpId);
+ consumer.updateEndOffsets(Collections.singletonMap(toRemoteLogPartition(metadataPartition), 2L));
+ consumerTask.addAssignmentsForPartitions(Collections.singleton(tpId));
+ thread.start();
+
+ TestUtils.waitForCondition(() -> consumerTask.isUserPartitionAssigned(tpId), "Waiting for " + tpId + " to be assigned");
+ assertTrue(consumerTask.isMetadataPartitionAssigned(metadataPartition));
+ assertFalse(handler.isPartitionInitialized.containsKey(tpId));
+ IntStream.range(0, 5).forEach(offset -> addRecord(consumer, metadataPartition, tpId, offset));
+ TestUtils.waitForCondition(() -> consumerTask.readOffsetForMetadataPartition(metadataPartition).equals(Optional.of(4L)), "Couldn't read record");
+ assertTrue(handler.isPartitionInitialized.get(tpId));
+ }
+
+ @ParameterizedTest
+ @CsvSource(value = {"0, 0", "500, 500"})
+ public void testMaybeMarkUserPartitionAsReadyWhenTopicIsEmpty(long beginOffset,
+ long endOffset) throws InterruptedException {
+ final TopicIdPartition tpId = getIdPartitions("world", 1).get(0);
+ final int metadataPartition = partitioner.metadataPartition(tpId);
+ consumer.updateBeginningOffsets(Collections.singletonMap(toRemoteLogPartition(metadataPartition), beginOffset));
+ consumer.updateEndOffsets(Collections.singletonMap(toRemoteLogPartition(metadataPartition), endOffset));
+ consumerTask.addAssignmentsForPartitions(Collections.singleton(tpId));
+ thread.start();
+
+ TestUtils.waitForCondition(() -> consumerTask.isUserPartitionAssigned(tpId), "Waiting for " + tpId + " to be assigned");
+ assertTrue(consumerTask.isMetadataPartitionAssigned(metadataPartition));
+ TestUtils.waitForCondition(() -> handler.isPartitionInitialized.containsKey(tpId),
+ "should have initialized the partition");
+ assertFalse(consumerTask.readOffsetForMetadataPartition(metadataPartition).isPresent());
+ }
+
+ @Test
+ public void testConcurrentAccess() throws InterruptedException {
+ thread.start();
+ final CountDownLatch latch = new CountDownLatch(1);
+ final TopicIdPartition tpId = getIdPartitions("concurrent", 1).get(0);
+ consumer.updateEndOffsets(Collections.singletonMap(toRemoteLogPartition(partitioner.metadataPartition(tpId)), 0L));
+ final Thread assignmentThread = new Thread(() -> {
+ try {
+ latch.await();
+ consumerTask.addAssignmentsForPartitions(Collections.singleton(tpId));
+ } catch (final InterruptedException e) {
+ fail("Shouldn't have thrown an exception");
+ }
+ });
+ final Thread closeThread = new Thread(() -> {
+ try {
+ latch.await();
+ consumerTask.close();
+ } catch (final InterruptedException e) {
+ fail("Shouldn't have thrown an exception");
+ }
+ });
+ assignmentThread.start();
+ closeThread.start();
+
+ latch.countDown();
+ assignmentThread.join();
+ closeThread.join();
+ }
+
+ @Test
+ public void testConsumerShouldNotCloseOnRetriableError() throws InterruptedException {
+ final TopicIdPartition tpId = getIdPartitions("world", 1).get(0);
+ final int metadataPartition = partitioner.metadataPartition(tpId);
+ consumer.updateEndOffsets(Collections.singletonMap(toRemoteLogPartition(metadataPartition), 1L));
+ consumerTask.addAssignmentsForPartitions(Collections.singleton(tpId));
+ thread.start();
+
+ TestUtils.waitForCondition(() -> consumerTask.isUserPartitionAssigned(tpId), "Waiting for " + tpId + " to be assigned");
+ assertTrue(consumerTask.isMetadataPartitionAssigned(metadataPartition));
+
+ consumer.setPollException(new LeaderNotAvailableException("leader not available!"));
+ addRecord(consumer, metadataPartition, tpId, 0);
+ consumer.setPollException(new TimeoutException("Not able to complete the operation within the timeout"));
+ addRecord(consumer, metadataPartition, tpId, 1);
+
+ TestUtils.waitForCondition(() -> consumerTask.readOffsetForMetadataPartition(metadataPartition).equals(Optional.of(1L)), "Couldn't read record");
+ assertEquals(2, handler.metadataCounter);
+ }
+
+ @Test
+ public void testConsumerShouldCloseOnNonRetriableError() throws InterruptedException {
+ final TopicIdPartition tpId = getIdPartitions("world", 1).get(0);
+ final int metadataPartition = partitioner.metadataPartition(tpId);
+ consumer.updateEndOffsets(Collections.singletonMap(toRemoteLogPartition(metadataPartition), 1L));
+ consumerTask.addAssignmentsForPartitions(Collections.singleton(tpId));
+ thread.start();
+
+ TestUtils.waitForCondition(() -> consumerTask.isUserPartitionAssigned(tpId), "Waiting for " + tpId + " to be assigned");
+ assertTrue(consumerTask.isMetadataPartitionAssigned(metadataPartition));
+
+ consumer.setPollException(new AuthorizationException("Unauthorized to read the topic!"));
+ TestUtils.waitForCondition(() -> consumer.closed(), "Should close the consume on non-retriable error");
+ }
+
+ private void addRecord(final MockConsumer consumer,
+ final int metadataPartition,
+ final TopicIdPartition idPartition,
+ final long recordOffset) {
+ final RemoteLogSegmentId segmentId = new RemoteLogSegmentId(idPartition, Uuid.randomUuid());
+ final RemoteLogMetadata metadata = new RemoteLogSegmentMetadata(segmentId, 0L, 1L, 0L, 0, 0L, 1, Collections.singletonMap(0, 0L));
+ final ConsumerRecord record = new ConsumerRecord<>(TopicBasedRemoteLogMetadataManagerConfig.REMOTE_LOG_METADATA_TOPIC_NAME, metadataPartition, recordOffset, null, serde.serialize(metadata));
+ consumer.addRecord(record);
+ }
+
+ private List getIdPartitions(final String topic, final int partitionCount) {
+ final List idPartitions = new ArrayList<>();
+ for (int partition = 0; partition < partitionCount; partition++) {
+ idPartitions.add(new TopicIdPartition(topicId, new TopicPartition(topic, partition)));
+ }
+ return idPartitions;
+ }
+
+ private static class DummyEventHandler extends RemotePartitionMetadataEventHandler {
+ private int metadataCounter = 0;
+ private final Map isPartitionInitialized = new HashMap<>();
+ private final Map isPartitionLoaded = new HashMap<>();
+ private final Map isPartitionCleared = new HashMap<>();
+
+ @Override
+ protected void handleRemoteLogSegmentMetadata(RemoteLogSegmentMetadata remoteLogSegmentMetadata) {
+ metadataCounter++;
+ }
+
+ @Override
+ protected void handleRemoteLogSegmentMetadataUpdate(RemoteLogSegmentMetadataUpdate remoteLogSegmentMetadataUpdate) {
+ }
+
+ @Override
+ protected void handleRemotePartitionDeleteMetadata(RemotePartitionDeleteMetadata remotePartitionDeleteMetadata) {
+ }
+
+ @Override
+ public void syncLogMetadataSnapshot(TopicIdPartition topicIdPartition, int metadataPartition, Long metadataPartitionOffset) {
+ }
+
+ @Override
+ public void clearTopicPartition(TopicIdPartition topicIdPartition) {
+ isPartitionCleared.put(topicIdPartition, true);
+ }
+
+ @Override
+ public void markInitialized(TopicIdPartition partition) {
+ isPartitionInitialized.put(partition, true);
+ }
+
+ @Override
+ public boolean isInitialized(TopicIdPartition partition) {
+ return true;
+ }
+
+ @Override
+ public void maybeLoadPartition(TopicIdPartition partition) {
+ isPartitionLoaded.put(partition, true);
+ }
+ }
+}
diff --git a/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerHarness.java b/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerHarness.java
index e39d872744a42..abad6ea767605 100644
--- a/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerHarness.java
+++ b/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerHarness.java
@@ -63,11 +63,12 @@ public void initialize(Set topicIdPartitions,
// Call setup to start the cluster.
super.setUp(new EmptyTestInfo());
- initializeRemoteLogMetadataManager(topicIdPartitions, startConsumerThread);
+ initializeRemoteLogMetadataManager(topicIdPartitions, startConsumerThread, null);
}
public void initializeRemoteLogMetadataManager(Set topicIdPartitions,
- boolean startConsumerThread) {
+ boolean startConsumerThread,
+ RemoteLogMetadataTopicPartitioner remoteLogMetadataTopicPartitioner) {
String logDir = TestUtils.tempDirectory("rlmm_segs_").getAbsolutePath();
topicBasedRemoteLogMetadataManager = new TopicBasedRemoteLogMetadataManager(startConsumerThread) {
@Override
@@ -104,6 +105,9 @@ public void onPartitionLeadershipChanges(Set leaderPartitions,
log.debug("TopicBasedRemoteLogMetadataManager configs after adding overridden properties: {}", configs);
topicBasedRemoteLogMetadataManager.configure(configs);
+ if (remoteLogMetadataTopicPartitioner != null) {
+ topicBasedRemoteLogMetadataManager.setRlmTopicPartitioner(remoteLogMetadataTopicPartitioner);
+ }
try {
waitUntilInitialized(60_000);
} catch (TimeoutException e) {
@@ -145,4 +149,4 @@ public void close() throws IOException {
public void closeRemoteLogMetadataManager() {
Utils.closeQuietly(topicBasedRemoteLogMetadataManager, "TopicBasedRemoteLogMetadataManager");
}
-}
\ No newline at end of file
+}
diff --git a/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerMultipleSubscriptionsTest.java b/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerMultipleSubscriptionsTest.java
new file mode 100644
index 0000000000000..3386b94f8957f
--- /dev/null
+++ b/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerMultipleSubscriptionsTest.java
@@ -0,0 +1,178 @@
+/*
+ * 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.server.log.remote.metadata.storage;
+
+
+import kafka.utils.EmptyTestInfo;
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.errors.TimeoutException;
+import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentId;
+import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentMetadata;
+import org.apache.kafka.server.log.remote.storage.RemoteStorageException;
+import org.apache.kafka.test.TestUtils;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import scala.collection.JavaConverters;
+import scala.collection.Seq;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+
+@SuppressWarnings("deprecation") // Added for Scala 2.12 compatibility for usages of JavaConverters
+public class TopicBasedRemoteLogMetadataManagerMultipleSubscriptionsTest {
+ private static final Logger log = LoggerFactory.getLogger(TopicBasedRemoteLogMetadataManagerMultipleSubscriptionsTest.class);
+
+ private static final int SEG_SIZE = 1024 * 1024;
+
+ private final Time time = new MockTime(1);
+ private final TopicBasedRemoteLogMetadataManagerHarness remoteLogMetadataManagerHarness = new TopicBasedRemoteLogMetadataManagerHarness();
+
+ private TopicBasedRemoteLogMetadataManager rlmm() {
+ return remoteLogMetadataManagerHarness.remoteLogMetadataManager();
+ }
+
+ @BeforeEach
+ public void setup() {
+ // Start the cluster only.
+ remoteLogMetadataManagerHarness.setUp(new EmptyTestInfo());
+ }
+
+ @AfterEach
+ public void teardown() throws IOException {
+ remoteLogMetadataManagerHarness.close();
+ }
+
+ @Test
+ public void testMultiplePartitionSubscriptions() throws Exception {
+ // Create topics.
+ String leaderTopic = "leader";
+ HashMap> assignedLeaderTopicReplicas = new HashMap<>();
+ List