You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "soarez (via GitHub)" <gi...@apache.org> on 2023/11/11 00:05:36 UTC
[PR] KAFKA-15362: Resolve offline replicas in metadata cache [kafka]
soarez opened a new pull request, #14737:
URL: https://github.com/apache/kafka/pull/14737
The metadata cache now considers registered log directories and directory assignments when determining offline replicas.
### 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-15362: Resolve offline replicas in metadata cache [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #14737:
URL: https://github.com/apache/kafka/pull/14737#discussion_r1396539701
##########
metadata/src/main/java/org/apache/kafka/metadata/BrokerRegistration.java:
##########
@@ -50,6 +54,7 @@ public static class Builder {
private boolean fenced = false;
private boolean inControlledShutdown = false;
private boolean isMigratingZkBroker = false;
+ private Set<Uuid> directories;
Review Comment:
I think a sorted array of directories would be easier to work with.
We need to pass an array to most of places that are using this.
Also, if the array is sorted, we can always do a binary search for whatever value we want (although given the list is short, that might not matter much either).
--
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-15362: Resolve offline replicas in metadata cache [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #14737:
URL: https://github.com/apache/kafka/pull/14737#discussion_r1396538658
##########
metadata/src/main/java/org/apache/kafka/metadata/PartitionRegistration.java:
##########
@@ -337,6 +338,15 @@ public int preferredReplica() {
return replicas.length == 0 ? LeaderConstants.NO_LEADER : replicas[0];
}
+ public Uuid directory(int replica) {
+ Map<Integer, Uuid> assignment = DirectoryId.createAssignmentMap(replicas, directories);
Review Comment:
I don't think it makes sense to create a whole map just for this function, then throw it away
If we're going to iterate over all the replicas, just find the index of the one that was passed in, and then return `directories[index]`
--
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-15362: Resolve offline replicas in metadata cache [kafka]
Posted by "soarez (via GitHub)" <gi...@apache.org>.
soarez commented on PR #14737:
URL: https://github.com/apache/kafka/pull/14737#issuecomment-1806580244
@pprovenzano @cmccabe please have a look
--
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-15362: Resolve offline replicas in metadata cache [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #14737:
URL: https://github.com/apache/kafka/pull/14737#discussion_r1396536766
##########
metadata/src/main/java/org/apache/kafka/metadata/BrokerRegistration.java:
##########
@@ -118,6 +124,13 @@ public Builder setIsMigratingZkBroker(boolean isMigratingZkBroker) {
return this;
}
+ public Builder setDirectories(Collection<Uuid> directories) {
Review Comment:
I think it would be better to just unconditionally copy this. It's a bit of a tricky pattern to sometimes copy and sometimes not (doesn't play well with mutability). Since this isn't a performance hotspot, it would be good to avoid.
--
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-15362: Resolve offline replicas in metadata cache [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on PR #14737:
URL: https://github.com/apache/kafka/pull/14737#issuecomment-1820033474
@soarez : This PR is causing failures in `KRaftClusterTest.testCreateClusterAndPerformReassignment`
```
Gradle Test Run :core:test > Gradle Test Executor 2 > KRaftClusterTest > testCreateClusterAndPerformReassignment() FAILED
java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.UnknownServerException: The server experienced an unexpected error when processing the request.
at java.base/java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:396)
at java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2073)
at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:165)
at kafka.server.KRaftClusterTest.testCreateClusterAndPerformReassignment(KRaftClusterTest.scala:479)
Caused by:
org.apache.kafka.common.errors.UnknownServerException: The server experienced an unexpected error when processing the request.```
--
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-15362: Resolve offline replicas in metadata cache [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #14737:
URL: https://github.com/apache/kafka/pull/14737#discussion_r1396538894
##########
metadata/src/test/java/org/apache/kafka/image/node/ClusterImageBrokersNodeTest.java:
##########
@@ -66,7 +66,7 @@ public void testNode1Child() {
"rack=Optional.empty, " +
"fenced=false, " +
"inControlledShutdown=false, " +
- "isMigratingZkBroker=false)", child.stringify());
+ "isMigratingZkBroker=false, directories=[])", child.stringify());
Review Comment:
let's have some output in this test, not just an empty directory, to verify that this all works
--
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-15362: Resolve offline replicas in metadata cache [kafka]
Posted by "soarez (via GitHub)" <gi...@apache.org>.
soarez commented on PR #14737:
URL: https://github.com/apache/kafka/pull/14737#issuecomment-1821057902
@cmccabe thanks for checking. tests are passing 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-15362: Resolve offline replicas in metadata cache [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe merged PR #14737:
URL: https://github.com/apache/kafka/pull/14737
--
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-15362: Resolve offline replicas in metadata cache [kafka]
Posted by "soarez (via GitHub)" <gi...@apache.org>.
soarez commented on PR #14737:
URL: https://github.com/apache/kafka/pull/14737#issuecomment-1816570498
Thanks for the review @cmccabe. I think I have incorporated all your suggestions. PTAL
--
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-15362: Resolve offline replicas in metadata cache [kafka]
Posted by "pprovenzano (via GitHub)" <gi...@apache.org>.
pprovenzano commented on code in PR #14737:
URL: https://github.com/apache/kafka/pull/14737#discussion_r1397547750
##########
core/src/main/scala/kafka/server/metadata/KRaftMetadataCache.scala:
##########
@@ -143,21 +143,30 @@ class KRaftMetadataCache(val brokerId: Int) extends MetadataCache with Logging w
private def getOfflineReplicas(image: MetadataImage,
partition: PartitionRegistration,
listenerName: ListenerName): util.List[Integer] = {
- // TODO: in order to really implement this correctly, we would need JBOD support.
- // That would require us to track which replicas were offline on a per-replica basis.
- // See KAFKA-13005.
val offlineReplicas = new util.ArrayList[Integer](0)
for (brokerId <- partition.replicas) {
Option(image.cluster().broker(brokerId)) match {
case None => offlineReplicas.add(brokerId)
- case Some(broker) => if (broker.fenced() || !broker.listeners().containsKey(listenerName.value())) {
+ case Some(broker) => if (isReplicaOffline(partition, listenerName, broker)) {
offlineReplicas.add(brokerId)
}
}
}
offlineReplicas
}
+ private def isReplicaOffline(partition: PartitionRegistration, listenerName: ListenerName, broker: BrokerRegistration) = {
+ broker.fenced() || !broker.listeners().containsKey(listenerName.value()) || isInOfflineDir(broker, partition)
+ }
+
+ private def isInOfflineDir(broker: BrokerRegistration, partition: PartitionRegistration): Boolean = {
Review Comment:
Do we have to handle UNASSIGNED here?
It would return false.
--
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-15362: Resolve offline replicas in metadata cache [kafka]
Posted by "soarez (via GitHub)" <gi...@apache.org>.
soarez commented on code in PR #14737:
URL: https://github.com/apache/kafka/pull/14737#discussion_r1397572189
##########
core/src/main/scala/kafka/server/metadata/KRaftMetadataCache.scala:
##########
@@ -143,21 +143,30 @@ class KRaftMetadataCache(val brokerId: Int) extends MetadataCache with Logging w
private def getOfflineReplicas(image: MetadataImage,
partition: PartitionRegistration,
listenerName: ListenerName): util.List[Integer] = {
- // TODO: in order to really implement this correctly, we would need JBOD support.
- // That would require us to track which replicas were offline on a per-replica basis.
- // See KAFKA-13005.
val offlineReplicas = new util.ArrayList[Integer](0)
for (brokerId <- partition.replicas) {
Option(image.cluster().broker(brokerId)) match {
case None => offlineReplicas.add(brokerId)
- case Some(broker) => if (broker.fenced() || !broker.listeners().containsKey(listenerName.value())) {
+ case Some(broker) => if (isReplicaOffline(partition, listenerName, broker)) {
offlineReplicas.add(brokerId)
}
}
}
offlineReplicas
}
+ private def isReplicaOffline(partition: PartitionRegistration, listenerName: ListenerName, broker: BrokerRegistration) = {
+ broker.fenced() || !broker.listeners().containsKey(listenerName.value()) || isInOfflineDir(broker, partition)
+ }
+
+ private def isInOfflineDir(broker: BrokerRegistration, partition: PartitionRegistration): Boolean = {
Review Comment:
Good point. Missed that. Updated, thanks!
--
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