+ * This class can be plugged in to Kafka cluster by adding the implementation class as
+ * remote.log.metadata.manager.class.name property value. There is an inbuilt implementation backed by
+ * topic storage in the local cluster. This is used as the default implementation if
+ * remote.log.metadata.manager.class.name is not configured.
+ *
+ * remote.log.metadata.manager.class.path property is about the class path of the RemoteLogStorageManager
+ * implementation. If specified, the RemoteLogStorageManager implementation and its dependent libraries will be loaded
+ * by a dedicated classloader which searches this class path before the Kafka broker class path. The syntax of this
+ * parameter is same with the standard Java class path string.
+ *
+ * remote.log.metadata.manager.listener.name property is about listener name of the local broker to which
+ * it should get connected if needed by RemoteLogMetadataManager implementation. When this is configured all other
+ * required properties can be passed as properties with prefix of 'remote.log.metadata.manager.listener.
+ *
+ */ +@InterfaceStability.Evolving +public interface RemoteLogMetadataManager extends Configurable, Closeable { + + /** + * Adds {@link RemoteLogSegmentMetadata} with the containing {@link RemoteLogSegmentId} into {@link RemoteLogMetadataManager}. + *
+ * RemoteLogSegmentMetadata is identified by RemoteLogSegmentId and it should have the initial state which is {@link RemoteLogSegmentState#COPY_SEGMENT_STARTED}. + *
+ * {@link #updateRemoteLogSegmentMetadata(RemoteLogSegmentMetadataUpdate)} should be used to update an existing RemoteLogSegmentMetadata. + * + * @param remoteLogSegmentMetadata metadata about the remote log segment. + * @throws RemoteStorageException if there are any storage related errors occurred. + * @throws IllegalArgumentException if the given metadata instance does not have the state as {@link RemoteLogSegmentState#COPY_SEGMENT_STARTED} + */ + void addRemoteLogSegmentMetadata(RemoteLogSegmentMetadata remoteLogSegmentMetadata) throws RemoteStorageException; + + /** + * This method is used to update the {@link RemoteLogSegmentMetadata}. Currently, it allows to update with the new + * state based on the life cycle of the segment. It can go through the below state transitions. + *
+ *
+ * +---------------------+ +----------------------+ + * |COPY_SEGMENT_STARTED |----------->|COPY_SEGMENT_FINISHED | + * +-------------------+-+ +--+-------------------+ + * | | + * | | + * v v + * +--+-----------------+-+ + * |DELETE_SEGMENT_STARTED| + * +-----------+----------+ + * | + * | + * v + * +-----------+-----------+ + * |DELETE_SEGMENT_FINISHED| + * +-----------------------+ + *+ *
+ * {@link RemoteLogSegmentState#COPY_SEGMENT_STARTED} - This state indicates that the segment copying to remote storage is started but not yet finished.
+ * {@link RemoteLogSegmentState#COPY_SEGMENT_FINISHED} - This state indicates that the segment copying to remote storage is finished.
+ *
+ * The leader broker copies the log segments to the remote storage and puts the remote log segment metadata with the
+ * state as “COPY_SEGMENT_STARTED” and updates the state as “COPY_SEGMENT_FINISHED” once the copy is successful.
+ *
+ * In the case of default RLMM implementation, remote partition remover processes {@link RemotePartitionDeleteState#DELETE_PARTITION_MARKED}
+ *
+ * Remote Partition Removers uses this method to fetch all the segments for a given topic partition, so that they
+ * can delete them.
+ *
+ * @return Iterator of remote log segment metadata for the given topic partition.
+ */
+ Iterator
+ * New instance is always created with the state as {@link RemoteLogSegmentState#COPY_SEGMENT_STARTED}. This can be
+ * updated by applying {@link RemoteLogSegmentMetadataUpdate} for the respective {@link RemoteLogSegmentId} of the
+ * {@code RemoteLogSegmentMetadata}.
+ */
+@InterfaceStability.Evolving
+public class RemoteLogSegmentMetadata {
+
+ /**
+ * Universally unique remote log segment id.
+ */
+ private final RemoteLogSegmentId remoteLogSegmentId;
+
+ /**
+ * Start offset of this segment.
+ */
+ private final long startOffset;
+
+ /**
+ * End offset of this segment.
+ */
+ private final long endOffset;
+
+ /**
+ * Broker id from which this event is generated.
+ */
+ private final int brokerId;
+
+ /**
+ * Maximum timestamp in the segment
+ */
+ private final long maxTimestamp;
+
+ /**
+ * Epoch time at which the respective {@link #state} is set.
+ */
+ private final long eventTimestamp;
+
+ /**
+ * LeaderEpoch vs offset for messages within this segment.
+ */
+ private final NavigableMap
+ * It goes through the below state transitions.
+ *
+ *
+ *
+ * A resource can be a log segment, any of the indexes or any which was stored in remote storage for a particular log
+ * segment.
+ */
+public class RemoteResourceNotFoundException extends RemoteStorageException {
+ private static final long serialVersionUID = 1L;
+
+ public RemoteResourceNotFoundException(final String message) {
+ super(message);
+ }
+
+ public RemoteResourceNotFoundException(final Throwable cause) {
+ super("Requested remote resource was not found", cause);
+ }
+
+ public RemoteResourceNotFoundException(String message, Throwable cause) {
+ super(message, cause);
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteStorageException.java b/clients/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteStorageException.java
new file mode 100644
index 0000000000000..f2488cb2957fb
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteStorageException.java
@@ -0,0 +1,36 @@
+/*
+ * 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.storage;
+
+/**
+ * Exception thrown when there is a remote storage error.
+ * This can be used as the base exception by implementors of
+ * {@link RemoteStorageManager} or {@link RemoteLogMetadataManager} to create extended exceptions.
+ */
+public class RemoteStorageException extends Exception {
+ private static final long serialVersionUID = 1L;
+
+ public RemoteStorageException(final String message) {
+ super(message);
+ }
+
+ public RemoteStorageException(final String message,
+ final Throwable cause) {
+ super(message, cause);
+ }
+
+}
diff --git a/clients/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteStorageManager.java b/clients/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteStorageManager.java
new file mode 100644
index 0000000000000..64e3fdd026616
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteStorageManager.java
@@ -0,0 +1,141 @@
+/*
+ * 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.storage;
+
+import org.apache.kafka.common.Configurable;
+import org.apache.kafka.common.annotation.InterfaceStability;
+
+import java.io.Closeable;
+import java.io.InputStream;
+
+/**
+ * This interface provides the lifecycle of remote log segments that includes copy, fetch, and delete from remote
+ * storage.
+ *
+ * Each upload or copy of a segment is initiated with {@link RemoteLogSegmentMetadata} containing {@link RemoteLogSegmentId}
+ * which is universally unique even for the same topic partition and offsets.
+ *
+ * {@link RemoteLogSegmentMetadata} is stored in {@link RemoteLogMetadataManager} before and after copy/delete operations on
+ * {@link RemoteStorageManager} with the respective {@link RemoteLogSegmentState}. {@link RemoteLogMetadataManager} is
+ * responsible for storing and fetching metadata about the remote log segments in a strongly consistent manner.
+ * This allows {@link RemoteStorageManager} to have eventual consistency on metadata (although the data is stored
+ * in strongly consistent semantics).
+ */
+@InterfaceStability.Evolving
+public interface RemoteStorageManager extends Configurable, Closeable {
+
+ /**
+ * Type of the index file.
+ */
+ enum IndexType {
+ /**
+ * Represents offset index.
+ */
+ Offset,
+
+ /**
+ * Represents timestamp index.
+ */
+ Timestamp,
+
+ /**
+ * Represents producer snapshot index.
+ */
+ ProducerSnapshot,
+
+ /**
+ * Represents transaction index.
+ */
+ Transaction,
+
+ /**
+ * Represents leader epoch index.
+ */
+ LeaderEpoch,
+ }
+
+ /**
+ * Copies the given {@link LogSegmentData} provided for the given {@code remoteLogSegmentMetadata}. This includes
+ * log segment and its auxiliary indexes like offset index, time index, transaction index, leader epoch index, and
+ * producer snapshot index.
+ *
+ * Invoker of this API should always send a unique id as part of {@link RemoteLogSegmentMetadata#remoteLogSegmentId()}
+ * even when it retries to invoke this method for the same log segment data.
+ *
+ * @param remoteLogSegmentMetadata metadata about the remote log segment.
+ * @param logSegmentData data to be copied to tiered storage.
+ * @throws RemoteStorageException if there are any errors in storing the data of the segment.
+ */
+ void copyLogSegmentData(RemoteLogSegmentMetadata remoteLogSegmentMetadata,
+ LogSegmentData logSegmentData)
+ throws RemoteStorageException;
+
+ /**
+ * Returns the remote log segment data file/object as InputStream for the given {@link RemoteLogSegmentMetadata}
+ * starting from the given startPosition. The stream will end at the end of the remote log segment data file/object.
+ *
+ * @param remoteLogSegmentMetadata metadata about the remote log segment.
+ * @param startPosition start position of log segment to be read, inclusive.
+ * @return input stream of the requested log segment data.
+ * @throws RemoteStorageException if there are any errors while fetching the desired segment.
+ * @throws RemoteResourceNotFoundException when there are no resources associated with the given remoteLogSegmentMetadata.
+ */
+ InputStream fetchLogSegment(RemoteLogSegmentMetadata remoteLogSegmentMetadata,
+ int startPosition) throws RemoteStorageException;
+
+ /**
+ * Returns the remote log segment data file/object as InputStream for the given {@link RemoteLogSegmentMetadata}
+ * starting from the given startPosition. The stream will end at the smaller of endPosition and the end of the
+ * remote log segment data file/object.
+ *
+ * @param remoteLogSegmentMetadata metadata about the remote log segment.
+ * @param startPosition start position of log segment to be read, inclusive.
+ * @param endPosition end position of log segment to be read, inclusive.
+ * @return input stream of the requested log segment data.
+ * @throws RemoteStorageException if there are any errors while fetching the desired segment.
+ * @throws RemoteResourceNotFoundException when there are no resources associated with the given remoteLogSegmentMetadata.
+ */
+ InputStream fetchLogSegment(RemoteLogSegmentMetadata remoteLogSegmentMetadata,
+ int startPosition,
+ int endPosition) throws RemoteStorageException;
+
+ /**
+ * Returns the index for the respective log segment of {@link RemoteLogSegmentMetadata}.
+ *
+ * @param remoteLogSegmentMetadata metadata about the remote log segment.
+ * @param indexType type of the index to be fetched for the segment.
+ * @return input stream of the requested index.
+ * @throws RemoteStorageException if there are any errors while fetching the index.
+ * @throws RemoteResourceNotFoundException when there are no resources associated with the given remoteLogSegmentMetadata.
+ */
+ InputStream fetchIndex(RemoteLogSegmentMetadata remoteLogSegmentMetadata,
+ IndexType indexType) throws RemoteStorageException;
+
+ /**
+ * Deletes the resources associated with the given {@code remoteLogSegmentMetadata}. Deletion is considered as
+ * successful if this call returns successfully without any errors. It will throw {@link RemoteStorageException} if
+ * there are any errors in deleting the file.
+ *
+ *
+ * @param remoteLogSegmentMetadata metadata about the remote log segment to be deleted.
+ * @throws RemoteResourceNotFoundException if the requested resource is not found
+ * @throws RemoteStorageException if there are any storage related errors occurred.
+ * @throws RemoteResourceNotFoundException when there are no resources associated with the given remoteLogSegmentMetadata.
+ */
+ void deleteLogSegmentData(RemoteLogSegmentMetadata remoteLogSegmentMetadata) throws RemoteStorageException;
+
+}
\ No newline at end of file
+ * {@link RemoteLogSegmentState#COPY_SEGMENT_STARTED}
+ * {@link RemoteLogSegmentState#COPY_SEGMENT_FINISHED}
+ * {@link RemoteLogSegmentState#DELETE_SEGMENT_STARTED}
+ * {@link RemoteLogSegmentState#DELETE_SEGMENT_FINISHED}
+ *
+ */
+ public RemoteLogSegmentState state() {
+ return state;
+ }
+
+ /**
+ * Creates a new RemoteLogSegmentMetadata applying the given {@code rlsmUpdate} on this instance. This method will
+ * not update this instance.
+ *
+ * @param rlsmUpdate update to be applied.
+ * @return a new instance created by applying the given update on this instance.
+ */
+ public RemoteLogSegmentMetadata createRemoteLogSegmentWithUpdates(RemoteLogSegmentMetadataUpdate rlsmUpdate) {
+ if (!remoteLogSegmentId.equals(rlsmUpdate.remoteLogSegmentId())) {
+ throw new IllegalArgumentException("Given rlsmUpdate does not have this instance's remoteLogSegmentId.");
+ }
+
+ return new RemoteLogSegmentMetadata(remoteLogSegmentId, startOffset,
+ endOffset, maxTimestamp, rlsmUpdate.brokerId(), rlsmUpdate.eventTimestamp(),
+ segmentSizeInBytes, rlsmUpdate.state(), segmentLeaderEpochs);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ RemoteLogSegmentMetadata that = (RemoteLogSegmentMetadata) o;
+ return startOffset == that.startOffset && endOffset == that.endOffset && brokerId == that.brokerId
+ && maxTimestamp == that.maxTimestamp && eventTimestamp == that.eventTimestamp
+ && segmentSizeInBytes == that.segmentSizeInBytes
+ && Objects.equals(remoteLogSegmentId, that.remoteLogSegmentId)
+ && Objects.equals(segmentLeaderEpochs, that.segmentLeaderEpochs) && state == that.state;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(remoteLogSegmentId, startOffset, endOffset, brokerId, maxTimestamp, eventTimestamp,
+ segmentLeaderEpochs, segmentSizeInBytes, state);
+ }
+
+ @Override
+ public String toString() {
+ return "RemoteLogSegmentMetadata{" +
+ "remoteLogSegmentId=" + remoteLogSegmentId +
+ ", startOffset=" + startOffset +
+ ", endOffset=" + endOffset +
+ ", brokerId=" + brokerId +
+ ", maxTimestamp=" + maxTimestamp +
+ ", eventTimestamp=" + eventTimestamp +
+ ", segmentLeaderEpochs=" + segmentLeaderEpochs +
+ ", segmentSizeInBytes=" + segmentSizeInBytes +
+ ", state=" + state +
+ '}';
+ }
+
+}
diff --git a/clients/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogSegmentMetadataUpdate.java b/clients/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogSegmentMetadataUpdate.java
new file mode 100644
index 0000000000000..4fc1ea1b4cc95
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogSegmentMetadataUpdate.java
@@ -0,0 +1,120 @@
+/*
+ * 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.storage;
+
+import org.apache.kafka.common.annotation.InterfaceStability;
+
+import java.util.Objects;
+
+/**
+ * It describes the metadata update about the log segment in the remote storage. This is currently used to update the
+ * state of the remote log segment by using {@link RemoteLogMetadataManager#updateRemoteLogSegmentMetadata(RemoteLogSegmentMetadataUpdate)}.
+ * This also includes the timestamp of this event.
+ */
+@InterfaceStability.Evolving
+public class RemoteLogSegmentMetadataUpdate {
+
+ /**
+ * Universally unique remote log segment id.
+ */
+ private final RemoteLogSegmentId remoteLogSegmentId;
+
+ /**
+ * Epoch time at which this event is generated.
+ */
+ private final long eventTimestamp;
+
+ /**
+ * It indicates the state in which the action is executed on this segment.
+ */
+ private final RemoteLogSegmentState state;
+
+ /**
+ * Broker id from which this event is generated.
+ */
+ private final int brokerId;
+
+ /**
+ * @param remoteLogSegmentId Universally unique remote log segment id.
+ * @param eventTimestamp Epoch time in milli seconds at which the remote log segment is copied to the remote tier storage.
+ * @param state State of the remote log segment.
+ * @param brokerId Broker id from which this event is generated.
+ */
+ public RemoteLogSegmentMetadataUpdate(RemoteLogSegmentId remoteLogSegmentId, long eventTimestamp,
+ RemoteLogSegmentState state, int brokerId) {
+ this.remoteLogSegmentId = Objects.requireNonNull(remoteLogSegmentId, "remoteLogSegmentId can not be null");
+ this.state = Objects.requireNonNull(state, "state can not be null");
+ this.brokerId = brokerId;
+ this.eventTimestamp = eventTimestamp;
+ }
+
+ /**
+ * @return Universally unique id of this remote log segment.
+ */
+ public RemoteLogSegmentId remoteLogSegmentId() {
+ return remoteLogSegmentId;
+ }
+
+ /**
+ * @return Epoch time at which this event is generated.
+ */
+ public long eventTimestamp() {
+ return eventTimestamp;
+ }
+
+ /**
+ * It represents the state of the remote log segment. It can be one of the values of {@link RemoteLogSegmentState}.
+ */
+ public RemoteLogSegmentState state() {
+ return state;
+ }
+
+ /**
+ * @return Broker id from which this event is generated.
+ */
+ public int brokerId() {
+ return brokerId;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ RemoteLogSegmentMetadataUpdate that = (RemoteLogSegmentMetadataUpdate) o;
+ return eventTimestamp == that.eventTimestamp && Objects
+ .equals(remoteLogSegmentId, that.remoteLogSegmentId) && state == that.state && brokerId == that.brokerId;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(remoteLogSegmentId, eventTimestamp, state, brokerId);
+ }
+
+ @Override
+ public String toString() {
+ return "RemoteLogSegmentMetadataUpdate{" +
+ "remoteLogSegmentId=" + remoteLogSegmentId +
+ ", eventTimestamp=" + eventTimestamp +
+ ", state=" + state +
+ ", brokerId=" + brokerId +
+ '}';
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogSegmentState.java b/clients/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogSegmentState.java
new file mode 100644
index 0000000000000..e27fdd19cd43b
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogSegmentState.java
@@ -0,0 +1,90 @@
+/*
+ * 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.storage;
+
+import org.apache.kafka.common.annotation.InterfaceStability;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/**
+ * It indicates the state of the remote log segment. This will be based on the action executed on this
+ * segment by the remote log service implementation.
+ *
+ * +---------------------+ +----------------------+
+ * |COPY_SEGMENT_STARTED |----------->|COPY_SEGMENT_FINISHED |
+ * +-------------------+-+ +--+-------------------+
+ * | |
+ * | |
+ * v v
+ * +--+-----------------+-+
+ * |DELETE_SEGMENT_STARTED|
+ * +-----------+----------+
+ * |
+ * |
+ * v
+ * +-----------+-----------+
+ * |DELETE_SEGMENT_FINISHED|
+ * +-----------------------+
+ *
+ */
+@InterfaceStability.Evolving
+public enum RemoteLogSegmentState {
+
+ /**
+ * This state indicates that the segment copying to remote storage is started but not yet finished.
+ */
+ COPY_SEGMENT_STARTED((byte) 0),
+
+ /**
+ * This state indicates that the segment copying to remote storage is finished.
+ */
+ COPY_SEGMENT_FINISHED((byte) 1),
+
+ /**
+ * This state indicates that the segment deletion is started but not yet finished.
+ */
+ DELETE_SEGMENT_STARTED((byte) 2),
+
+ /**
+ * This state indicates that the segment is deleted successfully.
+ */
+ DELETE_SEGMENT_FINISHED((byte) 3);
+
+ private static final Map
+ * +-------------------------+
+ * |DELETE_PARTITION_MARKED |
+ * +-----------+-------------+
+ * |
+ * |
+ * +-----------v--------------+
+ * |DELETE_PARTITION_STARTED |
+ * +-----------+--------------+
+ * |
+ * |
+ * +-----------v--------------+
+ * |DELETE_PARTITION_FINISHED |
+ * +--------------------------+
+ *
+ *