root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 2 Packages Projects Releases Wiki Activity

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

Reviewers: Christo Lolov <lolovc@amazon.com>
This commit is contained in:
Mickael Maison 2024-12-11 10:56:47 +01:00 committed by GitHub
parent 57737a357f
commit 7591868aea
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 35 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
core/src/main/java/kafka/log/remote/RemoteLogManager.java
View File

@ -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.server.storage.log.FetchIsolation;
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;

12
core/src/main/scala/kafka/log/OffsetResultHolder.scala
View File

@ -18,8 +18,7 @@ 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) {
@ -27,12 +26,3 @@ case class OffsetResultHolder(timestampAndOffsetOpt: Option[TimestampAndOffset],
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]) {
}

2
core/src/main/scala/kafka/server/ListOffsetsPartitionStatus.scala
View File

@ -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],

2
core/src/test/java/kafka/log/remote/RemoteLogOffsetReaderTest.java
View File

@ -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.log.remote.storage.RemoteStorageException;
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;

2
core/src/test/scala/integration/kafka/server/DelayedRemoteListOffsetsTest.scala
View File

@ -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

30
storage/src/main/java/org/apache/kafka/storage/internals/log/AsyncOffsetReadFutureHolder.java Normal file
View File

@ -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) {
}