Page History

/**
 * RemoteStorageManager provides the lifecycle of remote log segments whichthat includes copy, fetch, and delete operations.
 *
 * {@link RemoteLogMetadataManager} is responsible for storing and fetching metadata about the remote log segments in afrom remote
 * strongly consistent mannerstorage.
 * <p>
 * Each upload or copy of a segment is giveninitiated with a{@link RemoteLogSegmentMetadata} containing {@link RemoteLogSegmentId}
 * which is universally unique even for the
 * same topic partition and offsets.
 Once* the<p>
 copy* orRemoteLogSegmentMetadata uploadis isstored successful,in {@link RemoteLogSegmentMetadataRemoteLogMetadataManager} is
before *and createdafter withcopy/delete RemoteLogSegmentIdoperations andon
 other* logRemoteStorageManager segmentwith informationthe andrespective it is stored in {@link RemoteLogSegmentMetadata.State}. {@link RemoteLogMetadataManager}. is
 * Thisresponsible allowsfor RemoteStorageManagerstoring toand storefetching segmentsmetadata even about the remote log segments in a strongly consistent manner.
 * This allows RemoteStorageManager to store segments even in eventually consistent manner as the metadata is already
 * stored in a consistent store.
 * <p>
 * All these APIs are still experimentalevolving.
 */
@InterfaceStability.Unstable
public interface RemoteStorageManager extends Configurable, Closeable {
    InputStream EMPTY_INPUT_STREAM = new ByteArrayInputStream(new byte[0]);

    /**
     * Copies LogSegmentData provided for the given RemoteLogSegmentId{@param remoteLogSegmentMetadata}.
     * information about this copy operation. This can include path to the object in the store etc.<p>
     *
     * Invoker of this API should always send a unique id as part of {@link RemoteLogSegmentMetadata#remoteLogSegmentId#idRemoteLogSegmentMetadata#remoteLogSegmentId()#id()}
 even when it
  * even when *it retries to invoke this method for the same log segment data.
     *
     * @param remoteLogSegmentId
remoteLogSegmentMetadata metadata about the remote * @param logSegmentDatalog segment.
     * @return@param logSegmentData           data to be copied to tiered storage.
     * @throws IOException RemoteStorageException if there are any errors in storing the data of the segment.
     */
    void copyLogSegment(RemoteLogSegmentMetadata remoteLogSegmentMetadata, LogSegmentData logSegmentData)
            throws RemoteStorageException;

    /**
     * Returns the remote log segment data file/object as InputStream for the given 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
     * @param endPosition
     *start @return
position of log segment to *be @throwsread, IOExceptioninclusive.
     */
 @param endPosition  InputStream fetchLogSegmentData(RemoteLogSegmentMetadata remoteLogSegmentMetadata,
          end position of log segment to be read, inclusive.
     * @return
     * @throws RemoteStorageException if there are any Longerrors startPosition,while Longfetching endPosition)the throwsdesired RemoteStorageException;segment.

     */
    InputStream fetchLogSegmentData(RemoteLogSegmentMetadata remoteLogSegmentMetadata,
                                    Long startPosition, Long endPosition) throws RemoteStorageException;

    /**
     * Returns the offset index for the respective log segment of {@link RemoteLogSegmentMetadata}.
     *
     * @param remoteLogSegmentMetadata metadata about the remote log segment.
     * @return
     * @throws IOException RemoteStorageException if there are any errors while fetching the index.
     */
    InputStream fetchOffsetIndex(RemoteLogSegmentMetadata remoteLogSegmentMetadata) throws RemoteStorageException;

    /**
     * Returns the timestamp index for the respective log segment of {@link RemoteLogSegmentMetadata}.
     *
     * @param remoteLogSegmentMetadata metadata about the remote log segment.
     * @return
     * @throws IOExceptionRemoteStorageException if there are any errors while fetching the index.
     */
    InputStream fetchTimestampIndex(RemoteLogSegmentMetadata remoteLogSegmentMetadata) throws RemoteStorageException;

    /**
     *
 Returns the transaction index *for @paramthe remoteLogSegmentMetadata
the respective log segment of * @return{@link RemoteLogSegmentMetadata}.
     * @throws RemoteStorageException
     */
    default InputStream fetchTransactionIndex(RemoteLogSegmentMetadata @param remoteLogSegmentMetadata) throwsmetadata RemoteStorageExceptionabout {
the remote       throw new UnsupportedOperationException();
    }

log segment.
      /**
     * @return
     * @param@throws remoteLogSegmentMetadata
RemoteStorageException if there are any *errors @return
while fetching    * @throws RemoteStorageExceptionthe index.
     */
    default InputStream fetchProducerSnapshotIndexfetchTransactionIndex(RemoteLogSegmentMetadata remoteLogSegmentMetadata) throws RemoteStorageException {
        throw new UnsupportedOperationException()return EMPTY_INPUT_STREAM;
    }

    /**
     * Returns the leaderproducer epochsnapshot index tillfor the the respective log segment of {@link RemoteLogSegmentMetadata}.
     *
     * @param remoteLogSegmentMetadata
 metadata about the remote *log @returnsegment.
     * @throws RemoteStorageException@return
     */
 @throws RemoteStorageException if default InputStream fetchLeaderEpochIndexthere are any errors while fetching the index.
     */
    default InputStream fetchProducerSnapshotIndex(RemoteLogSegmentMetadata remoteLogSegmentMetadata) throws RemoteStorageException {
        throw new UnsupportedOperationException()return EMPTY_INPUT_STREAM;
    }

    /**
     * DeletesReturns the remoteleader logepoch segmentindex for the giventhe remoteLogSegmentMetadata.respective Deletionlog issegment consideredof as successful if{@link RemoteLogSegmentMetadata}.
     *
 this call returns successfully without* any@param exceptionsremoteLogSegmentMetadata aftermetadata aabout fewthe retries.remote Itlog willsegment.
 throw {@link RemoteStorageException} if there* are@return
     * @throws RemoteStorageException if there are any errors inwhile deletingfetching the fileindex.
     * /
    default * todo: Add retry configuration.InputStream fetchLeaderEpochIndex(RemoteLogSegmentMetadata remoteLogSegmentMetadata) throws RemoteStorageException {
     *
   return EMPTY_INPUT_STREAM;
  * @param remoteLogSegmentMetadata}

    /**
 * @throws IOException
  * Deletes the */
resources associated with the void deleteLogSegment(RemoteLogSegmentMetadata remoteLogSegmentMetadata) throws RemoteStorageException;

}



/**
 * This represents a universally unique id associated to a topic partition's log segment. This will be regenerated for
 * every attempt of copying a specific log segment in {@link RemoteLogStorageManager#copyLogSegment(RemoteLogSegmentId, LogSegmentData)}.
 */
public class RemoteLogSegmentId {
    private TopicPartition topicPartition;
    private UUID id;

    public RemoteLogSegmentId(TopicPartition topicPartition, UUID id) {
        this.topicPartition = requireNonNull(topicPartition);given {@param 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.
     * <p>
     * {@link RemoteResourceNotFoundException} is thrown when there are no resources associated with the given
     * {@param remoteLogSegmentMetadata}.
     *
     * @param remoteLogSegmentMetadata this.idmetadata = requireNonNull(id);
    }

    public TopicPartition topicPartition() {
        return topicPartition;about the remote log segment to be deleted.
     * @throws RemoteResourceNotFoundException if the requested resource is not found
    }

 * @throws RemoteStorageException public UUID id() {
      if there returnare id;
any storage related errors }
occurred...
}



public class LogSegmentData {

     */
    private final File logSegment;
    private final File offsetIndex;
    private final File timeIndex;
    private final File txnIndex;
    private final File producerIdSnapshotIndex;
    private final File leaderEpochIndex;

    public LogSegmentData(File logSegment, File offsetIndex, File timeIndex, File txnIndex, File producerIdSnapshotIndex,
         void deleteLogSegment(RemoteLogSegmentMetadata remoteLogSegmentMetadata) throws RemoteStorageException;

}



/**
 * This represents a universally unique id associated to a topic partition's log segment. This will be regenerated for
 * every attempt of copying a specific log segment in {@link RemoteLogStorageManager#copyLogSegment(RemoteLogSegmentId, LogSegmentData)}.
 */
public class RemoteLogSegmentId {
    private TopicPartition topicPartition;
    private UUID id;

    public RemoteLogSegmentId(TopicPartition topicPartition, UUID id) {
        this.topicPartition = requireNonNull(topicPartition);
        this.id File= leaderEpochIndexrequireNonNull(id);
 {
   }

    public this.logSegment = logSegment;TopicPartition topicPartition() {
        this.offsetIndex = offsetIndexreturn topicPartition;
    }

    this.timeIndex = timeIndex;
public UUID id() {
         this.txnIndex = txnIndex;
        this.producerIdSnapshotIndex = producerIdSnapshotIndex;
        this.leaderEpochIndex = leaderEpochIndex;
    }
...
}