Skip to content

MINOR Replaced File with Path in LogSegmentData. #10424

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
junrao merged 3 commits into from
Mar 31, 2021
Merged

MINOR Replaced File with Path in LogSegmentData. #10424

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
a5f302c
MINOR Replaced File with Path in LogSegment Data.
satishd Mar 29, 2021
2eef3ca
Renamed variables/methods to have suffix of Ms instead of Millis.
satishd Mar 30, 2021
3c4d6e5
Addressed reveiw comments.
satishd Mar 31, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 23 additions & 24 deletions clients/src/main/java/org/apache/kafka/server/log/remote/storage/LogSegmentData.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

/**
Expand All @@ -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,
Comment thread
satishd marked this conversation as resolved.
Outdated
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");
Expand All @@ -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;
}

Expand All @@ -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
Expand Down
62 changes: 31 additions & 31 deletions ...ts/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogSegmentMetadata.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -82,26 +82,26 @@ public class RemoteLogSegmentMetadata {

/**
* Creates an instance with the given metadata of remote log segment.
*
* <p>
* {@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.
*/
private RemoteLogSegmentMetadata(RemoteLogSegmentId remoteLogSegmentId,
long startOffset,
long endOffset,
long maxTimestamp,
long maxTimestampMs,
int brokerId,
long eventTimestamp,
long eventTimestampMs,
int segmentSizeInBytes,
RemoteLogSegmentState state,
Map<Integer, Long> segmentLeaderEpochs) {
Expand All @@ -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()) {
Expand All @@ -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}.
*
* <p>
* {@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<Integer, Long> segmentLeaderEpochs) {
this(remoteLogSegmentId,
startOffset,
endOffset,
maxTimestamp,
maxTimestampMs,
brokerId,
eventTimestamp, segmentSizeInBytes,
eventTimestampMs, segmentSizeInBytes,
RemoteLogSegmentState.COPY_SEGMENT_STARTED,
segmentLeaderEpochs);
}
Expand Down Expand Up @@ -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() {
Comment thread
satishd marked this conversation as resolved.
return eventTimestampMs;
}

/**
Expand All @@ -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;
}

/**
Expand Down Expand Up @@ -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);
}

Expand All @@ -252,16 +252,16 @@ 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;
}

@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
Expand All @@ -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 +
Expand Down
26 changes: 14 additions & 12 deletions .../main/java/org/apache/kafka/server/log/remote/storage/RemoteLogSegmentMetadataUpdate.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -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;
}

/**
Expand All @@ -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;
}

/**
Expand All @@ -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 +
'}';
Expand Down
Loading