You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "OmniaGM (via GitHub)" <gi...@apache.org> on 2023/11/17 17:03:04 UTC
[PR] KAFKA-15363: Broker log directory failure changes [kafka]
OmniaGM opened a new pull request, #14790:
URL: https://github.com/apache/kafka/pull/14790
- Extend the ReplicaAlterLogDirsThread to send an AssignReplicasToDirsRequestData RPC before promoting the future replica in JBOD mode with KRAFT.
- Extend AssignmentsManager and DirectoryEventHandler to attach a callback to the events so we can track the state.
- The state of RPCs is kept in memory in ReplicaAlterLogDirsThread, and it gets updated using the callback that is sent to AssignmentsManager. Once we promote the future replica we clear the state of the RPCs related to this replica in question
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15363: Broker log directory failure changes [kafka]
Posted by "OmniaGM (via GitHub)" <gi...@apache.org>.
OmniaGM commented on code in PR #14790:
URL: https://github.com/apache/kafka/pull/14790#discussion_r1399605671
##########
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:
Which callback you are referring to here? The order of checking partition future replica is different between DirectoryEventHandler.NOOP and when we have a handler.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15363: Broker log directory failure changes [kafka]
Posted by "OmniaGM (via GitHub)" <gi...@apache.org>.
OmniaGM commented on code in PR #14790:
URL: https://github.com/apache/kafka/pull/14790#discussion_r1399670915
##########
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.")
Review Comment:
fixed
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15363: Broker log directory failure changes [kafka]
Posted by "soarez (via GitHub)" <gi...@apache.org>.
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15363: Broker log directory failure changes [kafka]
Posted by "rondagostino (via GitHub)" <gi...@apache.org>.
rondagostino commented on PR #14790:
URL: https://github.com/apache/kafka/pull/14790#issuecomment-1846221720
Ah, ok, we can keep it as it is, then. WiIl wait for build to finish and them merge assuming the test failures can be discounted.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15363: Broker log directory failure changes [kafka]
Posted by "OmniaGM (via GitHub)" <gi...@apache.org>.
OmniaGM commented on code in PR #14790:
URL: https://github.com/apache/kafka/pull/14790#discussion_r1412423207
##########
server/src/main/java/org/apache/kafka/server/AssignmentsManager.java:
##########
@@ -391,4 +402,4 @@ static AssignReplicasToDirsRequestData buildRequestData(int brokerId, long broke
.setBrokerEpoch(brokerEpoch)
.setDirectories(new ArrayList<>(directoryMap.values()));
}
-}
+}
Review Comment:
revert it
##########
server/src/test/java/org/apache/kafka/server/AssignmentsManagerTest.java:
##########
@@ -248,4 +248,4 @@ void testRequeuesFailedAssignmentPropagations() throws InterruptedException {
}}
), captor.getAllValues().get(4).build().data());
}
-}
+}
Review Comment:
revert it
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15363: Broker log directory failure changes [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #14790:
URL: https://github.com/apache/kafka/pull/14790#discussion_r1411471028
##########
core/src/main/scala/kafka/cluster/Partition.scala:
##########
@@ -613,7 +613,16 @@ class Partition(val topicPartition: TopicPartition,
// Only ReplicaAlterDirThread will call this method and ReplicaAlterDirThread should remove the partition
// from its partitionStates if this method returns true
def maybeReplaceCurrentWithFutureReplica(): Boolean = {
- // lock to prevent the log append by followers while checking if the log dir could be replaced with future log.
+ maybeFutureReplicaCaughtUp((futurePartitionLog: UnifiedLog) => {
+ logManager.replaceCurrentWithFutureLog(topicPartition)
+ futurePartitionLog.setLogOffsetsListener(logOffsetsListener)
+ log = futureLog
+ removeFutureLocalReplica(false)
+ })
+ }
+
+ def maybeFutureReplicaCaughtUp(callback: UnifiedLog => Unit): Boolean = {
Review Comment:
"maybeFutureReplicaCaughtUp" doesn't suggest anything to me. Maybe it did? So what?
how about a name for this function like "runCallbackIfFutureReplicaCaughtUp" or something.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15363: Broker log directory failure changes [kafka]
Posted by "OmniaGM (via GitHub)" <gi...@apache.org>.
OmniaGM commented on code in PR #14790:
URL: https://github.com/apache/kafka/pull/14790#discussion_r1399564930
##########
core/src/main/java/kafka/server/AssignmentsManager.java:
##########
@@ -210,6 +220,9 @@ public void run() throws Exception {
channelManager.sendRequest(new AssignReplicasToDirsRequest.Builder(
buildRequestData(brokerId, brokerEpochSupplier.get(), assignment)),
new AssignReplicasToDirsRequestCompletionHandler());
+ inflight.values().stream()
+ .filter(assignmentEvent -> assignmentEvent.callback != null)
+ .forEach(assignmentEvent -> assignmentEvent.callback.accept(DirectoryEventRequestState.DISPATCHED));
Review Comment:
For dispatched this is true but for queued I think we still need this so we don't send the reassignment multiple times to AssignmentManager. Especially that as far as I see the manager isn't idempotent.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15363: Broker log directory failure changes [kafka]
Posted by "OmniaGM (via GitHub)" <gi...@apache.org>.
OmniaGM commented on code in PR #14790:
URL: https://github.com/apache/kafka/pull/14790#discussion_r1399627137
##########
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:
I updated the pr to only deal with `QUEUED` and `COMPLETED`. And dropped `DISPATCHED`
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15363: Broker log directory failure changes [kafka]
Posted by "OmniaGM (via GitHub)" <gi...@apache.org>.
OmniaGM commented on code in PR #14790:
URL: https://github.com/apache/kafka/pull/14790#discussion_r1418894845
##########
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
+ directoryId <- partition.logDirectoryId()
+ topicIdPartition = new TopicIdPartition(topicId, topicPartition.partition())
+ } yield directoryEventHandler.handleAssignment(topicIdPartition, directoryId)
Review Comment:
No reason that I can call. I removed the `yield`
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15363: Broker log directory failure changes [kafka]
Posted by "rondagostino (via GitHub)" <gi...@apache.org>.
rondagostino commented on PR #14790:
URL: https://github.com/apache/kafka/pull/14790#issuecomment-1845784514
Seeing this locally:
```
org.mockito.exceptions.misusing.InvalidUseOfMatchersException:
Invalid use of argument matchers!
3 matchers expected, 2 recorded:
-> at kafka.server.ReplicaAlterLogDirsThreadTest.shouldRevertAnyScheduledAssignmentRequestIfAssignmentIsCancelled(ReplicaAlterLogDirsThreadTest.scala:461)
-> at kafka.server.ReplicaAlterLogDirsThreadTest.shouldRevertAnyScheduledAssignmentRequestIfAssignmentIsCancelled(ReplicaAlterLogDirsThreadTest.scala:461)
This exception may occur if matchers are combined with raw values:
//incorrect:
someMethod(any(), "raw String");
When using matchers, all arguments have to be provided by matchers.
For example:
//correct:
someMethod(any(), eq("String by matcher"));
For more info see javadoc for Matchers class.
at kafka.server.ReplicaAlterLogDirsThreadTest.shouldRevertAnyScheduledAssignmentRequestIfAssignmentIsCancelled(ReplicaAlterLogDirsThreadTest.scala:461)
...
```
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15363: Broker log directory failure changes [kafka]
Posted by "OmniaGM (via GitHub)" <gi...@apache.org>.
OmniaGM commented on code in PR #14790:
URL: https://github.com/apache/kafka/pull/14790#discussion_r1400640555
##########
core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala:
##########
@@ -40,6 +44,8 @@ class ReplicaAlterLogDirsThread(name: String,
isInterruptible = false,
brokerTopicStats) {
+ private val assignmentRequestStates: ConcurrentHashMap[TopicPartition, DirectoryEventRequestState] = new ConcurrentHashMap()
Review Comment:
I updated `removePartitions` which get invoked from `ReplicaManager.alterReplicaLogDirs` during cancellation to fire the revert of the assignment request if we have on queued.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15363: Broker log directory failure changes [kafka]
Posted by "OmniaGM (via GitHub)" <gi...@apache.org>.
OmniaGM commented on code in PR #14790:
URL: https://github.com/apache/kafka/pull/14790#discussion_r1400641749
##########
core/src/main/java/kafka/server/AssignmentsManager.java:
##########
@@ -336,6 +356,27 @@ private static boolean responseIsError(ClientResponse response) {
return false;
}
+ private static void applyCallbackOnComplete(
+ AssignReplicasToDirsResponseData data,
+ Map<TopicIdPartition, AssignmentEvent> sent) {
+ for (AssignReplicasToDirsResponseData.DirectoryData directory : data.directories()) {
+ for (AssignReplicasToDirsResponseData.TopicData topic : directory.topics()) {
+ for (AssignReplicasToDirsResponseData.PartitionData partition : topic.partitions()) {
+ TopicIdPartition topicPartition = new TopicIdPartition(topic.topicId(), partition.partitionIndex());
+ AssignmentEvent event = sent.get(topicPartition);
+ if (event == null) {
+ log.error("AssignReplicasToDirsResponse contains unexpected partition {} into directory {}. No callback to apply.", partition, directory.id());
+ } else {
+ Errors error = Errors.forCode(partition.errorCode());
+ if (error == Errors.NONE && event.callback != null) {
+ event.callback.accept(DirectoryEventRequestState.COMPLETED);
+ }
Review Comment:
good point, updated the code
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15363: Broker log directory failure changes [kafka]
Posted by "soarez (via GitHub)" <gi...@apache.org>.
soarez commented on code in PR #14790:
URL: https://github.com/apache/kafka/pull/14790#discussion_r1398483277
##########
core/src/main/java/kafka/server/AssignmentsManager.java:
##########
@@ -146,6 +153,9 @@ public void run() throws Exception {
log.debug("Received new assignment {}", this);
}
pending.put(partition, this);
+ if (callback != null) {
+ callback.accept(DirectoryEventRequestState.QUEUED);
+ }
Review Comment:
Is this necessary? Can't we simply assume it's queued as soon as `onAssingment` returns?
##########
core/src/main/java/kafka/server/AssignmentsManager.java:
##########
@@ -336,6 +356,27 @@ private static boolean responseIsError(ClientResponse response) {
return false;
}
+ private static void applyCallbackOnComplete(
+ AssignReplicasToDirsResponseData data,
+ Map<TopicIdPartition, AssignmentEvent> sent) {
+ for (AssignReplicasToDirsResponseData.DirectoryData directory : data.directories()) {
+ for (AssignReplicasToDirsResponseData.TopicData topic : directory.topics()) {
+ for (AssignReplicasToDirsResponseData.PartitionData partition : topic.partitions()) {
+ TopicIdPartition topicPartition = new TopicIdPartition(topic.topicId(), partition.partitionIndex());
+ AssignmentEvent event = sent.get(topicPartition);
+ if (event == null) {
+ log.error("AssignReplicasToDirsResponse contains unexpected partition {} into directory {}. No callback to apply.", partition, directory.id());
+ } else {
+ Errors error = Errors.forCode(partition.errorCode());
+ if (error == Errors.NONE && event.callback != null) {
+ event.callback.accept(DirectoryEventRequestState.COMPLETED);
+ }
Review Comment:
Instead of repeating the processing of the response for errors we can Set.diff the result of `filterFailures` against `inFlight` at the calling site for this function.
##########
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)(_))
Review Comment:
```suggestion
updatedAssignmentRequestState(topicPartition)(_))
```
##########
core/src/main/java/kafka/server/AssignmentsManager.java:
##########
@@ -210,6 +220,9 @@ public void run() throws Exception {
channelManager.sendRequest(new AssignReplicasToDirsRequest.Builder(
buildRequestData(brokerId, brokerEpochSupplier.get(), assignment)),
new AssignReplicasToDirsRequestCompletionHandler());
+ inflight.values().stream()
+ .filter(assignmentEvent -> assignmentEvent.callback != null)
+ .forEach(assignmentEvent -> assignmentEvent.callback.accept(DirectoryEventRequestState.DISPATCHED));
Review Comment:
I don't think we need the state in the callback. It makes sense to distinguish between the 3 states, but I don't see why we callback should fire for anything but `COMPLETED`.
##########
core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala:
##########
@@ -40,6 +44,8 @@ class ReplicaAlterLogDirsThread(name: String,
isInterruptible = false,
brokerTopicStats) {
+ private val assignmentRequestStates: ConcurrentHashMap[TopicPartition, DirectoryEventRequestState] = new ConcurrentHashMap()
Review Comment:
Don't we also need to consider this state when the reassignment is cancelled?
##########
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:
It seems we don't actually care about `DISPATCHED`, maybe we don't need that state.
##########
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.
Review Comment:
```suggestion
// Schedule assignment request and don't promote the future replica yet until the controller has accepted the request.
```
##########
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 = {
Review Comment:
```suggestion
def updatedAssignmentRequestState(topicPartition: TopicPartition)(state: DirectoryEventRequestState): Unit = {
```
##########
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.")
Review Comment:
```suggestion
throw new IllegalStateException(s"Topic ${topicPartition.topic()} does not have an ID")
```
##########
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:
Perhaps we make DirectoryEventHandler.NOOP immediately callback with success so we don't need this extra logic.
##########
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 _ =>
+ log.info("Waiting for AssignmentRequest to succeed before promoting the future replica.")
Review Comment:
```suggestion
log.trace("Waiting for AssignmentRequest to succeed before promoting the future replica.")
```
##########
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)(_))
Review Comment:
I think we can just update the state to `QUEUED` after the ` directoryEventHandler.handleAssignment()` call.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15363: Broker log directory failure changes [kafka]
Posted by "rondagostino (via GitHub)" <gi...@apache.org>.
rondagostino merged PR #14790:
URL: https://github.com/apache/kafka/pull/14790
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15363: Broker log directory failure changes [kafka]
Posted by "OmniaGM (via GitHub)" <gi...@apache.org>.
OmniaGM commented on code in PR #14790:
URL: https://github.com/apache/kafka/pull/14790#discussion_r1418888757
##########
core/src/test/scala/unit/kafka/server/ReplicaAlterLogDirsThreadTest.scala:
##########
@@ -268,6 +268,200 @@ class ReplicaAlterLogDirsThreadTest {
assertEquals(0, thread.partitionCount)
}
+ def updateAssignmentRequestState(thread: ReplicaAlterLogDirsThread, partitionId:Int, newState: ReplicaAlterLogDirsThread.DirectoryEventRequestState) = {
+ topicNames.get(topicId).map(topicName => {
+ thread.updatedAssignmentRequestState(new TopicPartition(topicName, partitionId))(newState)
+ })
+ }
+ @Test
Review Comment:
added a newline
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15363: Broker log directory failure changes [kafka]
Posted by "soarez (via GitHub)" <gi...@apache.org>.
soarez commented on code in PR #14790:
URL: https://github.com/apache/kafka/pull/14790#discussion_r1419076878
##########
server/src/main/java/org/apache/kafka/server/AssignmentsManager.java:
##########
@@ -95,7 +96,10 @@ public void close() throws InterruptedException {
}
public void onAssignment(TopicIdPartition topicPartition, Uuid dirId) {
Review Comment:
Should we do the same to `DirectoryEventHandler.handleAssignment` for consistency?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15363: Broker log directory failure changes [kafka]
Posted by "OmniaGM (via GitHub)" <gi...@apache.org>.
OmniaGM commented on code in PR #14790:
URL: https://github.com/apache/kafka/pull/14790#discussion_r1412455807
##########
core/src/main/scala/kafka/cluster/Partition.scala:
##########
@@ -613,7 +613,16 @@ class Partition(val topicPartition: TopicPartition,
// Only ReplicaAlterDirThread will call this method and ReplicaAlterDirThread should remove the partition
// from its partitionStates if this method returns true
def maybeReplaceCurrentWithFutureReplica(): Boolean = {
- // lock to prevent the log append by followers while checking if the log dir could be replaced with future log.
+ maybeFutureReplicaCaughtUp((futurePartitionLog: UnifiedLog) => {
+ logManager.replaceCurrentWithFutureLog(topicPartition)
+ futurePartitionLog.setLogOffsetsListener(logOffsetsListener)
+ log = futureLog
+ removeFutureLocalReplica(false)
+ })
+ }
+
+ def maybeFutureReplicaCaughtUp(callback: UnifiedLog => Unit): Boolean = {
Review Comment:
rename it.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15363: Broker log directory failure changes [kafka]
Posted by "OmniaGM (via GitHub)" <gi...@apache.org>.
OmniaGM commented on code in PR #14790:
URL: https://github.com/apache/kafka/pull/14790#discussion_r1399667766
##########
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)(_))
Review Comment:
on second thought, I changed the signature and structure of these callbacks and did some refactoring. Now AssignmentManager just runs the callback as runnable when the assignment is completed without explicitly sending the state `COMPLETE` to the callback method.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15363: Broker log directory failure changes [kafka]
Posted by "OmniaGM (via GitHub)" <gi...@apache.org>.
OmniaGM commented on code in PR #14790:
URL: https://github.com/apache/kafka/pull/14790#discussion_r1412422869
##########
server/src/main/java/org/apache/kafka/server/AssignmentsManager.java:
##########
@@ -95,7 +96,10 @@ public void close() throws InterruptedException {
}
public void onAssignment(TopicIdPartition topicPartition, Uuid dirId) {
Review Comment:
I changed this to have one `onAssignment` and `AssignmentEvent` constructor that will replace callback null with runnable that does nothing
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15363: Broker log directory failure changes [kafka]
Posted by "OmniaGM (via GitHub)" <gi...@apache.org>.
OmniaGM commented on code in PR #14790:
URL: https://github.com/apache/kafka/pull/14790#discussion_r1399669761
##########
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 _ =>
+ log.info("Waiting for AssignmentRequest to succeed before promoting the future replica.")
Review Comment:
changed the log level to trace
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15363: Broker log directory failure changes [kafka]
Posted by "rondagostino (via GitHub)" <gi...@apache.org>.
rondagostino commented on PR #14790:
URL: https://github.com/apache/kafka/pull/14790#issuecomment-1846426477
`66 tests have failed. There are 33 new tests failing, 33 existing failing.`
Many of them are common issues in recent build, and I ran several other test classes locally -- they all passed. Merging.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15363: Broker log directory failure changes [kafka]
Posted by "OmniaGM (via GitHub)" <gi...@apache.org>.
OmniaGM commented on PR #14790:
URL: https://github.com/apache/kafka/pull/14790#issuecomment-1846172826
> @OmniaGM The commit is worded as `KAFKA-15363: Broker log directory failure changes`. Is this correct? The branch is `KAFKA-15357`. So I'm a bit confused as to what is correct. The branch doesn't matter, but the commit should be reworded before this gets merged if it is currently not correct.
The branch name is wrong. But the commit message matches the Jira ticket. Do we want to name the Jira ticket?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15363: Broker log directory failure changes [kafka]
Posted by "OmniaGM (via GitHub)" <gi...@apache.org>.
OmniaGM commented on PR #14790:
URL: https://github.com/apache/kafka/pull/14790#issuecomment-1845803714
> Seeing this locally:
>
> ```
> org.mockito.exceptions.misusing.InvalidUseOfMatchersException:
> Invalid use of argument matchers!
> 3 matchers expected, 2 recorded:
> -> at kafka.server.ReplicaAlterLogDirsThreadTest.shouldRevertAnyScheduledAssignmentRequestIfAssignmentIsCancelled(ReplicaAlterLogDirsThreadTest.scala:461)
> -> at kafka.server.ReplicaAlterLogDirsThreadTest.shouldRevertAnyScheduledAssignmentRequestIfAssignmentIsCancelled(ReplicaAlterLogDirsThreadTest.scala:461)
>
> This exception may occur if matchers are combined with raw values:
> //incorrect:
> someMethod(any(), "raw String");
> When using matchers, all arguments have to be provided by matchers.
> For example:
> //correct:
> someMethod(any(), eq("String by matcher"));
>
> For more info see javadoc for Matchers class.
>
> at kafka.server.ReplicaAlterLogDirsThreadTest.shouldRevertAnyScheduledAssignmentRequestIfAssignmentIsCancelled(ReplicaAlterLogDirsThreadTest.scala:461)
> ...
> ```
fixed now
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15363: Broker log directory failure changes [kafka]
Posted by "rondagostino (via GitHub)" <gi...@apache.org>.
rondagostino commented on code in PR #14790:
URL: https://github.com/apache/kafka/pull/14790#discussion_r1418016489
##########
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
+ directoryId <- partition.logDirectoryId()
+ topicIdPartition = new TopicIdPartition(topicId, topicPartition.partition())
+ } yield directoryEventHandler.handleAssignment(topicIdPartition, directoryId)
+
+ super.removePartitions(topicPartitions)
+ }
+
+ // Visible for testing
+ def updatedAssignmentRequestState(topicPartition: TopicPartition)(state: ReplicaAlterLogDirsThread.DirectoryEventRequestState): Unit = {
+ assignmentRequestStates.put(topicPartition, state)
+ }
+ private def maybePromoteFutureReplica(topicPartition: TopicPartition, partition: Partition) = {
Review Comment:
nit: missing newline separator between defs
##########
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
+ directoryId <- partition.logDirectoryId()
+ topicIdPartition = new TopicIdPartition(topicId, topicPartition.partition())
+ } yield directoryEventHandler.handleAssignment(topicIdPartition, directoryId)
Review Comment:
Seems odd to use `yield` when `directoryEventHandler.handleAssignment()` is a void method. Is there a reason to use `yield`?
##########
core/src/test/scala/unit/kafka/server/ReplicaAlterLogDirsThreadTest.scala:
##########
@@ -268,6 +268,200 @@ class ReplicaAlterLogDirsThreadTest {
assertEquals(0, thread.partitionCount)
}
+ def updateAssignmentRequestState(thread: ReplicaAlterLogDirsThread, partitionId:Int, newState: ReplicaAlterLogDirsThread.DirectoryEventRequestState) = {
+ topicNames.get(topicId).map(topicName => {
+ thread.updatedAssignmentRequestState(new TopicPartition(topicName, partitionId))(newState)
+ })
+ }
+ @Test
Review Comment:
nit: missing newline separator
##########
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
+ directoryId <- partition.logDirectoryId()
+ topicIdPartition = new TopicIdPartition(topicId, topicPartition.partition())
+ } yield directoryEventHandler.handleAssignment(topicIdPartition, directoryId)
+
+ super.removePartitions(topicPartitions)
+ }
+
+ // Visible for testing
+ def updatedAssignmentRequestState(topicPartition: TopicPartition)(state: ReplicaAlterLogDirsThread.DirectoryEventRequestState): Unit = {
Review Comment:
`s/def/private[server] def/`
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15363: Broker log directory failure changes [kafka]
Posted by "soarez (via GitHub)" <gi...@apache.org>.
soarez commented on PR #14790:
URL: https://github.com/apache/kafka/pull/14790#issuecomment-1846145896
@rondagostino: The branch name is wrong/misleading. This is the JIRA https://issues.apache.org/jira/browse/KAFKA-15363
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15363: Broker log directory failure changes [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on PR #14790:
URL: https://github.com/apache/kafka/pull/14790#issuecomment-1847650922
I noticed PlaintextAdminIntegrationTest failing on all versions. It seems to have started on this commit
https://github.com/apache/kafka/pull/14790
Can we please take a look.
trunk build before: https://ci-builds.apache.org/blue/organizations/jenkins/Kafka%2Fkafka/detail/trunk/2460/tests
trunk build with commit: https://ci-builds.apache.org/blue/organizations/jenkins/Kafka%2Fkafka/detail/trunk/2461/tests
PR build: https://ci-builds.apache.org/blue/organizations/jenkins/Kafka%2Fkafka-pr/detail/PR-14790/16/
It is not sufficient to run the tests locally I think. Especially if they are failing on ALL versions of the PR build. Especially ESPECIALLY if the test are related to the feature you are changing.
I know it is near the feature freeze deadline but merging your changes makes it harder for everyone to keep the builds clean.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15363: Broker log directory failure changes [kafka]
Posted by "OmniaGM (via GitHub)" <gi...@apache.org>.
OmniaGM commented on code in PR #14790:
URL: https://github.com/apache/kafka/pull/14790#discussion_r1399559787
##########
core/src/main/java/kafka/server/AssignmentsManager.java:
##########
@@ -146,6 +153,9 @@ public void run() throws Exception {
log.debug("Received new assignment {}", this);
}
pending.put(partition, this);
+ if (callback != null) {
+ callback.accept(DirectoryEventRequestState.QUEUED);
+ }
Review Comment:
I think the Dispatched state is the questionable one. We may not need it and we keep the reassignment queued until it it is completed
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15363: Broker log directory failure changes [kafka]
Posted by "OmniaGM (via GitHub)" <gi...@apache.org>.
OmniaGM commented on code in PR #14790:
URL: https://github.com/apache/kafka/pull/14790#discussion_r1403404963
##########
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:
renamed it
##########
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:
Nope didn't fail. I think I removed it by mistake during refactoring.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15363: Broker log directory failure changes [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #14790:
URL: https://github.com/apache/kafka/pull/14790#discussion_r1411470125
##########
server/src/main/java/org/apache/kafka/server/AssignmentsManager.java:
##########
@@ -95,7 +96,10 @@ public void close() throws InterruptedException {
}
public void onAssignment(TopicIdPartition topicPartition, Uuid dirId) {
Review Comment:
Do we need two constructors for this? Seems like someone could just pass a callback that doesn't do anything, if they don't want any.
Also, I don't see a null check later for `assignmentEvent.callback` ... so is null even allowed here?
##########
server/src/main/java/org/apache/kafka/server/AssignmentsManager.java:
##########
@@ -391,4 +402,4 @@ static AssignReplicasToDirsRequestData buildRequestData(int brokerId, long broke
.setBrokerEpoch(brokerEpoch)
.setDirectories(new ArrayList<>(directoryMap.values()));
}
-}
+}
Review Comment:
don't need this change
##########
server/src/test/java/org/apache/kafka/server/AssignmentsManagerTest.java:
##########
@@ -248,4 +248,4 @@ void testRequeuesFailedAssignmentPropagations() throws InterruptedException {
}}
), captor.getAllValues().get(4).build().data());
}
-}
+}
Review Comment:
don't need this change
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15363: Broker log directory failure changes [kafka]
Posted by "rondagostino (via GitHub)" <gi...@apache.org>.
rondagostino commented on PR #14790:
URL: https://github.com/apache/kafka/pull/14790#issuecomment-1846126373
@OmniaGM The commit is worded as `KAFKA-15363: Broker log directory failure changes`. Is this correct? The branch is `KAFKA-15357`. So I'm a bit confused as to what is correct. The branch doesn't matter, but the commit should be reworded before this merged if it is currently not correct.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15363: Broker log directory failure changes [kafka]
Posted by "OmniaGM (via GitHub)" <gi...@apache.org>.
OmniaGM commented on PR #14790:
URL: https://github.com/apache/kafka/pull/14790#issuecomment-1847797722
> I noticed PlaintextAdminIntegrationTest failing on all versions. It seems to have started on this commit #14790 Can we please take a look.
>
> trunk build before: https://ci-builds.apache.org/blue/organizations/jenkins/Kafka%2Fkafka/detail/trunk/2460/tests trunk build with commit: https://ci-builds.apache.org/blue/organizations/jenkins/Kafka%2Fkafka/detail/trunk/2461/tests PR build: https://ci-builds.apache.org/blue/organizations/jenkins/Kafka%2Fkafka-pr/detail/PR-14790/16/
>
> It is not sufficient to run the tests locally I think. Especially if they are failing on ALL versions of the PR build. Especially ESPECIALLY if the test are related to the feature you are changing. I know it is near the feature freeze deadline but merging your changes makes it harder for everyone to keep the builds clean.
Hi @jolshan , I believe this will be fixed once #14954 gets merged
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15363: Broker log directory failure changes [kafka]
Posted by "OmniaGM (via GitHub)" <gi...@apache.org>.
OmniaGM commented on code in PR #14790:
URL: https://github.com/apache/kafka/pull/14790#discussion_r1418895320
##########
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
+ directoryId <- partition.logDirectoryId()
+ topicIdPartition = new TopicIdPartition(topicId, topicPartition.partition())
+ } yield directoryEventHandler.handleAssignment(topicIdPartition, directoryId)
+
+ super.removePartitions(topicPartitions)
+ }
+
+ // Visible for testing
+ def updatedAssignmentRequestState(topicPartition: TopicPartition)(state: ReplicaAlterLogDirsThread.DirectoryEventRequestState): Unit = {
+ assignmentRequestStates.put(topicPartition, state)
+ }
+ private def maybePromoteFutureReplica(topicPartition: TopicPartition, partition: Partition) = {
Review Comment:
added newline
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15363: Broker log directory failure changes [kafka]
Posted by "OmniaGM (via GitHub)" <gi...@apache.org>.
OmniaGM commented on code in PR #14790:
URL: https://github.com/apache/kafka/pull/14790#discussion_r1419294087
##########
server/src/main/java/org/apache/kafka/server/AssignmentsManager.java:
##########
@@ -95,7 +96,10 @@ public void close() throws InterruptedException {
}
public void onAssignment(TopicIdPartition topicPartition, Uuid dirId) {
Review Comment:
it is removed now.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15363: Broker log directory failure changes [kafka]
Posted by "OmniaGM (via GitHub)" <gi...@apache.org>.
OmniaGM commented on code in PR #14790:
URL: https://github.com/apache/kafka/pull/14790#discussion_r1399558750
##########
core/src/main/java/kafka/server/AssignmentsManager.java:
##########
@@ -146,6 +153,9 @@ public void run() throws Exception {
log.debug("Received new assignment {}", this);
}
pending.put(partition, this);
+ if (callback != null) {
+ callback.accept(DirectoryEventRequestState.QUEUED);
+ }
Review Comment:
We need to mark the request has been sent to we don't send it again and just wait on the state to be `COMPLETED`
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15363: Broker log directory failure changes [kafka]
Posted by "OmniaGM (via GitHub)" <gi...@apache.org>.
OmniaGM commented on code in PR #14790:
URL: https://github.com/apache/kafka/pull/14790#discussion_r1399627999
##########
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)(_))
Review Comment:
updated this, now `AssignmentManager` will only update the state to `COMPLETED` and `QUEUED` will be set immediately here.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org