soarez commented on code in PR #14790:
URL: https://github.com/apache/kafka/pull/14790#discussion_r1402362044
##########
server/src/main/java/org/apache/kafka/server/AssignmentsManager.java:
##########
@@ -231,13 +238,12 @@ public void run() throws Exception {
} else {
failedAttempts = 0;
AssignReplicasToDirsResponseData data =
((AssignReplicasToDirsResponse) response.responseBody()).data();
+
Set<AssignmentEvent> failed = filterFailures(data, inflight);
+ Set<AssignmentEvent> completed = Utils.diff(HashSet::new,
inflight.values().stream().collect(Collectors.toSet()), failed);
+ completed.forEach(assignmentEvent ->
assignmentEvent.callback.run());
+
log.warn("Re-queueing assignments: {}", failed);
- if (!failed.isEmpty()) {
- for (AssignmentEvent event : failed) {
- pending.put(event.partition, event);
- }
- }
Review Comment:
I think we need to requeue these, did this not break the tests in
AssignmentsManagerTest?
##########
core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala:
##########
@@ -76,13 +82,49 @@ class ReplicaAlterLogDirsThread(name: String,
futureLog.updateHighWatermark(partitionData.highWatermark)
futureLog.maybeIncrementLogStartOffset(partitionData.logStartOffset,
LogStartOffsetIncrementReason.LeaderOffsetIncremented)
- if (partition.maybeReplaceCurrentWithFutureReplica())
- removePartitions(Set(topicPartition))
+ directoryEventHandler match {
+ case DirectoryEventHandler.NOOP =>
+ if (partition.maybeReplaceCurrentWithFutureReplica())
+ removePartitions(Set(topicPartition))
+ case _ =>
+ maybePromoteFutureReplica(topicPartition, partition)
+ }
quota.record(records.sizeInBytes)
logAppendInfo
}
+ // Visible for testing
+ def updatedAssignmentRequestStat(topicPartition: TopicPartition)(state:
DirectoryEventRequestState): Unit = {
+ assignmentRequestStates.put(topicPartition, state)
+ }
+ private def maybePromoteFutureReplica(topicPartition: TopicPartition,
partition: Partition) = {
+ val partitionRequestState =
Option(assignmentRequestStates.get(topicPartition))
+ val topicId = partition.topicId
+ if (topicId.isEmpty)
+ throw new IllegalStateException(s"Topic ${topicPartition.topic()} exists
but its ID doesn't exist.")
+
+ partitionRequestState match {
+ case None =>
+ // Schedule assignment request and don't promote the future replica
yet until the controller accepted the request.
+ partition.maybeFutureReplicaCaughtUp(_ => {
+ partition.futureReplicaDirectoryId()
+ .map {
+ directoryEventHandler.handleAssignment(new
TopicIdPartition(topicId.get, topicPartition.partition()), _,
+ updatedAssignmentRequestStat(topicPartition)(_))
+ }
+ })
+ case Some(DirectoryEventRequestState.COMPLETED) =>
+ // Promote future replica if controller accepted the request and the
replica caught-up with the original log.
+ if (partition.maybeReplaceCurrentWithFutureReplica()) {
+ removePartitions(Set(topicPartition))
+ assignmentRequestStates.remove(topicPartition)
+ }
+ case _ =>
Review Comment:
Looks better. Thanks
##########
server-common/src/main/java/org/apache/kafka/server/common/DirectoryEventHandler.java:
##########
@@ -19,13 +19,16 @@
import org.apache.kafka.common.Uuid;
+import java.util.function.Consumer;
Review Comment:
Looks like this is unused. Did you run checkstyle?
##########
core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala:
##########
@@ -76,13 +83,69 @@ class ReplicaAlterLogDirsThread(name: String,
futureLog.updateHighWatermark(partitionData.highWatermark)
futureLog.maybeIncrementLogStartOffset(partitionData.logStartOffset,
LogStartOffsetIncrementReason.LeaderOffsetIncremented)
- if (partition.maybeReplaceCurrentWithFutureReplica())
- removePartitions(Set(topicPartition))
+ directoryEventHandler match {
+ case DirectoryEventHandler.NOOP =>
+ if (partition.maybeReplaceCurrentWithFutureReplica())
+ removePartitions(Set(topicPartition))
+ case _ =>
+ maybePromoteFutureReplica(topicPartition, partition)
+ }
quota.record(records.sizeInBytes)
logAppendInfo
}
+ override def removePartitions(topicPartitions: Set[TopicPartition]):
Map[TopicPartition, PartitionFetchState] = {
+ // Schedule assignment request to revert any queued request before
cancelling
+ for {
+ topicPartition <- topicPartitions
+ partitionState <- partitionAssignmentRequestState(topicPartition)
+ if partitionState == QUEUED
+ partition = replicaMgr.getPartitionOrException(topicPartition)
+ topicId <- partition.topicId
+ logId <- partition.logDirectoryId()
Review Comment:
`logId` might be confusing, log directories contain logs but they're not the
same thing, I think a better choice would be `directory` or `directoryId` or
`logDirectory` etc
##########
core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala:
##########
@@ -76,13 +82,49 @@ class ReplicaAlterLogDirsThread(name: String,
futureLog.updateHighWatermark(partitionData.highWatermark)
futureLog.maybeIncrementLogStartOffset(partitionData.logStartOffset,
LogStartOffsetIncrementReason.LeaderOffsetIncremented)
- if (partition.maybeReplaceCurrentWithFutureReplica())
- removePartitions(Set(topicPartition))
+ directoryEventHandler match {
+ case DirectoryEventHandler.NOOP =>
+ if (partition.maybeReplaceCurrentWithFutureReplica())
+ removePartitions(Set(topicPartition))
+ case _ =>
+ maybePromoteFutureReplica(topicPartition, partition)
+ }
Review Comment:
I was suggesting that NOOP immediatly calls the `Runnable callback` in
`handleAssignment()`, but given the current logic in
`maybePromoteFutureReplica` think my suggestion isn't as straightforward as I
was imagining.
--
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
