diff --git a/clients/src/main/java/org/apache/kafka/server/log/remote/storage/LogSegmentData.java b/clients/src/main/java/org/apache/kafka/server/log/remote/storage/LogSegmentData.java index 662e396352142..b1a4d89b65934 100644 --- a/clients/src/main/java/org/apache/kafka/server/log/remote/storage/LogSegmentData.java +++ b/clients/src/main/java/org/apache/kafka/server/log/remote/storage/LogSegmentData.java @@ -18,8 +18,8 @@ import org.apache.kafka.common.annotation.InterfaceStability; -import java.io.File; import java.nio.ByteBuffer; +import java.nio.file.Path; import java.util.Objects; /** @@ -30,28 +30,27 @@ @InterfaceStability.Evolving public class LogSegmentData { - private final File logSegment; - private final File offsetIndex; - private final File timeIndex; - private final File txnIndex; - private final File producerSnapshotIndex; + private final Path logSegment; + private final Path offsetIndex; + private final Path timeIndex; + private final Path txnIndex; + private final Path producerSnapshotIndex; private final ByteBuffer leaderEpochIndex; /** * Creates a LogSegmentData instance with data and indexes. - * - * @param logSegment actual log segment file + * @param logSegment actual log segment file * @param offsetIndex offset index file * @param timeIndex time index file * @param txnIndex transaction index file * @param producerSnapshotIndex producer snapshot until this segment * @param leaderEpochIndex leader-epoch-index until this segment */ - public LogSegmentData(File logSegment, - File offsetIndex, - File timeIndex, - File txnIndex, - File producerSnapshotIndex, + public LogSegmentData(Path logSegment, + Path offsetIndex, + Path timeIndex, + Path txnIndex, + Path producerSnapshotIndex, ByteBuffer leaderEpochIndex) { this.logSegment = Objects.requireNonNull(logSegment, "logSegment can not be null"); this.offsetIndex = Objects.requireNonNull(offsetIndex, "offsetIndex can not be null"); @@ -64,35 +63,35 @@ public LogSegmentData(File logSegment, /** * @return Log segment file of this segment. */ - public File logSegment() { + public Path logSegment() { return logSegment; } /** * @return Offset index file. */ - public File offsetIndex() { + public Path offsetIndex() { return offsetIndex; } /** * @return Time index file of this segment. */ - public File timeIndex() { + public Path timeIndex() { return timeIndex; } /** * @return Transaction index file of this segment. */ - public File txnIndex() { + public Path txnIndex() { return txnIndex; } /** * @return Producer snapshot file until this segment. */ - public File producerSnapshotIndex() { + public Path producerSnapshotIndex() { return producerSnapshotIndex; } @@ -112,12 +111,12 @@ public boolean equals(Object o) { return false; } LogSegmentData that = (LogSegmentData) o; - return Objects.equals(logSegment, that.logSegment) && Objects - .equals(offsetIndex, that.offsetIndex) && Objects - .equals(timeIndex, that.timeIndex) && Objects - .equals(txnIndex, that.txnIndex) && Objects - .equals(producerSnapshotIndex, that.producerSnapshotIndex) && Objects - .equals(leaderEpochIndex, that.leaderEpochIndex); + return Objects.equals(logSegment, that.logSegment) && + Objects.equals(offsetIndex, that.offsetIndex) && + Objects.equals(timeIndex, that.timeIndex) && + Objects.equals(txnIndex, that.txnIndex) && + Objects.equals(producerSnapshotIndex, that.producerSnapshotIndex) && + Objects.equals(leaderEpochIndex, that.leaderEpochIndex); } @Override diff --git a/clients/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogSegmentMetadata.java b/clients/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogSegmentMetadata.java index 9f0dd817be849..4a59e3093ca6b 100644 --- a/clients/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogSegmentMetadata.java +++ b/clients/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogSegmentMetadata.java @@ -56,14 +56,14 @@ public class RemoteLogSegmentMetadata { private final int brokerId; /** - * Maximum timestamp in the segment + * Maximum timestamp in milli seconds in the segment */ - private final long maxTimestamp; + private final long maxTimestampMs; /** - * Epoch time at which the respective {@link #state} is set. + * Epoch time in milli seconds at which the respective {@link #state} is set. */ - private final long eventTimestamp; + private final long eventTimestampMs; /** * LeaderEpoch vs offset for messages within this segment. @@ -82,16 +82,16 @@ public class RemoteLogSegmentMetadata { /** * Creates an instance with the given metadata of remote log segment. - * + *

* {@code segmentLeaderEpochs} can not be empty. If all the records in this segment belong to the same leader epoch * then it should have an entry with epoch mapping to start-offset of this segment. * * @param remoteLogSegmentId Universally unique remote log segment id. * @param startOffset Start offset of this segment (inclusive). * @param endOffset End offset of this segment (inclusive). - * @param maxTimestamp Maximum timestamp in this segment. + * @param maxTimestampMs Maximum timestamp in milli seconds in this segment. * @param brokerId Broker id from which this event is generated. - * @param eventTimestamp Epoch time in milli seconds at which the remote log segment is copied to the remote tier storage. + * @param eventTimestampMs Epoch time in milli seconds at which the remote log segment is copied to the remote tier storage. * @param segmentSizeInBytes Size of this segment in bytes. * @param state State of the respective segment of remoteLogSegmentId. * @param segmentLeaderEpochs leader epochs occurred within this segment. @@ -99,9 +99,9 @@ public class RemoteLogSegmentMetadata { private RemoteLogSegmentMetadata(RemoteLogSegmentId remoteLogSegmentId, long startOffset, long endOffset, - long maxTimestamp, + long maxTimestampMs, int brokerId, - long eventTimestamp, + long eventTimestampMs, int segmentSizeInBytes, RemoteLogSegmentState state, Map segmentLeaderEpochs) { @@ -111,8 +111,8 @@ private RemoteLogSegmentMetadata(RemoteLogSegmentId remoteLogSegmentId, this.startOffset = startOffset; this.endOffset = endOffset; this.brokerId = brokerId; - this.maxTimestamp = maxTimestamp; - this.eventTimestamp = eventTimestamp; + this.maxTimestampMs = maxTimestampMs; + this.eventTimestampMs = eventTimestampMs; this.segmentSizeInBytes = segmentSizeInBytes; if (segmentLeaderEpochs == null || segmentLeaderEpochs.isEmpty()) { @@ -124,33 +124,33 @@ private RemoteLogSegmentMetadata(RemoteLogSegmentId remoteLogSegmentId, /** * Creates an instance with the given metadata of remote log segment and its state as {@link RemoteLogSegmentState#COPY_SEGMENT_STARTED}. - * + *

* {@code segmentLeaderEpochs} can not be empty. If all the records in this segment belong to the same leader epoch * then it should have an entry with epoch mapping to start-offset of this segment. * * @param remoteLogSegmentId Universally unique remote log segment id. * @param startOffset Start offset of this segment (inclusive). * @param endOffset End offset of this segment (inclusive). - * @param maxTimestamp Maximum timestamp in this segment + * @param maxTimestampMs Maximum timestamp in this segment * @param brokerId Broker id from which this event is generated. - * @param eventTimestamp Epoch time in milli seconds at which the remote log segment is copied to the remote tier storage. + * @param eventTimestampMs Epoch time in milli seconds at which the remote log segment is copied to the remote tier storage. * @param segmentSizeInBytes Size of this segment in bytes. * @param segmentLeaderEpochs leader epochs occurred within this segment */ public RemoteLogSegmentMetadata(RemoteLogSegmentId remoteLogSegmentId, long startOffset, long endOffset, - long maxTimestamp, + long maxTimestampMs, int brokerId, - long eventTimestamp, + long eventTimestampMs, int segmentSizeInBytes, Map segmentLeaderEpochs) { this(remoteLogSegmentId, startOffset, endOffset, - maxTimestamp, + maxTimestampMs, brokerId, - eventTimestamp, segmentSizeInBytes, + eventTimestampMs, segmentSizeInBytes, RemoteLogSegmentState.COPY_SEGMENT_STARTED, segmentLeaderEpochs); } @@ -178,10 +178,10 @@ public long endOffset() { } /** - * @return Epoch time at which this event is occurred. + * @return Epoch time in milli seconds at which this event is occurred. */ - public long eventTimestamp() { - return eventTimestamp; + public long eventTimestampMs() { + return eventTimestampMs; } /** @@ -192,10 +192,10 @@ public int segmentSizeInBytes() { } /** - * @return Maximum timestamp of a record within this segment. + * @return Maximum timestamp in milli seconds of a record within this segment. */ - public long maxTimestamp() { - return maxTimestamp; + public long maxTimestampMs() { + return maxTimestampMs; } /** @@ -232,13 +232,13 @@ public RemoteLogSegmentState state() { * @param rlsmUpdate update to be applied. * @return a new instance created by applying the given update on this instance. */ - public RemoteLogSegmentMetadata createRemoteLogSegmentWithUpdates(RemoteLogSegmentMetadataUpdate rlsmUpdate) { + public RemoteLogSegmentMetadata createWithUpdates(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(), + endOffset, maxTimestampMs, rlsmUpdate.brokerId(), rlsmUpdate.eventTimestampMs(), segmentSizeInBytes, rlsmUpdate.state(), segmentLeaderEpochs); } @@ -252,7 +252,7 @@ public boolean equals(Object o) { } RemoteLogSegmentMetadata that = (RemoteLogSegmentMetadata) o; return startOffset == that.startOffset && endOffset == that.endOffset && brokerId == that.brokerId - && maxTimestamp == that.maxTimestamp && eventTimestamp == that.eventTimestamp + && maxTimestampMs == that.maxTimestampMs && eventTimestampMs == that.eventTimestampMs && segmentSizeInBytes == that.segmentSizeInBytes && Objects.equals(remoteLogSegmentId, that.remoteLogSegmentId) && Objects.equals(segmentLeaderEpochs, that.segmentLeaderEpochs) && state == that.state; @@ -260,8 +260,8 @@ public boolean equals(Object o) { @Override public int hashCode() { - return Objects.hash(remoteLogSegmentId, startOffset, endOffset, brokerId, maxTimestamp, eventTimestamp, - segmentLeaderEpochs, segmentSizeInBytes, state); + return Objects.hash(remoteLogSegmentId, startOffset, endOffset, brokerId, maxTimestampMs, + eventTimestampMs, segmentLeaderEpochs, segmentSizeInBytes, state); } @Override @@ -271,8 +271,8 @@ public String toString() { ", startOffset=" + startOffset + ", endOffset=" + endOffset + ", brokerId=" + brokerId + - ", maxTimestamp=" + maxTimestamp + - ", eventTimestamp=" + eventTimestamp + + ", maxTimestampMs=" + maxTimestampMs + + ", eventTimestampMs=" + eventTimestampMs + ", 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 index 4fc1ea1b4cc95..56ed8dbff20cc 100644 --- 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 @@ -34,9 +34,9 @@ public class RemoteLogSegmentMetadataUpdate { private final RemoteLogSegmentId remoteLogSegmentId; /** - * Epoch time at which this event is generated. + * Epoch time in milli seconds at which this event is generated. */ - private final long eventTimestamp; + private final long eventTimestampMs; /** * It indicates the state in which the action is executed on this segment. @@ -50,16 +50,16 @@ public class RemoteLogSegmentMetadataUpdate { /** * @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 eventTimestampMs 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, + public RemoteLogSegmentMetadataUpdate(RemoteLogSegmentId remoteLogSegmentId, long eventTimestampMs, 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; + this.eventTimestampMs = eventTimestampMs; } /** @@ -70,10 +70,10 @@ public RemoteLogSegmentId remoteLogSegmentId() { } /** - * @return Epoch time at which this event is generated. + * @return Epoch time in milli seconds at which this event is generated. */ - public long eventTimestamp() { - return eventTimestamp; + public long eventTimestampMs() { + return eventTimestampMs; } /** @@ -99,20 +99,22 @@ public boolean equals(Object o) { return false; } RemoteLogSegmentMetadataUpdate that = (RemoteLogSegmentMetadataUpdate) o; - return eventTimestamp == that.eventTimestamp && Objects - .equals(remoteLogSegmentId, that.remoteLogSegmentId) && state == that.state && brokerId == that.brokerId; + return eventTimestampMs == that.eventTimestampMs && + Objects.equals(remoteLogSegmentId, that.remoteLogSegmentId) && + state == that.state && + brokerId == that.brokerId; } @Override public int hashCode() { - return Objects.hash(remoteLogSegmentId, eventTimestamp, state, brokerId); + return Objects.hash(remoteLogSegmentId, eventTimestampMs, state, brokerId); } @Override public String toString() { return "RemoteLogSegmentMetadataUpdate{" + "remoteLogSegmentId=" + remoteLogSegmentId + - ", eventTimestamp=" + eventTimestamp + + ", eventTimestampMs=" + eventTimestampMs + ", state=" + state + ", brokerId=" + brokerId + '}'; 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 index 64e3fdd026616..6231d8e25c184 100644 --- 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 @@ -45,27 +45,27 @@ enum IndexType { /** * Represents offset index. */ - Offset, + OFFSET, /** * Represents timestamp index. */ - Timestamp, + TIMESTAMP, /** * Represents producer snapshot index. */ - ProducerSnapshot, + PRODUCER_SNAPSHOT, /** * Represents transaction index. */ - Transaction, + TRANSACTION, /** * Represents leader epoch index. */ - LeaderEpoch, + LEADER_EPOCH, } /** @@ -137,5 +137,4 @@ InputStream fetchIndex(RemoteLogSegmentMetadata remoteLogSegmentMetadata, * @throws RemoteResourceNotFoundException when there are no resources associated with the given remoteLogSegmentMetadata. */ void deleteLogSegmentData(RemoteLogSegmentMetadata remoteLogSegmentMetadata) throws RemoteStorageException; - } \ No newline at end of file