dajac commented on code in PR #13474:
URL: https://github.com/apache/kafka/pull/13474#discussion_r1153140237
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java:
##########
@@ -1635,6 +1641,38 @@ public String toString() {
return "(version" + version + ": " + partitionsPerTopic + ")";
}
}
+ private static class RackInfo {
Review Comment:
nit: I would add an empty line before this one.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java:
##########
@@ -1613,14 +1615,18 @@ private ConsumerCoordinatorMetrics(Metrics metrics,
String metricGrpPrefix) {
private static class MetadataSnapshot {
private final int version;
- private final Map<String, Integer> partitionsPerTopic;
+ private final Map<String, List<RackInfo>> partitionsPerTopic;
Review Comment:
nit: I wonder if `PartitionInfo` would be a better name here. It would at
least make clearer that we have one `PartitionInfo` object per partition. When
I first read this, I was wondering why we have a list of RackInfo per topic.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java:
##########
@@ -1635,6 +1641,38 @@ public String toString() {
return "(version" + version + ": " + partitionsPerTopic + ")";
}
}
+ private static class RackInfo {
+ private final Set<String> racks;
+ RackInfo(Optional<String> clientRack, PartitionInfo partition) {
+ if (clientRack.isPresent() && partition.replicas() != null) {
+ racks =
Arrays.stream(partition.replicas()).map(Node::rack).collect(Collectors.toSet());
+ } else {
+ racks = Collections.emptySet();
+ }
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof RackInfo)) {
+ return false;
+ }
+ RackInfo rackInfo = (RackInfo) o;
+ return Objects.equals(racks, rackInfo.racks);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(racks);
+ }
+
+ @Override
+ public String toString() {
+ return racks.isEmpty() ? "NO_RACKS" : "racks=" + racks;
Review Comment:
nit: I wonder if we should include the partition index. That may be helpful
when debugging.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java:
##########
@@ -1635,6 +1641,38 @@ public String toString() {
return "(version" + version + ": " + partitionsPerTopic + ")";
}
}
+ private static class RackInfo {
+ private final Set<String> racks;
Review Comment:
nit: I would add an empty line after this one.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
