KAFKA-18179: Move AsyncOffsetReadFutureHolder to storage module (#18095)

Reviewers: Christo Lolov <[email protected]>

core/src/main/java/kafka/log/remote/RemoteLogManager.java

@@ -17,7 +17,6 @@
 package kafka.log.remote;
 
 import kafka.cluster.Partition;
-import kafka.log.AsyncOffsetReadFutureHolder;
 import kafka.log.UnifiedLog;
 import kafka.server.DelayedRemoteListOffsets;
 
@@ -74,6 +73,7 @@
 import org.apache.kafka.storage.internals.checkpoint.LeaderEpochCheckpointFile;
 import org.apache.kafka.storage.internals.epoch.LeaderEpochFileCache;
 import org.apache.kafka.storage.internals.log.AbortedTxn;
+import org.apache.kafka.storage.internals.log.AsyncOffsetReadFutureHolder;
 import org.apache.kafka.storage.internals.log.EpochEntry;
 import org.apache.kafka.storage.internals.log.FetchDataInfo;
 import org.apache.kafka.storage.internals.log.LogOffsetMetadata;

core/src/main/scala/kafka/log/OffsetResultHolder.scala

@@ -18,21 +18,11 @@ package kafka.log
 
 import org.apache.kafka.common.errors.ApiException
 import org.apache.kafka.common.record.FileRecords.TimestampAndOffset
-
-import java.util.concurrent.{CompletableFuture, Future}
+import org.apache.kafka.storage.internals.log.AsyncOffsetReadFutureHolder
 
 case class OffsetResultHolder(timestampAndOffsetOpt: Option[TimestampAndOffset],
                               futureHolderOpt: Option[AsyncOffsetReadFutureHolder[Either[Exception, Option[TimestampAndOffset]]]] = None) {
 
   var maybeOffsetsError: Option[ApiException] = None
   var lastFetchableOffset: Option[Long] = None
 }
-
-/**
- * A remote log offset read task future holder. It contains two futures:
- * 1. JobFuture - Use this future to cancel the running job.
- * 2. TaskFuture - Use this future to get the result of the job/computation.
- */
-case class AsyncOffsetReadFutureHolder[T](jobFuture: Future[Void], taskFuture: CompletableFuture[T]) {
-
-}

core/src/main/scala/kafka/server/ListOffsetsPartitionStatus.scala

@@ -16,10 +16,10 @@
  */
 package kafka.server
 
-import kafka.log.AsyncOffsetReadFutureHolder
 import org.apache.kafka.common.errors.ApiException
 import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsPartitionResponse
 import org.apache.kafka.common.record.FileRecords.TimestampAndOffset
+import org.apache.kafka.storage.internals.log.AsyncOffsetReadFutureHolder
 
 class ListOffsetsPartitionStatus(val futureHolderOpt: Option[AsyncOffsetReadFutureHolder[Either[Exception, Option[TimestampAndOffset]]]],
                                  val lastFetchableOffset: Option[Long],

core/src/test/java/kafka/log/remote/RemoteLogOffsetReaderTest.java

@@ -16,7 +16,6 @@
  */
 package kafka.log.remote;
 
-import kafka.log.AsyncOffsetReadFutureHolder;
 import kafka.utils.TestUtils;
 
 import org.apache.kafka.common.TopicPartition;
@@ -28,6 +27,7 @@
 import org.apache.kafka.server.util.MockTime;
 import org.apache.kafka.storage.internals.checkpoint.LeaderEpochCheckpointFile;
 import org.apache.kafka.storage.internals.epoch.LeaderEpochFileCache;
+import org.apache.kafka.storage.internals.log.AsyncOffsetReadFutureHolder;
 import org.apache.kafka.storage.internals.log.LogDirFailureChannel;
 import org.apache.kafka.storage.log.metrics.BrokerTopicStats;
 

core/src/test/scala/integration/kafka/server/DelayedRemoteListOffsetsTest.scala

@@ -16,7 +16,6 @@
  */
 package kafka.server
 
-import kafka.log.AsyncOffsetReadFutureHolder
 import org.apache.kafka.common.TopicPartition
 import org.apache.kafka.common.errors.NotLeaderOrFollowerException
 import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsTopicResponse
@@ -25,6 +24,7 @@ import org.apache.kafka.common.record.FileRecords.TimestampAndOffset
 import org.apache.kafka.common.requests.ListOffsetsResponse
 import org.apache.kafka.server.purgatory.{DelayedOperationPurgatory, TopicPartitionOperationKey}
 import org.apache.kafka.server.util.timer.MockTimer
+import org.apache.kafka.storage.internals.log.AsyncOffsetReadFutureHolder
 import org.junit.jupiter.api.{AfterEach, Test}
 import org.junit.jupiter.api.Assertions.assertEquals
 import org.mockito.ArgumentMatchers.anyBoolean

storage/src/main/java/org/apache/kafka/storage/internals/log/AsyncOffsetReadFutureHolder.java

@@ -0,0 +1,30 @@
+/*
+ * 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.storage.internals.log;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Future;
+
+/**
+ * A remote log offset read task future holder. It contains two futures:
+ * <ol>
+ *   <li>JobFuture - Use this future to cancel the running job.
+ *   <li>TaskFuture - Use this future to get the result of the job/computation.
+ * </ol>
+ */
+public record AsyncOffsetReadFutureHolder<T>(Future<Void> jobFuture, CompletableFuture<T> taskFuture) {
+}