MINOR; Add accessor methods to OffsetAndEpoch

metadata/src/main/java/org/apache/kafka/image/MetadataDelta.java

@@ -71,8 +71,8 @@ public final class MetadataDelta {
 
     public MetadataDelta(MetadataImage image) {
         this.image = image;
-        this.highestOffset = image.highestOffsetAndEpoch().offset;
-        this.highestEpoch = image.highestOffsetAndEpoch().epoch;
+        this.highestOffset = image.highestOffsetAndEpoch().offset();
+        this.highestEpoch = image.highestOffsetAndEpoch().epoch();
     }
 
     public MetadataImage image() {

metadata/src/test/java/org/apache/kafka/image/MetadataImageTest.java

@@ -91,8 +91,8 @@ private void testToImageAndBack(MetadataImage image) throws Throwable {
         image.write(writer, new ImageWriterOptions.Builder().build());
         MetadataDelta delta = new MetadataDelta(MetadataImage.EMPTY);
         RecordTestUtils.replayAll(delta,
-                image.highestOffsetAndEpoch().offset,
-                image.highestOffsetAndEpoch().epoch,
+                image.highestOffsetAndEpoch().offset(),
+                image.highestOffsetAndEpoch().epoch(),
                 writer.records());
         MetadataImage nextImage = delta.apply();
         assertEquals(image, nextImage);

metadata/src/test/java/org/apache/kafka/image/writer/RaftSnapshotWriterTest.java

@@ -48,12 +48,12 @@ public OffsetAndEpoch snapshotId() {
 
         @Override
         public long lastContainedLogOffset() {
-            return snapshotId().offset;
+            return snapshotId().offset();
         }
 
         @Override
         public int lastContainedLogEpoch() {
-            return snapshotId().epoch;
+            return snapshotId().epoch();
         }
 
         @Override

metadata/src/test/java/org/apache/kafka/metadata/RecordTestUtils.java

@@ -60,8 +60,8 @@ public static void replayAll(Object target,
         if (target instanceof MetadataDelta) {
             MetadataDelta delta = (MetadataDelta) target;
             replayAll(delta,
-                    delta.image().highestOffsetAndEpoch().offset,
-                    delta.image().highestOffsetAndEpoch().epoch,
+                    delta.image().highestOffsetAndEpoch().offset(),
+                    delta.image().highestOffsetAndEpoch().epoch(),
                     recordsAndVersions);
             return;
         }

metadata/src/test/java/org/apache/kafka/metalog/LocalLogManager.java

@@ -198,7 +198,7 @@ public static class SharedLogData {
         public SharedLogData(Optional<RawSnapshotReader> snapshot) {
             if (snapshot.isPresent()) {
                 RawSnapshotReader initialSnapshot = snapshot.get();
-                prevOffset = initialSnapshot.snapshotId().offset - 1;
+                prevOffset = initialSnapshot.snapshotId().offset() - 1;
                 snapshots.put(prevOffset, initialSnapshot);
             } else {
                 prevOffset = -1;
@@ -305,14 +305,14 @@ synchronized Optional<RawSnapshotReader> nextSnapshot(long offset) {
          * Stores a new snapshot and notifies all threads waiting for a snapshot.
          */
         synchronized void addSnapshot(RawSnapshotReader newSnapshot) {
-            if (newSnapshot.snapshotId().offset - 1 > prevOffset) {
+            if (newSnapshot.snapshotId().offset() - 1 > prevOffset) {
                 log.error(
                     "Ignored attempt to add a snapshot {} that is greater than the latest offset {}",
                     newSnapshot,
                     prevOffset
                 );
             } else {
-                snapshots.put(newSnapshot.snapshotId().offset - 1, newSnapshot);
+                snapshots.put(newSnapshot.snapshotId().offset() - 1, newSnapshot);
                 this.notifyAll();
             }
         }

raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java

@@ -262,7 +262,7 @@ private void updateFollowerHighWatermark(
         OptionalLong highWatermarkOpt
     ) {
         highWatermarkOpt.ifPresent(highWatermark -> {
-            long newHighWatermark = Math.min(endOffset().offset, highWatermark);
+            long newHighWatermark = Math.min(endOffset().offset(), highWatermark);
             if (state.updateHighWatermark(OptionalLong.of(newHighWatermark))) {
                 logger.debug("Follower high watermark updated to {}", newHighWatermark);
                 log.updateHighWatermark(new LogOffsetMetadata(newHighWatermark));
@@ -884,13 +884,13 @@ private FetchResponseData buildFetchResponse(
             switch (validOffsetAndEpoch.kind()) {
                 case DIVERGING:
                     partitionData.divergingEpoch()
-                        .setEpoch(validOffsetAndEpoch.offsetAndEpoch().epoch)
-                        .setEndOffset(validOffsetAndEpoch.offsetAndEpoch().offset);
+                        .setEpoch(validOffsetAndEpoch.offsetAndEpoch().epoch())
+                        .setEndOffset(validOffsetAndEpoch.offsetAndEpoch().offset());
                     break;
                 case SNAPSHOT:
                     partitionData.snapshotId()
-                        .setEpoch(validOffsetAndEpoch.offsetAndEpoch().epoch)
-                        .setEndOffset(validOffsetAndEpoch.offsetAndEpoch().offset);
+                        .setEpoch(validOffsetAndEpoch.offsetAndEpoch().epoch())
+                        .setEndOffset(validOffsetAndEpoch.offsetAndEpoch().offset());
                     break;
                 default:
             }
@@ -1081,9 +1081,9 @@ private boolean handleFetchResponse(
                     divergingEpoch.endOffset(), divergingEpoch.epoch());
 
                 state.highWatermark().ifPresent(highWatermark -> {
-                    if (divergingOffsetAndEpoch.offset < highWatermark.offset) {
+                    if (divergingOffsetAndEpoch.offset() < highWatermark.offset) {
                         throw new KafkaException("The leader requested truncation to offset " +
-                            divergingOffsetAndEpoch.offset + ", which is below the current high watermark" +
+                            divergingOffsetAndEpoch.offset() + ", which is below the current high watermark" +
                             " " + highWatermark);
                     }
                 });
@@ -1288,8 +1288,8 @@ private FetchSnapshotResponseData handleFetchSnapshotRequest(
             responsePartitionSnapshot -> {
                 addQuorumLeader(responsePartitionSnapshot)
                     .snapshotId()
-                    .setEndOffset(snapshotId.offset)
-                    .setEpoch(snapshotId.epoch);
+                    .setEndOffset(snapshotId.offset())
+                    .setEpoch(snapshotId.epoch());
 
                 return responsePartitionSnapshot
                     .setSize(snapshotSize)
@@ -1771,8 +1771,8 @@ private VoteRequestData buildVoteRequest() {
             clusterId,
             quorum.epoch(),
             quorum.localIdOrThrow(),
-            endOffset.epoch,
-            endOffset.offset
+            endOffset.epoch(),
+            endOffset.offset()
         );
     }
 
@@ -1805,8 +1805,8 @@ private long maybeSendAnyVoterFetch(long currentTimeMs) {
 
     private FetchSnapshotRequestData buildFetchSnapshotRequest(OffsetAndEpoch snapshotId, long snapshotSize) {
         FetchSnapshotRequestData.SnapshotId requestSnapshotId = new FetchSnapshotRequestData.SnapshotId()
-            .setEpoch(snapshotId.epoch)
-            .setEndOffset(snapshotId.offset);
+            .setEpoch(snapshotId.epoch())
+            .setEndOffset(snapshotId.offset());
 
         FetchSnapshotRequestData request = FetchSnapshotRequest.singleton(
             clusterId,
@@ -1852,7 +1852,7 @@ private void appendBatch(
             LogAppendInfo info = appendAsLeader(batch.data);
             OffsetAndEpoch offsetAndEpoch = new OffsetAndEpoch(info.lastOffset, epoch);
             CompletableFuture<Long> future = appendPurgatory.await(
-                offsetAndEpoch.offset + 1, Integer.MAX_VALUE);
+                offsetAndEpoch.offset() + 1, Integer.MAX_VALUE);
 
             future.whenComplete((commitTimeMs, exception) -> {
                 if (exception != null) {
@@ -2494,7 +2494,7 @@ private synchronized OptionalLong nextExpectedOffset() {
          */
         private void fireHandleSnapshot(SnapshotReader<T> reader) {
             synchronized (this) {
-                nextOffset = reader.snapshotId().offset;
+                nextOffset = reader.snapshotId().offset();
                 lastSent = null;
             }
 

raft/src/main/java/org/apache/kafka/raft/OffsetAndEpoch.java

@@ -17,14 +17,22 @@
 package org.apache.kafka.raft;
 
 public class OffsetAndEpoch implements Comparable<OffsetAndEpoch> {
-    public final long offset;
-    public final int epoch;
+    private final long offset;
+    private final int epoch;
 
     public OffsetAndEpoch(long offset, int epoch) {
         this.offset = offset;
         this.epoch = epoch;
     }
 
+    public long offset() {
+        return offset;
+    }
+
+    public int epoch() {
+        return epoch;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) return true;

raft/src/main/java/org/apache/kafka/raft/QuorumState.java

@@ -139,13 +139,13 @@ public void initialize(OffsetAndEpoch logEndOffsetAndEpoch) throws IllegalStateE
             throw new IllegalStateException("Initialized quorum state " + election
                 + " with a voted candidate, which indicates this node was previously "
                 + " a voter, but the local id " + localIdDescription);
-        } else if (election.epoch < logEndOffsetAndEpoch.epoch) {
+        } else if (election.epoch < logEndOffsetAndEpoch.epoch()) {
             log.warn("Epoch from quorum-state file is {}, which is " +
                 "smaller than last written epoch {} in the log",
-                election.epoch, logEndOffsetAndEpoch.epoch);
+                election.epoch, logEndOffsetAndEpoch.epoch());
             initialState = new UnattachedState(
                 time,
-                logEndOffsetAndEpoch.epoch,
+                logEndOffsetAndEpoch.epoch(),
                 voters,
                 Optional.empty(),
                 randomElectionTimeoutMs(),

raft/src/main/java/org/apache/kafka/raft/ReplicatedLog.java

@@ -88,8 +88,8 @@ default ValidOffsetAndEpoch validateOffsetAndEpoch(long offset, int epoch) {
         Optional<OffsetAndEpoch> earliestSnapshotId = earliestSnapshotId();
         if (earliestSnapshotId.isPresent() &&
             ((offset < startOffset()) ||
-             (offset == startOffset() && epoch != earliestSnapshotId.get().epoch) ||
-             (epoch < earliestSnapshotId.get().epoch))
+             (offset == startOffset() && epoch != earliestSnapshotId.get().epoch()) ||
+             (epoch < earliestSnapshotId.get().epoch()))
         ) {
             /* Send a snapshot if the leader has a snapshot at the log start offset and
              * 1. the fetch offset is less than the log start offset or
@@ -108,7 +108,7 @@ default ValidOffsetAndEpoch validateOffsetAndEpoch(long offset, int epoch) {
         } else {
             OffsetAndEpoch endOffsetAndEpoch = endOffsetForEpoch(epoch);
 
-            if (endOffsetAndEpoch.epoch != epoch || endOffsetAndEpoch.offset < offset) {
+            if (endOffsetAndEpoch.epoch() != epoch || endOffsetAndEpoch.offset() < offset) {
                 return ValidOffsetAndEpoch.diverging(endOffsetAndEpoch);
             } else {
                 return ValidOffsetAndEpoch.valid(new OffsetAndEpoch(offset, epoch));
@@ -217,15 +217,15 @@ default ValidOffsetAndEpoch validateOffsetAndEpoch(long offset, int epoch) {
      */
     default long truncateToEndOffset(OffsetAndEpoch endOffset) {
         final long truncationOffset;
-        int leaderEpoch = endOffset.epoch;
+        int leaderEpoch = endOffset.epoch();
         if (leaderEpoch == 0) {
-            truncationOffset = Math.min(endOffset.offset, endOffset().offset);
+            truncationOffset = Math.min(endOffset.offset(), endOffset().offset);
         } else {
             OffsetAndEpoch localEndOffset = endOffsetForEpoch(leaderEpoch);
-            if (localEndOffset.epoch == leaderEpoch) {
-                truncationOffset = Math.min(localEndOffset.offset, endOffset.offset);
+            if (localEndOffset.epoch() == leaderEpoch) {
+                truncationOffset = Math.min(localEndOffset.offset(), endOffset.offset());
             } else {
-                truncationOffset = localEndOffset.offset;
+                truncationOffset = localEndOffset.offset();
             }
         }
 

raft/src/main/java/org/apache/kafka/raft/internals/KafkaRaftMetrics.java

@@ -102,10 +102,10 @@ public KafkaRaftMetrics(Metrics metrics, String metricGrpPrefix, QuorumState sta
         metrics.addMetric(this.highWatermarkMetricName, (mConfig, currentTimeMs) -> state.highWatermark().map(hw -> hw.offset).orElse(-1L));
 
         this.logEndOffsetMetricName = metrics.metricName("log-end-offset", metricGroupName, "The current raft log end offset.");
-        metrics.addMetric(this.logEndOffsetMetricName, (mConfig, currentTimeMs) -> logEndOffset.offset);
+        metrics.addMetric(this.logEndOffsetMetricName, (mConfig, currentTimeMs) -> logEndOffset.offset());
 
         this.logEndEpochMetricName = metrics.metricName("log-end-epoch", metricGroupName, "The current raft log end epoch.");
-        metrics.addMetric(this.logEndEpochMetricName, (mConfig, currentTimeMs) -> logEndOffset.epoch);
+        metrics.addMetric(this.logEndEpochMetricName, (mConfig, currentTimeMs) -> logEndOffset.epoch());
 
         this.numUnknownVoterConnectionsMetricName = metrics.metricName("number-unknown-voter-connections", metricGroupName,
                 "Number of unknown voters whose connection information is not cached; would never be larger than quorum-size.");

raft/src/main/java/org/apache/kafka/snapshot/RecordsSnapshotReader.java

@@ -49,12 +49,12 @@ public OffsetAndEpoch snapshotId() {
 
     @Override
     public long lastContainedLogOffset() {
-        return snapshotId.offset - 1;
+        return snapshotId.offset() - 1;
     }
 
     @Override
     public int lastContainedLogEpoch() {
-        return snapshotId.epoch;
+        return snapshotId.epoch();
     }
 
     @Override

raft/src/main/java/org/apache/kafka/snapshot/RecordsSnapshotWriter.java

@@ -52,7 +52,7 @@ private RecordsSnapshotWriter(
         this.lastContainedLogTimestamp = lastContainedLogTimestamp;
 
         this.accumulator = new BatchAccumulator<>(
-            snapshot.snapshotId().epoch,
+            snapshot.snapshotId().epoch(),
             0,
             Integer.MAX_VALUE,
             maxBatchSize,
@@ -140,12 +140,12 @@ public OffsetAndEpoch snapshotId() {
 
     @Override
     public long lastContainedLogOffset() {
-        return snapshot.snapshotId().offset - 1;
+        return snapshot.snapshotId().offset() - 1;
     }
 
     @Override
     public int lastContainedLogEpoch() {
-        return snapshot.snapshotId().epoch;
+        return snapshot.snapshotId().epoch();
     }
 
     @Override
@@ -164,7 +164,7 @@ public void append(List<T> records) {
             throw new IllegalStateException(message);
         }
 
-        accumulator.append(snapshot.snapshotId().epoch, records);
+        accumulator.append(snapshot.snapshotId().epoch(), records);
 
         if (accumulator.needsDrain(time.milliseconds())) {
             appendBatches(accumulator.drain());

raft/src/main/java/org/apache/kafka/snapshot/Snapshots.java

@@ -53,7 +53,7 @@ static Path snapshotDir(Path logDir) {
     }
 
     static String filenameFromSnapshotId(OffsetAndEpoch snapshotId) {
-        return String.format("%s-%s", OFFSET_FORMATTER.format(snapshotId.offset), EPOCH_FORMATTER.format(snapshotId.epoch));
+        return String.format("%s-%s", OFFSET_FORMATTER.format(snapshotId.offset()), EPOCH_FORMATTER.format(snapshotId.epoch()));
     }
 
     static Path moveRename(Path source, OffsetAndEpoch snapshotId) {