|
| 1 | +/** |
| 2 | + * Licensed to the Apache Software Foundation (ASF) under one or more |
| 3 | + * contributor license agreements. See the NOTICE file distributed with |
| 4 | + * this work for additional information regarding copyright ownership. |
| 5 | + * The ASF licenses this file to You under the Apache License, Version 2.0 |
| 6 | + * (the "License"); you may not use this file except in compliance with |
| 7 | + * the License. You may obtain a copy of the License at |
| 8 | + * |
| 9 | + * http://www.apache.org/licenses/LICENSE-2.0 |
| 10 | + * |
| 11 | + * Unless required by applicable law or agreed to in writing, software |
| 12 | + * distributed under the License is distributed on an "AS IS" BASIS, |
| 13 | + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 14 | + * See the License for the specific language governing permissions and |
| 15 | + * limitations under the License. |
| 16 | + */ |
| 17 | + |
| 18 | +package kafka.server |
| 19 | + |
| 20 | +import org.apache.kafka.common.TopicIdPartition |
| 21 | +import org.apache.kafka.common.errors._ |
| 22 | +import org.apache.kafka.common.protocol.Errors |
| 23 | +import org.apache.kafka.storage.internals.log.{FetchParams, FetchPartitionData, LogOffsetMetadata, RemoteLogReadResult, RemoteStorageFetchInfo} |
| 24 | + |
| 25 | +import java.util.concurrent.{CompletableFuture, Future} |
| 26 | +import java.util.{Optional, OptionalInt, OptionalLong} |
| 27 | +import scala.collection._ |
| 28 | + |
| 29 | +/** |
| 30 | + * A remote fetch operation that can be created by the replica manager and watched |
| 31 | + * in the remote fetch operation purgatory |
| 32 | + */ |
| 33 | +class DelayedRemoteFetch(remoteFetchTask: Future[Void], |
| 34 | + remoteFetchResult: CompletableFuture[RemoteLogReadResult], |
| 35 | + remoteFetchInfo: RemoteStorageFetchInfo, |
| 36 | + fetchPartitionStatus: Seq[(TopicIdPartition, FetchPartitionStatus)], |
| 37 | + fetchParams: FetchParams, |
| 38 | + localReadResults: Seq[(TopicIdPartition, LogReadResult)], |
| 39 | + replicaManager: ReplicaManager, |
| 40 | + responseCallback: Seq[(TopicIdPartition, FetchPartitionData)] => Unit) |
| 41 | + extends DelayedOperation(fetchParams.maxWaitMs) { |
| 42 | + |
| 43 | + /** |
| 44 | + * The operation can be completed if: |
| 45 | + * |
| 46 | + * Case a: This broker is no longer the leader of the partition it tries to fetch |
| 47 | + * Case b: This broker does not know the partition it tries to fetch |
| 48 | + * Case c: The remote storage read request completed (succeeded or failed) |
| 49 | + * Case d: The partition is in an offline log directory on this broker |
| 50 | + * |
| 51 | + * Upon completion, should return whatever data is available for each valid partition |
| 52 | + */ |
| 53 | + override def tryComplete(): Boolean = { |
| 54 | + fetchPartitionStatus.foreach { |
| 55 | + case (topicPartition, fetchStatus) => |
| 56 | + val fetchOffset = fetchStatus.startOffsetMetadata |
| 57 | + try { |
| 58 | + if (fetchOffset != LogOffsetMetadata.UNKNOWN_OFFSET_METADATA) { |
| 59 | + replicaManager.getPartitionOrException(topicPartition.topicPartition()) |
| 60 | + } |
| 61 | + } catch { |
| 62 | + case _: KafkaStorageException => // Case d |
| 63 | + debug(s"Partition $topicPartition is in an offline log directory, satisfy $fetchParams immediately") |
| 64 | + return forceComplete() |
| 65 | + case _: UnknownTopicOrPartitionException => // Case b |
| 66 | + debug(s"Broker no longer knows of partition $topicPartition, satisfy $fetchParams immediately") |
| 67 | + return forceComplete() |
| 68 | + case _: NotLeaderOrFollowerException => // Case a |
| 69 | + debug("Broker is no longer the leader or follower of %s, satisfy %s immediately".format(topicPartition, fetchParams)) |
| 70 | + return forceComplete() |
| 71 | + } |
| 72 | + } |
| 73 | + if (remoteFetchResult.isDone) // Case c |
| 74 | + forceComplete() |
| 75 | + else |
| 76 | + false |
| 77 | + } |
| 78 | + |
| 79 | + override def onExpiration(): Unit = { |
| 80 | + // cancel the remote storage read task, if it has not been executed yet |
| 81 | + val cancelled = remoteFetchTask.cancel(true) |
| 82 | + if (!cancelled) debug(s"Remote fetch task for for RemoteStorageFetchInfo: $remoteFetchInfo could not be cancelled and its isDone value is ${remoteFetchTask.isDone}") |
| 83 | + } |
| 84 | + |
| 85 | + /** |
| 86 | + * Upon completion, read whatever data is available and pass to the complete callback |
| 87 | + */ |
| 88 | + override def onComplete(): Unit = { |
| 89 | + val fetchPartitionData = localReadResults.map { case (tp, result) => |
| 90 | + if (tp.topicPartition().equals(remoteFetchInfo.topicPartition) |
| 91 | + && remoteFetchResult.isDone |
| 92 | + && result.error == Errors.NONE |
| 93 | + && result.info.delayedRemoteStorageFetch.isPresent) { |
| 94 | + if (remoteFetchResult.get.error.isPresent) { |
| 95 | + tp -> ReplicaManager.createLogReadResult(remoteFetchResult.get.error.get).toFetchPartitionData(false) |
| 96 | + } else { |
| 97 | + val info = remoteFetchResult.get.fetchDataInfo.get |
| 98 | + tp -> new FetchPartitionData( |
| 99 | + result.error, |
| 100 | + result.highWatermark, |
| 101 | + result.leaderLogStartOffset, |
| 102 | + info.records, |
| 103 | + Optional.empty(), |
| 104 | + if (result.lastStableOffset.isDefined) OptionalLong.of(result.lastStableOffset.get) else OptionalLong.empty(), |
| 105 | + info.abortedTransactions, |
| 106 | + if (result.preferredReadReplica.isDefined) OptionalInt.of(result.preferredReadReplica.get) else OptionalInt.empty(), |
| 107 | + false) |
| 108 | + } |
| 109 | + } else { |
| 110 | + tp -> result.toFetchPartitionData(false) |
| 111 | + } |
| 112 | + } |
| 113 | + |
| 114 | + responseCallback(fetchPartitionData) |
| 115 | + } |
| 116 | +} |
0 commit comments