apoorvmittal10 commented on code in PR #19846: URL: https://github.com/apache/kafka/pull/19846#discussion_r2113498469
########## server/src/main/java/org/apache/kafka/server/LogReadResult.java: ########## @@ -0,0 +1,129 @@ +/* + * 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; + +import org.apache.kafka.common.message.FetchResponseData; +import org.apache.kafka.common.protocol.Errors; +import org.apache.kafka.server.storage.log.FetchPartitionData; +import org.apache.kafka.storage.internals.log.FetchDataInfo; + +import java.util.Optional; +import java.util.OptionalInt; +import java.util.OptionalLong; + +/** + * Result metadata of a log read operation on the log + * + * @param info @FetchDataInfo returned by the @Log read + * @param divergingEpoch Optional epoch and end offset which indicates the largest epoch such + * that subsequent records are known to diverge on the follower/consumer + * @param highWatermark high watermark of the local replica + * @param leaderLogStartOffset The log start offset of the leader at the time of the read + * @param leaderLogEndOffset The log end offset of the leader at the time of the read + * @param followerLogStartOffset The log start offset of the follower taken from the Fetch request + * @param fetchTimeMs The time the fetch was received + * @param lastStableOffset Current LSO or None if the result has an exception + * @param preferredReadReplica the preferred read replica to be used for future fetches + * @param exception Exception if error encountered while reading from the log + */ +public record LogReadResult( + FetchDataInfo info, + Optional<FetchResponseData.EpochEndOffset> divergingEpoch, + long highWatermark, + long leaderLogStartOffset, + long leaderLogEndOffset, + long followerLogStartOffset, + long fetchTimeMs, + OptionalLong lastStableOffset, + OptionalInt preferredReadReplica, + Optional<Throwable> exception +) { + public LogReadResult( + FetchDataInfo info, + Optional<FetchResponseData.EpochEndOffset> divergingEpoch, + long highWatermark, + long leaderLogStartOffset, + long leaderLogEndOffset, + long followerLogStartOffset, + long fetchTimeMs, + OptionalLong lastStableOffset) { + this(info, divergingEpoch, highWatermark, leaderLogStartOffset, leaderLogEndOffset, followerLogStartOffset, + fetchTimeMs, lastStableOffset, OptionalInt.empty(), Optional.empty()); + } + + public LogReadResult( + FetchDataInfo info, + Optional<FetchResponseData.EpochEndOffset> divergingEpoch, + long highWatermark, + long leaderLogStartOffset, + long leaderLogEndOffset, + long followerLogStartOffset, + long fetchTimeMs, + OptionalLong lastStableOffset, + Optional<Throwable> exception) { + this(info, divergingEpoch, highWatermark, leaderLogStartOffset, leaderLogEndOffset, followerLogStartOffset, + fetchTimeMs, lastStableOffset, OptionalInt.empty(), exception); + } + + public LogReadResult( + FetchDataInfo info, + Optional<FetchResponseData.EpochEndOffset> divergingEpoch, + long highWatermark, + long leaderLogStartOffset, + long leaderLogEndOffset, + long followerLogStartOffset, + long fetchTimeMs, + OptionalLong lastStableOffset, + OptionalInt preferredReadReplica) { + this(info, divergingEpoch, highWatermark, leaderLogStartOffset, leaderLogEndOffset, followerLogStartOffset, + fetchTimeMs, lastStableOffset, preferredReadReplica, Optional.empty()); + } + + public Errors error() { + if (exception.isPresent()) { + return Errors.forException(exception.get()); + } + return Errors.NONE; + } + + @Override + public String toString() { + return "LogReadResult(info=" + info + + ", divergingEpoch=" + divergingEpoch + + ", highWatermark=" + highWatermark + + ", leaderLogStartOffset" + leaderLogStartOffset + + ", leaderLogEndOffset" + leaderLogEndOffset + + ", followerLogStartOffset" + followerLogStartOffset + + ", fetchTimeMs=" + fetchTimeMs + + ", preferredReadReplica=" + preferredReadReplica + + ", lastStableOffset=" + lastStableOffset + + ", error=" + error() + ")"; + } + + public FetchPartitionData toFetchPartitionData(boolean isReassignmentFetch) { + return new FetchPartitionData( + this.error(), + this.highWatermark, + this.leaderLogStartOffset, + this.info.records, + this.divergingEpoch, + this.lastStableOffset.isPresent() ? OptionalLong.of(this.lastStableOffset.getAsLong()) : OptionalLong.empty(), Review Comment: Why not just simply return the value? Earlier in scala it was Option to Optional... conversion, seems not required now. ```suggestion this.lastStableOffset, ``` ########## core/src/main/java/kafka/server/share/DelayedShareFetch.java: ########## @@ -849,9 +849,9 @@ private void completeRemoteStorageShareFetchRequest() { logReadResult.leaderLogStartOffset(), info.records, Optional.empty(), - logReadResult.lastStableOffset().isDefined() ? OptionalLong.of((Long) logReadResult.lastStableOffset().get()) : OptionalLong.empty(), + logReadResult.lastStableOffset().isPresent() ? OptionalLong.of((Long) logReadResult.lastStableOffset().getAsLong()) : OptionalLong.empty(), Review Comment: Do we still need the casting (Long)? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
