You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "rreddy-22 (via GitHub)" <gi...@apache.org> on 2023/07/25 22:34:45 UTC
[GitHub] [kafka] rreddy-22 opened a new pull request, #14099: KAFKA-14702:Extend server side assignor to support rack aware replica placement
rreddy-22 opened a new pull request, #14099:
URL: https://github.com/apache/kafka/pull/14099
We want to make the assignors rack aware and in order to obtain and provide the rack Information the following interface modifications were made:-
SubcribedTopicDescribor - Interface passed to the assignor to obtain topic and partition metadata
SubcribedTopicMetadata - Implementation of the interface above
Add partitionMetadata field to the ConsumerGroupPartitionMetadataValue.json
TopicMetadata has a new attribute Map<Integer, Set> partitionRackInfo
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1276666851
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -5692,12 +5725,60 @@ private void assertApiMessageAndVersionEquals(
fromTopicPartitions(actualValue.partitionsPendingRevocation()));
assertEquals(fromTopicPartitions(expectedValue.partitionsPendingAssignment()),
fromTopicPartitions(actualValue.partitionsPendingAssignment()));
- } else {
- assertEquals(expected.message(), actual.message());
+ } else if (actual.message() instanceof ConsumerGroupPartitionMetadataValue) {
+ // The order of the racks stored in PartitionMetadata of ConsumerGroupPartitionMetadataValue is not
+ // always guaranteed. Therefore, we need a special comparator.
+ ConsumerGroupPartitionMetadataValue expectedValue =
+ (ConsumerGroupPartitionMetadataValue) expected.message();
+ ConsumerGroupPartitionMetadataValue actualValue =
+ (ConsumerGroupPartitionMetadataValue) actual.message();
+
+ List<ConsumerGroupPartitionMetadataValue.TopicMetadata> expectedTopicMetadataList =
+ expectedValue.topics();
+ List<ConsumerGroupPartitionMetadataValue.TopicMetadata> actualTopicMetadataList =
+ actualValue.topics();
+
+ if (expectedTopicMetadataList.size() != actualTopicMetadataList.size()) {
+ fail("Topic metadata lists have different sizes");
+ }
+
+ for (int i = 0; i < expectedValue.topics().size(); i++) {
+ ConsumerGroupPartitionMetadataValue.TopicMetadata expectedTopicMetadata =
+ expectedTopicMetadataList.get(i);
+ ConsumerGroupPartitionMetadataValue.TopicMetadata actualTopicMetadata =
+ actualTopicMetadataList.get(i);
+
+ assertEquals(expectedTopicMetadata.topicId(), actualTopicMetadata.topicId());
+ assertEquals(expectedTopicMetadata.topicName(), actualTopicMetadata.topicName());
+ assertEquals(expectedTopicMetadata.numPartitions(), actualTopicMetadata.numPartitions());
+
+ List<ConsumerGroupPartitionMetadataValue.PartitionMetadata> expectedPartitionMetadataList =
+ expectedTopicMetadata.partitionMetadata();
+ List<ConsumerGroupPartitionMetadataValue.PartitionMetadata> actualPartitionMetadataList =
+ actualTopicMetadata.partitionMetadata();
+
+ // If the list is empty, rack information wasn't available for any replica of
+ // the partition and hence, the entry wasn't added to the record.
+ if (expectedPartitionMetadataList.size() != actualPartitionMetadataList.size()) {
+ fail("Partition metadata lists have different sizes");
+ } else if (!expectedPartitionMetadataList.isEmpty() && !actualPartitionMetadataList.isEmpty()) {
+ for (int j = 0; j < expectedTopicMetadata.numPartitions(); j++) {
+ ConsumerGroupPartitionMetadataValue.PartitionMetadata expectedPartitionMetadata =
+ expectedPartitionMetadataList.get(j);
+ ConsumerGroupPartitionMetadataValue.PartitionMetadata actualPartitionMetadata =
+ actualPartitionMetadataList.get(j);
+
+ assertEquals(expectedPartitionMetadata.partition(), actualPartitionMetadata.partition());
+ assertUnorderedListEquals(expectedPartitionMetadata.racks(), actualPartitionMetadata.racks());
+ }
+ } else {
+ assertEquals(expected.message(), actual.message());
Review Comment:
Why are we doing this? If the partition metadata are empty, we could just run the previous branch as well. It would be a no-op.
--
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
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1276701201
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -5567,7 +5600,7 @@ public void testJoinGroupAppendErrorConversion() {
assertEquals(Errors.LEADER_NOT_AVAILABLE, Errors.LEADER_NOT_AVAILABLE);
}
- private <T> void assertUnorderedListEquals(
+ public static <T> void assertUnorderedListEquals(
Review Comment:
My bad I had to only make the recordEquals method public to be able to use in the RecordHelpersTest, reverted the others to private
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1274909957
##########
checkstyle/suppressions.xml:
##########
@@ -321,13 +321,13 @@
<!-- group coordinator -->
<suppress checks="CyclomaticComplexity"
+ files="(ConsumerGroupMember).java"/>
+ <suppress checks="ParameterNumber"
files="(ConsumerGroupMember|GroupMetadataManager).java"/>
<suppress checks="(NPathComplexity|MethodLength)"
files="(GroupMetadataManager|ConsumerGroupTest|GroupMetadataManagerTest).java"/>
<suppress checks="ClassFanOutComplexity"
files="(GroupMetadataManager|GroupMetadataManagerTest).java"/>
- <suppress checks="ParameterNumber"
- files="(ConsumerGroupMember|GroupMetadataManager).java"/>
Review Comment:
Why are we moving 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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1276677238
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroupTest.java:
##########
@@ -440,12 +442,13 @@ public void testUpdateSubscriptionMetadata() {
// It should return foo now.
assertEquals(
mkMap(
- mkEntry("foo", new TopicMetadata(fooTopicId, "foo", 1))
+ mkEntry("foo", new TopicMetadata(fooTopicId, "foo", 1, Collections.emptyMap()))
Review Comment:
Should we update `testUpdateSubscriptionMetadata` to include racks for some topics? Otherwise, it seems that we don't really test the newly added logic into `computeSubscriptionMetadata`.
--
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
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1275226238
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/RecordHelpers.java:
##########
@@ -131,13 +131,24 @@ public static Record newGroupSubscriptionMetadataRecord(
Map<String, TopicMetadata> newSubscriptionMetadata
) {
ConsumerGroupPartitionMetadataValue value = new ConsumerGroupPartitionMetadataValue();
- newSubscriptionMetadata.forEach((topicName, topicMetadata) ->
+ newSubscriptionMetadata.forEach((topicName, topicMetadata) -> {
+ List<ConsumerGroupPartitionMetadataValue.PartitionMetadata> partitionMetadata = new ArrayList<>();
+ if (!topicMetadata.partitionRacks().isEmpty()) {
+ topicMetadata.partitionRacks().forEach((partition, racks) -> {
+ partitionMetadata.add(new ConsumerGroupPartitionMetadataValue.PartitionMetadata()
+ .setPartition(partition)
+ .setRacks(new ArrayList<>(racks))
+ );
+ });
+ }
+ // If partition rack information is empty, store an empty list in the record.
value.topics().add(new ConsumerGroupPartitionMetadataValue.TopicMetadata()
.setTopicId(topicMetadata.id())
.setTopicName(topicMetadata.name())
.setNumPartitions(topicMetadata.numPartitions())
- )
- );
+ .setPartitionMetadata(partitionMetadata.isEmpty() ? Collections.emptyList() : partitionMetadata)
Review Comment:
I thought we prefer Collections.emptyList over new ArrayList()
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1276671418
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/TargetAssignmentBuilderTest.java:
##########
@@ -84,14 +85,15 @@ public void addGroupMember(
public Uuid addTopicMetadata(
String topicName,
- int numPartitions
+ int numPartitions,
+ Map<Integer, Set<String>> partitionRacks
) {
Uuid topicId = Uuid.randomUuid();
subscriptionMetadata.put(topicName, new TopicMetadata(
topicId,
topicName,
- numPartitions
- ));
+ numPartitions,
+ partitionRacks));
Review Comment:
nit: Let bring back the closing parenthesis as they were.
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1276659754
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -5692,12 +5725,60 @@ private void assertApiMessageAndVersionEquals(
fromTopicPartitions(actualValue.partitionsPendingRevocation()));
assertEquals(fromTopicPartitions(expectedValue.partitionsPendingAssignment()),
fromTopicPartitions(actualValue.partitionsPendingAssignment()));
- } else {
- assertEquals(expected.message(), actual.message());
+ } else if (actual.message() instanceof ConsumerGroupPartitionMetadataValue) {
+ // The order of the racks stored in PartitionMetadata of ConsumerGroupPartitionMetadataValue is not
+ // always guaranteed. Therefore, we need a special comparator.
+ ConsumerGroupPartitionMetadataValue expectedValue =
+ (ConsumerGroupPartitionMetadataValue) expected.message();
+ ConsumerGroupPartitionMetadataValue actualValue =
+ (ConsumerGroupPartitionMetadataValue) actual.message();
+
Review Comment:
Should we assert numPartitions as well?
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1276659754
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -5692,12 +5725,60 @@ private void assertApiMessageAndVersionEquals(
fromTopicPartitions(actualValue.partitionsPendingRevocation()));
assertEquals(fromTopicPartitions(expectedValue.partitionsPendingAssignment()),
fromTopicPartitions(actualValue.partitionsPendingAssignment()));
- } else {
- assertEquals(expected.message(), actual.message());
+ } else if (actual.message() instanceof ConsumerGroupPartitionMetadataValue) {
+ // The order of the racks stored in PartitionMetadata of ConsumerGroupPartitionMetadataValue is not
+ // always guaranteed. Therefore, we need a special comparator.
+ ConsumerGroupPartitionMetadataValue expectedValue =
+ (ConsumerGroupPartitionMetadataValue) expected.message();
+ ConsumerGroupPartitionMetadataValue actualValue =
+ (ConsumerGroupPartitionMetadataValue) actual.message();
+
Review Comment:
Should we assert numPartitions as well?
--
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
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1275225268
##########
checkstyle/suppressions.xml:
##########
@@ -321,13 +321,13 @@
<!-- group coordinator -->
<suppress checks="CyclomaticComplexity"
+ files="(ConsumerGroupMember).java"/>
+ <suppress checks="ParameterNumber"
files="(ConsumerGroupMember|GroupMetadataManager).java"/>
<suppress checks="(NPathComplexity|MethodLength)"
files="(GroupMetadataManager|ConsumerGroupTest|GroupMetadataManagerTest).java"/>
<suppress checks="ClassFanOutComplexity"
files="(GroupMetadataManager|GroupMetadataManagerTest).java"/>
- <suppress checks="ParameterNumber"
- files="(ConsumerGroupMember|GroupMetadataManager).java"/>
Review Comment:
You had put a comment to remove it in the last PR
--
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
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1275362470
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/TargetAssignmentBuilderTest.java:
##########
@@ -222,8 +231,25 @@ public void testCreateAssignmentMemberSpec() {
Map<String, TopicMetadata> subscriptionMetadata = new HashMap<String, TopicMetadata>() {
{
- put("foo", new TopicMetadata(fooTopicId, "foo", 5));
- put("bar", new TopicMetadata(barTopicId, "bar", 5));
+ put("foo", new TopicMetadata(fooTopicId, "foo", 5,
+ mkMap(
+ mkEntry(0, Collections.emptySet()),
+ mkEntry(1, Collections.emptySet()),
+ mkEntry(2, Collections.emptySet()),
+ mkEntry(3, Collections.emptySet()),
+ mkEntry(4, Collections.emptySet())
+ )
Review Comment:
I kept it just in case we wanted to add the rack information, also wanted some tests with a map with information rather than just an empty map but since these tests aren't related to the rack info I'll just pass the empty map for most and add a test that checks for the rack info
--
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
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1276612481
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/RecordHelpersTest.java:
##########
@@ -612,15 +616,14 @@ public void testNewGroupMetadataRecordThrowsWhenEmptyAssignment() {
MetadataVersion.IBP_3_5_IV2
));
}
-
Review Comment:
just a line after the test, I think it can stay
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1276670286
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/RangeAssignorTest.java:
##########
@@ -571,4 +716,14 @@ private void assertAssignment(Map<String, Map<Uuid, Set<Integer>>> expectedAssig
assertEquals(expectedAssignment.get(memberId), computedAssignmentForMember);
}
}
+
+ // When rack awareness is enabled for this assignor, rack information can be updated in this method.
+ private Map<Integer, Set<String>> createPartitionMetadata(int numPartitions) {
Review Comment:
nit: static?
--
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
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1275306576
##########
checkstyle/suppressions.xml:
##########
@@ -321,13 +321,13 @@
<!-- group coordinator -->
<suppress checks="CyclomaticComplexity"
+ files="(ConsumerGroupMember).java"/>
+ <suppress checks="ParameterNumber"
files="(ConsumerGroupMember|GroupMetadataManager).java"/>
<suppress checks="(NPathComplexity|MethodLength)"
files="(GroupMetadataManager|ConsumerGroupTest|GroupMetadataManagerTest).java"/>
<suppress checks="ClassFanOutComplexity"
files="(GroupMetadataManager|GroupMetadataManagerTest).java"/>
- <suppress checks="ParameterNumber"
- files="(ConsumerGroupMember|GroupMetadataManager).java"/>
Review Comment:
sorry I did move it, when I double checked in the new PR I thought this moving was what I did, my bad
--
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
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1275306576
##########
checkstyle/suppressions.xml:
##########
@@ -321,13 +321,13 @@
<!-- group coordinator -->
<suppress checks="CyclomaticComplexity"
+ files="(ConsumerGroupMember).java"/>
+ <suppress checks="ParameterNumber"
files="(ConsumerGroupMember|GroupMetadataManager).java"/>
<suppress checks="(NPathComplexity|MethodLength)"
files="(GroupMetadataManager|ConsumerGroupTest|GroupMetadataManagerTest).java"/>
<suppress checks="ClassFanOutComplexity"
files="(GroupMetadataManager|GroupMetadataManagerTest).java"/>
- <suppress checks="ParameterNumber"
- files="(ConsumerGroupMember|GroupMetadataManager).java"/>
Review Comment:
sorry I did remove it, when I double checked in the new PR I thought this removal was what I did, my bad I'll remove 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
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1276693242
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/TopicMetadata.java:
##########
@@ -40,23 +45,31 @@ public class TopicMetadata {
*/
private final int numPartitions;
+ /**
+ * Map of every partition to a set of its rackIds.
+ * If the rack information is unavailable, this is an empty map.
Review Comment:
changed the wording
--
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
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1275342001
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/TargetAssignmentBuilderTest.java:
##########
@@ -145,13 +147,13 @@ public TargetAssignmentBuilder.TargetAssignmentResult build() {
Map<String, AssignmentMemberSpec> memberSpecs = new HashMap<>();
// All the existing members are prepared.
- members.forEach((memberId, member) -> {
- memberSpecs.put(memberId, createAssignmentMemberSpec(
+ members.forEach((memberId, member) -> memberSpecs.put(memberId,
+ createAssignmentMemberSpec(
member,
targetAssignment.getOrDefault(memberId, Assignment.EMPTY),
subscriptionMetadata
- ));
- });
+ )
Review Comment:
I think this is more readable since the next lines are arguments for createAssignmentMemberSpec and the way above I have to look a lot from right to left to see what the arguments are for. I feel like this makes more sense 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
[GitHub] [kafka] dajac commented on pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on PR #14099:
URL: https://github.com/apache/kafka/pull/14099#issuecomment-1656057225
The build for JDK 17 is stuck... The three others look good. I have seen a few build stuck this week in other PRs so this is clearly not related to changes made in this one. Moreover, we had successful builds previously and we addressed nits in the mean time. Therefore, I will merge this PR.
--
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
[GitHub] [kafka] dajac merged pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac merged PR #14099:
URL: https://github.com/apache/kafka/pull/14099
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1276653838
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -5567,7 +5600,7 @@ public void testJoinGroupAppendErrorConversion() {
assertEquals(Errors.LEADER_NOT_AVAILABLE, Errors.LEADER_NOT_AVAILABLE);
}
- private <T> void assertUnorderedListEquals(
+ public static <T> void assertUnorderedListEquals(
Review Comment:
Is there a reason for making it public? I have the same question for the following two methods.
--
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
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1276706144
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -5692,12 +5725,60 @@ private void assertApiMessageAndVersionEquals(
fromTopicPartitions(actualValue.partitionsPendingRevocation()));
assertEquals(fromTopicPartitions(expectedValue.partitionsPendingAssignment()),
fromTopicPartitions(actualValue.partitionsPendingAssignment()));
- } else {
- assertEquals(expected.message(), actual.message());
+ } else if (actual.message() instanceof ConsumerGroupPartitionMetadataValue) {
+ // The order of the racks stored in PartitionMetadata of ConsumerGroupPartitionMetadataValue is not
+ // always guaranteed. Therefore, we need a special comparator.
+ ConsumerGroupPartitionMetadataValue expectedValue =
+ (ConsumerGroupPartitionMetadataValue) expected.message();
+ ConsumerGroupPartitionMetadataValue actualValue =
+ (ConsumerGroupPartitionMetadataValue) actual.message();
+
+ List<ConsumerGroupPartitionMetadataValue.TopicMetadata> expectedTopicMetadataList =
+ expectedValue.topics();
+ List<ConsumerGroupPartitionMetadataValue.TopicMetadata> actualTopicMetadataList =
+ actualValue.topics();
+
+ if (expectedTopicMetadataList.size() != actualTopicMetadataList.size()) {
+ fail("Topic metadata lists have different sizes");
+ }
+
+ for (int i = 0; i < expectedValue.topics().size(); i++) {
+ ConsumerGroupPartitionMetadataValue.TopicMetadata expectedTopicMetadata =
+ expectedTopicMetadataList.get(i);
+ ConsumerGroupPartitionMetadataValue.TopicMetadata actualTopicMetadata =
+ actualTopicMetadataList.get(i);
+
+ assertEquals(expectedTopicMetadata.topicId(), actualTopicMetadata.topicId());
+ assertEquals(expectedTopicMetadata.topicName(), actualTopicMetadata.topicName());
+ assertEquals(expectedTopicMetadata.numPartitions(), actualTopicMetadata.numPartitions());
+
+ List<ConsumerGroupPartitionMetadataValue.PartitionMetadata> expectedPartitionMetadataList =
+ expectedTopicMetadata.partitionMetadata();
+ List<ConsumerGroupPartitionMetadataValue.PartitionMetadata> actualPartitionMetadataList =
+ actualTopicMetadata.partitionMetadata();
+
+ // If the list is empty, rack information wasn't available for any replica of
+ // the partition and hence, the entry wasn't added to the record.
+ if (expectedPartitionMetadataList.size() != actualPartitionMetadataList.size()) {
+ fail("Partition metadata lists have different sizes");
+ } else if (!expectedPartitionMetadataList.isEmpty() && !actualPartitionMetadataList.isEmpty()) {
+ for (int j = 0; j < expectedTopicMetadata.numPartitions(); j++) {
+ ConsumerGroupPartitionMetadataValue.PartitionMetadata expectedPartitionMetadata =
+ expectedPartitionMetadataList.get(j);
+ ConsumerGroupPartitionMetadataValue.PartitionMetadata actualPartitionMetadata =
+ actualPartitionMetadataList.get(j);
+
+ assertEquals(expectedPartitionMetadata.partition(), actualPartitionMetadata.partition());
+ assertUnorderedListEquals(expectedPartitionMetadata.racks(), actualPartitionMetadata.racks());
+ }
+ } else {
+ assertEquals(expected.message(), actual.message());
Review Comment:
Sorry that was indented wrong, this is the else for the outermost if and else if where the message type is checked
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1276742165
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -5567,7 +5600,7 @@ public void testJoinGroupAppendErrorConversion() {
assertEquals(Errors.LEADER_NOT_AVAILABLE, Errors.LEADER_NOT_AVAILABLE);
}
- private <T> void assertUnorderedListEquals(
+ public static <T> void assertUnorderedListEquals(
Review Comment:
I see. I wonder if we should actually move `recordEquals` to `RecordHelpersTest`. The dependency would make more sense this way, no?
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1276670928
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/SubscribedTopicMetadataTest.java:
##########
@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.consumer;
+
+import org.apache.kafka.common.Uuid;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+import static org.apache.kafka.coordinator.group.RecordHelpersTest.mkMapOfPartitionRacks;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class SubscribedTopicMetadataTest {
+
+ @Test
+ public void testAttribute() {
+ Map<Uuid, TopicMetadata> topicMetadataMap = new HashMap<>();
+ for (int i = 0; i < 5; i++) {
+ Uuid topicId = Uuid.randomUuid();
+ String topicName = "topic" + i;
+ Map<Integer, Set<String>> partitionRacks = mkMapOfPartitionRacks(5);
+ topicMetadataMap.put(
+ topicId,
+ new TopicMetadata(topicId, topicName, 5, partitionRacks)
+ );
+ }
+ assertEquals(topicMetadataMap, new SubscribedTopicMetadata(topicMetadataMap).topicMetadata());
+ }
+
+ @Test
+ public void testTopicMetadataCannotBeNull() {
+ assertThrows(NullPointerException.class, () -> new SubscribedTopicMetadata(null));
+ }
+
+ @Test
+ public void testEquals() {
+ Map<Uuid, TopicMetadata> topicMetadataMap = new HashMap<>();
+ for (int i = 0; i < 5; i++) {
+ Uuid topicId = Uuid.randomUuid();
+ String topicName = "topic" + i;
+ Map<Integer, Set<String>> partitionRacks = mkMapOfPartitionRacks(5);
+ topicMetadataMap.put(
+ topicId,
+ new TopicMetadata(topicId, topicName, 5, partitionRacks)
+ );
+ }
+ SubscribedTopicMetadata subscribedTopicMetadata = new SubscribedTopicMetadata(topicMetadataMap);
+ assertEquals(new SubscribedTopicMetadata(topicMetadataMap), subscribedTopicMetadata);
+
+ Map<Uuid, TopicMetadata> topicMetadataMap2 = new HashMap<>();
+ Uuid topicId = Uuid.randomUuid();
+ topicMetadataMap2.put(topicId, new TopicMetadata(topicId, "newTopic", 5, Collections.emptyMap()));
+ assertNotEquals(new SubscribedTopicMetadata(topicMetadataMap2), subscribedTopicMetadata);
+ }
+}
Review Comment:
nit: Let's add a new line at the end.
--
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
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1276830285
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -5567,7 +5600,7 @@ public void testJoinGroupAppendErrorConversion() {
assertEquals(Errors.LEADER_NOT_AVAILABLE, Errors.LEADER_NOT_AVAILABLE);
}
- private <T> void assertUnorderedListEquals(
+ public static <T> void assertUnorderedListEquals(
Review Comment:
Yeah that makes sense, I'll have to move all the corresponding methods also to that class
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1275243699
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/RecordHelpers.java:
##########
@@ -131,13 +131,24 @@ public static Record newGroupSubscriptionMetadataRecord(
Map<String, TopicMetadata> newSubscriptionMetadata
) {
ConsumerGroupPartitionMetadataValue value = new ConsumerGroupPartitionMetadataValue();
- newSubscriptionMetadata.forEach((topicName, topicMetadata) ->
+ newSubscriptionMetadata.forEach((topicName, topicMetadata) -> {
+ List<ConsumerGroupPartitionMetadataValue.PartitionMetadata> partitionMetadata = new ArrayList<>();
+ if (!topicMetadata.partitionRacks().isEmpty()) {
+ topicMetadata.partitionRacks().forEach((partition, racks) -> {
+ partitionMetadata.add(new ConsumerGroupPartitionMetadataValue.PartitionMetadata()
+ .setPartition(partition)
+ .setRacks(new ArrayList<>(racks))
+ );
+ });
+ }
+ // If partition rack information is empty, store an empty list in the record.
value.topics().add(new ConsumerGroupPartitionMetadataValue.TopicMetadata()
.setTopicId(topicMetadata.id())
.setTopicName(topicMetadata.name())
.setNumPartitions(topicMetadata.numPartitions())
- )
- );
+ .setPartitionMetadata(partitionMetadata.isEmpty() ? Collections.emptyList() : partitionMetadata)
Review Comment:
Using `Collections.emptyList` makes sense when you can avoid allocating an empty `ArrayList`. In this case, the array is already allocated so we could just use 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
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1275337959
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroupTest.java:
##########
@@ -417,19 +417,23 @@ public void testUpdateSubscriptionMetadata() {
consumerGroup.computeSubscriptionMetadata(
null,
null,
- image.topics()
+ image.topics(),
+ image.cluster()
)
);
// Compute while taking into account member 1.
assertEquals(
mkMap(
- mkEntry("foo", new TopicMetadata(fooTopicId, "foo", 1))
+ mkEntry("foo",
+ new TopicMetadata(fooTopicId, "foo", 1, Collections.emptyMap())
+ )
Review Comment:
Made it one line
--
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
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1275444245
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/TargetAssignmentBuilderTest.java:
##########
@@ -624,8 +745,27 @@ public void testDeleteMember() {
20
);
- Uuid fooTopicId = context.addTopicMetadata("foo", 6);
- Uuid barTopicId = context.addTopicMetadata("bar", 6);
+ Uuid fooTopicId = context.addTopicMetadata("foo", 6,
+ mkMap(
+ mkEntry(0, Collections.emptySet()),
+ mkEntry(1, Collections.emptySet()),
+ mkEntry(2, Collections.emptySet()),
+ mkEntry(3, Collections.emptySet()),
+ mkEntry(4, Collections.emptySet()),
+ mkEntry(5, Collections.emptySet())
+ )
+ );
+
+ Uuid barTopicId = context.addTopicMetadata("bar", 6,
Review Comment:
https://github.com/apache/kafka/pull/14099/files/70a6214809fba59511dab58a1da21f74be4edbae#r1275362470
--
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
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1276487843
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/TargetAssignmentBuilderTest.java:
##########
@@ -624,8 +745,27 @@ public void testDeleteMember() {
20
);
- Uuid fooTopicId = context.addTopicMetadata("foo", 6);
- Uuid barTopicId = context.addTopicMetadata("bar", 6);
+ Uuid fooTopicId = context.addTopicMetadata("foo", 6,
+ mkMap(
+ mkEntry(0, Collections.emptySet()),
+ mkEntry(1, Collections.emptySet()),
+ mkEntry(2, Collections.emptySet()),
+ mkEntry(3, Collections.emptySet()),
+ mkEntry(4, Collections.emptySet()),
+ mkEntry(5, Collections.emptySet())
+ )
+ );
+
+ Uuid barTopicId = context.addTopicMetadata("bar", 6,
Review Comment:
Added racks to one of the tests `testPartialAssignmentUpdate`. and removed for the rest
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1276663083
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -5692,12 +5725,60 @@ private void assertApiMessageAndVersionEquals(
fromTopicPartitions(actualValue.partitionsPendingRevocation()));
assertEquals(fromTopicPartitions(expectedValue.partitionsPendingAssignment()),
fromTopicPartitions(actualValue.partitionsPendingAssignment()));
- } else {
- assertEquals(expected.message(), actual.message());
+ } else if (actual.message() instanceof ConsumerGroupPartitionMetadataValue) {
+ // The order of the racks stored in PartitionMetadata of ConsumerGroupPartitionMetadataValue is not
+ // always guaranteed. Therefore, we need a special comparator.
+ ConsumerGroupPartitionMetadataValue expectedValue =
+ (ConsumerGroupPartitionMetadataValue) expected.message();
+ ConsumerGroupPartitionMetadataValue actualValue =
+ (ConsumerGroupPartitionMetadataValue) actual.message();
+
+ List<ConsumerGroupPartitionMetadataValue.TopicMetadata> expectedTopicMetadataList =
+ expectedValue.topics();
+ List<ConsumerGroupPartitionMetadataValue.TopicMetadata> actualTopicMetadataList =
+ actualValue.topics();
+
+ if (expectedTopicMetadataList.size() != actualTopicMetadataList.size()) {
+ fail("Topic metadata lists have different sizes");
+ }
+
+ for (int i = 0; i < expectedValue.topics().size(); i++) {
+ ConsumerGroupPartitionMetadataValue.TopicMetadata expectedTopicMetadata =
+ expectedTopicMetadataList.get(i);
+ ConsumerGroupPartitionMetadataValue.TopicMetadata actualTopicMetadata =
+ actualTopicMetadataList.get(i);
+
+ assertEquals(expectedTopicMetadata.topicId(), actualTopicMetadata.topicId());
+ assertEquals(expectedTopicMetadata.topicName(), actualTopicMetadata.topicName());
+ assertEquals(expectedTopicMetadata.numPartitions(), actualTopicMetadata.numPartitions());
+
+ List<ConsumerGroupPartitionMetadataValue.PartitionMetadata> expectedPartitionMetadataList =
+ expectedTopicMetadata.partitionMetadata();
+ List<ConsumerGroupPartitionMetadataValue.PartitionMetadata> actualPartitionMetadataList =
+ actualTopicMetadata.partitionMetadata();
+
+ // If the list is empty, rack information wasn't available for any replica of
+ // the partition and hence, the entry wasn't added to the record.
+ if (expectedPartitionMetadataList.size() != actualPartitionMetadataList.size()) {
+ fail("Partition metadata lists have different sizes");
+ } else if (!expectedPartitionMetadataList.isEmpty() && !actualPartitionMetadataList.isEmpty()) {
+ for (int j = 0; j < expectedTopicMetadata.numPartitions(); j++) {
Review Comment:
Using numPartitions seems incorrect here as it could be larger than the number of partition metadata.
--
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
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1276830477
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/RecordHelpersTest.java:
##########
@@ -764,4 +813,39 @@ public void testNewOffsetCommitTombstoneRecord() {
Record record = RecordHelpers.newOffsetCommitTombstoneRecord("group-id", "foo", 1);
assertEquals(expectedRecord, record);
}
+
+ // For the purpose of testing let:
+ // a) Number of replicas for each partition = 2
+ // b) Number of racks available in the cluster = 4
Review Comment:
Okay I will add it to the javadoc
--
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
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1275444245
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/TargetAssignmentBuilderTest.java:
##########
@@ -624,8 +745,27 @@ public void testDeleteMember() {
20
);
- Uuid fooTopicId = context.addTopicMetadata("foo", 6);
- Uuid barTopicId = context.addTopicMetadata("bar", 6);
+ Uuid fooTopicId = context.addTopicMetadata("foo", 6,
+ mkMap(
+ mkEntry(0, Collections.emptySet()),
+ mkEntry(1, Collections.emptySet()),
+ mkEntry(2, Collections.emptySet()),
+ mkEntry(3, Collections.emptySet()),
+ mkEntry(4, Collections.emptySet()),
+ mkEntry(5, Collections.emptySet())
+ )
+ );
+
+ Uuid barTopicId = context.addTopicMetadata("bar", 6,
Review Comment:
https://github.com/apache/kafka/pull/14099/files/70a6214809fba59511dab58a1da21f74be4edbae#r1275362470
--
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
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1276573086
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/SubscribedTopicMetadata.java:
##########
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.consumer;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.coordinator.group.assignor.PartitionAssignor;
+import org.apache.kafka.coordinator.group.assignor.SubscribedTopicDescriber;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * The subscribed topic metadata class is used by the {@link PartitionAssignor} to obtain
+ * topic and partition metadata for the topics that the consumer group is subscribed to.
+ */
+public class SubscribedTopicMetadata implements SubscribedTopicDescriber {
Review Comment:
done!
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1276669652
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/RecordHelpersTest.java:
##########
@@ -764,4 +813,39 @@ public void testNewOffsetCommitTombstoneRecord() {
Record record = RecordHelpers.newOffsetCommitTombstoneRecord("group-id", "foo", 1);
assertEquals(expectedRecord, record);
}
+
+ // For the purpose of testing let:
+ // a) Number of replicas for each partition = 2
+ // b) Number of racks available in the cluster = 4
Review Comment:
Should we remove this? It does not make sense here. Or was it supposed to be somewhere else?
--
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
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1275346208
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/TargetAssignmentBuilderTest.java:
##########
@@ -145,13 +147,13 @@ public TargetAssignmentBuilder.TargetAssignmentResult build() {
Map<String, AssignmentMemberSpec> memberSpecs = new HashMap<>();
// All the existing members are prepared.
- members.forEach((memberId, member) -> {
- memberSpecs.put(memberId, createAssignmentMemberSpec(
+ members.forEach((memberId, member) -> memberSpecs.put(memberId,
+ createAssignmentMemberSpec(
member,
targetAssignment.getOrDefault(memberId, Assignment.EMPTY),
subscriptionMetadata
- ));
- });
+ )
Review Comment:
moved memberSpec to the next line
--
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
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1275342001
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/TargetAssignmentBuilderTest.java:
##########
@@ -145,13 +147,13 @@ public TargetAssignmentBuilder.TargetAssignmentResult build() {
Map<String, AssignmentMemberSpec> memberSpecs = new HashMap<>();
// All the existing members are prepared.
- members.forEach((memberId, member) -> {
- memberSpecs.put(memberId, createAssignmentMemberSpec(
+ members.forEach((memberId, member) -> memberSpecs.put(memberId,
+ createAssignmentMemberSpec(
member,
targetAssignment.getOrDefault(memberId, Assignment.EMPTY),
subscriptionMetadata
- ));
- });
+ )
Review Comment:
I think this is more readable since the next lines are arguments for createAssignmentMemberSpec and the way above I have to look a lot from right to left to see what the arguments are for. I feel like this makes more sense 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
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1275362470
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/TargetAssignmentBuilderTest.java:
##########
@@ -222,8 +231,25 @@ public void testCreateAssignmentMemberSpec() {
Map<String, TopicMetadata> subscriptionMetadata = new HashMap<String, TopicMetadata>() {
{
- put("foo", new TopicMetadata(fooTopicId, "foo", 5));
- put("bar", new TopicMetadata(barTopicId, "bar", 5));
+ put("foo", new TopicMetadata(fooTopicId, "foo", 5,
+ mkMap(
+ mkEntry(0, Collections.emptySet()),
+ mkEntry(1, Collections.emptySet()),
+ mkEntry(2, Collections.emptySet()),
+ mkEntry(3, Collections.emptySet()),
+ mkEntry(4, Collections.emptySet())
+ )
Review Comment:
I kept it just in case we wanted to add the rack information
--
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
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1275362470
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/TargetAssignmentBuilderTest.java:
##########
@@ -222,8 +231,25 @@ public void testCreateAssignmentMemberSpec() {
Map<String, TopicMetadata> subscriptionMetadata = new HashMap<String, TopicMetadata>() {
{
- put("foo", new TopicMetadata(fooTopicId, "foo", 5));
- put("bar", new TopicMetadata(barTopicId, "bar", 5));
+ put("foo", new TopicMetadata(fooTopicId, "foo", 5,
+ mkMap(
+ mkEntry(0, Collections.emptySet()),
+ mkEntry(1, Collections.emptySet()),
+ mkEntry(2, Collections.emptySet()),
+ mkEntry(3, Collections.emptySet()),
+ mkEntry(4, Collections.emptySet())
+ )
Review Comment:
I kept it just in case we wanted to add the rack information, also wanted some tests with a map with information rather than just an empty map but since these tests aren't related to the rack info I'll just pass the empty map
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1274915017
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/RecordHelpersTest.java:
##########
@@ -756,4 +759,24 @@ public void testNewOffsetCommitTombstoneRecord() {
Record record = RecordHelpers.newOffsetCommitTombstoneRecord("group-id", "foo", 1);
assertEquals(expectedRecord, record);
}
+
Review Comment:
Should we add a test which verify that partitions with no racks are not put into the record?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/RecordHelpers.java:
##########
@@ -131,13 +131,24 @@ public static Record newGroupSubscriptionMetadataRecord(
Map<String, TopicMetadata> newSubscriptionMetadata
) {
ConsumerGroupPartitionMetadataValue value = new ConsumerGroupPartitionMetadataValue();
- newSubscriptionMetadata.forEach((topicName, topicMetadata) ->
+ newSubscriptionMetadata.forEach((topicName, topicMetadata) -> {
+ List<ConsumerGroupPartitionMetadataValue.PartitionMetadata> partitionMetadata = new ArrayList<>();
+ if (!topicMetadata.partitionRacks().isEmpty()) {
+ topicMetadata.partitionRacks().forEach((partition, racks) -> {
+ partitionMetadata.add(new ConsumerGroupPartitionMetadataValue.PartitionMetadata()
+ .setPartition(partition)
+ .setRacks(new ArrayList<>(racks))
+ );
+ });
+ }
+ // If partition rack information is empty, store an empty list in the record.
Review Comment:
nit: Would it make sense to put this comment before the `if` statement?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/RecordHelpersTest.java:
##########
@@ -612,15 +616,14 @@ public void testNewGroupMetadataRecordThrowsWhenEmptyAssignment() {
MetadataVersion.IBP_3_5_IV2
));
}
-
Review Comment:
nit: Let's revert this.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/RecordHelpers.java:
##########
@@ -131,13 +131,24 @@ public static Record newGroupSubscriptionMetadataRecord(
Map<String, TopicMetadata> newSubscriptionMetadata
) {
ConsumerGroupPartitionMetadataValue value = new ConsumerGroupPartitionMetadataValue();
- newSubscriptionMetadata.forEach((topicName, topicMetadata) ->
+ newSubscriptionMetadata.forEach((topicName, topicMetadata) -> {
+ List<ConsumerGroupPartitionMetadataValue.PartitionMetadata> partitionMetadata = new ArrayList<>();
+ if (!topicMetadata.partitionRacks().isEmpty()) {
+ topicMetadata.partitionRacks().forEach((partition, racks) -> {
+ partitionMetadata.add(new ConsumerGroupPartitionMetadataValue.PartitionMetadata()
+ .setPartition(partition)
+ .setRacks(new ArrayList<>(racks))
+ );
+ });
+ }
+ // If partition rack information is empty, store an empty list in the record.
value.topics().add(new ConsumerGroupPartitionMetadataValue.TopicMetadata()
.setTopicId(topicMetadata.id())
.setTopicName(topicMetadata.name())
.setNumPartitions(topicMetadata.numPartitions())
- )
- );
+ .setPartitionMetadata(partitionMetadata.isEmpty() ? Collections.emptyList() : partitionMetadata)
Review Comment:
nit: Why can't we use `partitionMetadata` if it is empty? That seems just fine.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/PartitionAssignor.java:
##########
@@ -26,18 +26,17 @@
*/
@InterfaceStability.Unstable
public interface PartitionAssignor {
-
/**
* Unique name for this assignor.
*/
String name();
/**
- * Perform the group assignment given the current members and
- * topic metadata.
+ * Assigns partitions to group members based on the given assignment specification and topic metadata.
*
- * @param assignmentSpec The assignment spec.
+ * @param assignmentSpec The assignment spec which included member metadata.
Review Comment:
nit: `includes`?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/SubscribedTopicMetadata.java:
##########
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.consumer;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.coordinator.group.assignor.PartitionAssignor;
+import org.apache.kafka.coordinator.group.assignor.SubscribedTopicDescriber;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * The subscribed topic metadata class is used by the {@link PartitionAssignor} to obtain
+ * topic and partition metadata for the topics that the consumer group is subscribed to.
+ */
+public class SubscribedTopicMetadata implements SubscribedTopicDescriber {
+ /**
+ * The topic IDs mapped to their corresponding {@link TopicMetadata}
+ * object, which contains topic and partition metadata.
+ */
+ Map<Uuid, TopicMetadata> topicMetadata;
+
+ public SubscribedTopicMetadata(Map<Uuid, TopicMetadata> topicMetadata) {
+ this.topicMetadata = topicMetadata;
+ }
+
+ /**
+ * The number of partitions for the given topic ID.
+ *
+ * @param topicId Uuid corresponding to the topic.
+ * @return The number of partitions corresponding to the given topic ID,
+ * or -1 if the topic ID does not exist.
+ */
+ @Override
+ public int numPartitions(Uuid topicId) {
+ return this.topicMetadata.containsKey(topicId) ? this.topicMetadata.get(topicId).numPartitions() : -1;
Review Comment:
nit: It is usually better to write this as follow:
```
TopicMetadata topic = this.topicMetadata.get(topicId);
return topic == null ? -1 : topic.numPartitions();
```
This ways you do only one lookup.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/SubscribedTopicDescriber.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.assignor;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.annotation.InterfaceStability;
+
+import java.util.Set;
+
+/**
+ * The subscribed topic describer is used by the {@link PartitionAssignor}
+ * to obtain topic and partition metadata of the subscribed topics.
+ *
+ * The interface is kept in an internal module until KIP-848 is fully
+ * implemented and ready to be released.
+ */
+@InterfaceStability.Unstable
+public interface SubscribedTopicDescriber {
+ /**
+ * The number of partitions for the given topic ID.
Review Comment:
nit: In AssignmentSpec, you used `member Id`. Should we use `Id` in this file as well?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/TopicMetadata.java:
##########
@@ -40,23 +45,31 @@ public class TopicMetadata {
*/
private final int numPartitions;
+ /**
+ * Map of every partition to a set of its rackIds.
Review Comment:
nit: `rack Ids`?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/TopicMetadata.java:
##########
@@ -40,23 +45,31 @@ public class TopicMetadata {
*/
private final int numPartitions;
+ /**
+ * Map of every partition to a set of its rackIds.
+ * If the rack information is unavailable, this is an empty map.
Review Comment:
nit: Should we say that the map contains mapping for partitions only of the partitions has racks or something like this?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/SubscribedTopicMetadata.java:
##########
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.consumer;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.coordinator.group.assignor.PartitionAssignor;
+import org.apache.kafka.coordinator.group.assignor.SubscribedTopicDescriber;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * The subscribed topic metadata class is used by the {@link PartitionAssignor} to obtain
+ * topic and partition metadata for the topics that the consumer group is subscribed to.
+ */
+public class SubscribedTopicMetadata implements SubscribedTopicDescriber {
+ /**
+ * The topic IDs mapped to their corresponding {@link TopicMetadata}
+ * object, which contains topic and partition metadata.
+ */
+ Map<Uuid, TopicMetadata> topicMetadata;
+
+ public SubscribedTopicMetadata(Map<Uuid, TopicMetadata> topicMetadata) {
+ this.topicMetadata = topicMetadata;
+ }
+
+ /**
+ * The number of partitions for the given topic ID.
+ *
+ * @param topicId Uuid corresponding to the topic.
+ * @return The number of partitions corresponding to the given topic ID,
+ * or -1 if the topic ID does not exist.
+ */
+ @Override
+ public int numPartitions(Uuid topicId) {
+ return this.topicMetadata.containsKey(topicId) ? this.topicMetadata.get(topicId).numPartitions() : -1;
+ }
+
+ /**
+ * Returns all the available racks associated with the replicas of the given partition.
+ *
+ * @param topicId Uuid corresponding to the partition's topic.
+ * @param partition Partition number within topic.
Review Comment:
nit: `Partition Id` or `Partition Index`.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/SubscribedTopicMetadata.java:
##########
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.consumer;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.coordinator.group.assignor.PartitionAssignor;
+import org.apache.kafka.coordinator.group.assignor.SubscribedTopicDescriber;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * The subscribed topic metadata class is used by the {@link PartitionAssignor} to obtain
+ * topic and partition metadata for the topics that the consumer group is subscribed to.
+ */
+public class SubscribedTopicMetadata implements SubscribedTopicDescriber {
+ /**
+ * The topic IDs mapped to their corresponding {@link TopicMetadata}
+ * object, which contains topic and partition metadata.
+ */
+ Map<Uuid, TopicMetadata> topicMetadata;
+
+ public SubscribedTopicMetadata(Map<Uuid, TopicMetadata> topicMetadata) {
+ this.topicMetadata = topicMetadata;
+ }
+
+ /**
+ * The number of partitions for the given topic ID.
+ *
+ * @param topicId Uuid corresponding to the topic.
+ * @return The number of partitions corresponding to the given topic ID,
+ * or -1 if the topic ID does not exist.
+ */
+ @Override
+ public int numPartitions(Uuid topicId) {
+ return this.topicMetadata.containsKey(topicId) ? this.topicMetadata.get(topicId).numPartitions() : -1;
+ }
+
+ /**
+ * Returns all the available racks associated with the replicas of the given partition.
+ *
+ * @param topicId Uuid corresponding to the partition's topic.
+ * @param partition Partition number within topic.
+ * @return The set of racks corresponding to the replicas of the topics partition.
+ * If the topic ID does not exist, an empty set is returned
+ */
+ @Override
+ public Set<String> racksForPartition(Uuid topicId, int partition) {
+ return this.topicMetadata.containsKey(topicId) ?
+ this.topicMetadata.get(topicId).partitionRacks().get(partition) :
+ Collections.emptySet();
Review Comment:
nit: Same comment as before.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/TopicMetadata.java:
##########
@@ -40,23 +45,31 @@ public class TopicMetadata {
*/
private final int numPartitions;
+ /**
+ * Map of every partition to a set of its rackIds.
+ * If the rack information is unavailable, this is an empty map.
+ */
+ private final Map<Integer, Set<String>> partitionRacks;
+
public TopicMetadata(
Uuid id,
String name,
- int numPartitions
+ int numPartitions,
+ Map<Integer, Set<String>> partitionRacks
) {
- this.id = Objects.requireNonNull(id);
- if (Uuid.ZERO_UUID.equals(id)) {
- throw new IllegalArgumentException("Topic id cannot be ZERO_UUID.");
- }
- this.name = Objects.requireNonNull(name);
- if (name.isEmpty()) {
- throw new IllegalArgumentException("Topic name cannot be empty.");
- }
- this.numPartitions = numPartitions;
- if (numPartitions < 0) {
- throw new IllegalArgumentException("Number of partitions cannot be negative.");
- }
+ this.id = Objects.requireNonNull(id);
+ this.partitionRacks = Objects.requireNonNull(partitionRacks);
Review Comment:
nit: Could we put this one last in the constructor? The line before it and after it goes together.
##########
group-coordinator/src/main/resources/common/message/ConsumerGroupPartitionMetadataValue.json:
##########
@@ -29,7 +29,14 @@
{ "name": "TopicName", "versions": "0+", "type": "string",
"about": "The topic name." },
{ "name": "NumPartitions", "versions": "0+", "type": "int32",
- "about": "The number of partitions of the topic." }
+ "about": "The number of partitions of the topic." },
+ { "name": "PartitionMetadata", "versions": "0+", "type": "[]PartitionMetadata",
+ "about": "Partitions mapped to a set of racks.", "fields": [
Review Comment:
nit: Could we update the doc here to explain that we only store partition metadata if racks is non-empty?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/TopicMetadata.java:
##########
@@ -106,16 +129,26 @@ public String toString() {
"id=" + id +
", name=" + name +
", numPartitions=" + numPartitions +
+ ", partitionRacks=" + partitionRacks +
')';
}
public static TopicMetadata fromRecord(
ConsumerGroupPartitionMetadataValue.TopicMetadata record
) {
+ // Converting the data type from a list stored in the record to a map.
+ Map<Integer, Set<String>> partitionRacks = new HashMap<>();
+ for (ConsumerGroupPartitionMetadataValue.PartitionMetadata partitionMetadata : record.partitionMetadata()) {
+ partitionRacks.put(
+ partitionMetadata.partition(),
+ Collections.unmodifiableSet(new HashSet<>(partitionMetadata.racks()))
+ );
+ }
+
return new TopicMetadata(
record.topicId(),
record.topicName(),
- record.numPartitions()
- );
+ record.numPartitions(),
+ partitionRacks.isEmpty() ? Collections.emptyMap() : partitionRacks);
Review Comment:
nit: You can use `partitionRacks` in all cases here.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/TargetAssignmentBuilderTest.java:
##########
@@ -145,13 +147,13 @@ public TargetAssignmentBuilder.TargetAssignmentResult build() {
Map<String, AssignmentMemberSpec> memberSpecs = new HashMap<>();
// All the existing members are prepared.
- members.forEach((memberId, member) -> {
- memberSpecs.put(memberId, createAssignmentMemberSpec(
+ members.forEach((memberId, member) -> memberSpecs.put(memberId,
+ createAssignmentMemberSpec(
member,
targetAssignment.getOrDefault(memberId, Assignment.EMPTY),
subscriptionMetadata
- ));
- });
+ )
Review Comment:
nit: It would be better like this if we really want to bing `memberSpecs.put` on the previous line. Otherwise, we can just keep the old format.
```
members.forEach((memberId, member) -> memberSpecs.put(memberId, createAssignmentMemberSpec(
member,
targetAssignment.getOrDefault(memberId, Assignment.EMPTY),
subscriptionMetadata
));
```
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroupTest.java:
##########
@@ -417,19 +417,23 @@ public void testUpdateSubscriptionMetadata() {
consumerGroup.computeSubscriptionMetadata(
null,
null,
- image.topics()
+ image.topics(),
+ image.cluster()
)
);
// Compute while taking into account member 1.
assertEquals(
mkMap(
- mkEntry("foo", new TopicMetadata(fooTopicId, "foo", 1))
+ mkEntry("foo",
+ new TopicMetadata(fooTopicId, "foo", 1, Collections.emptyMap())
+ )
Review Comment:
nit: This format is weird.
It should be as follow:
```
mkEntry(
"foo",
new TopicMetadata(fooTopicId, "foo", 1, Collections.emptyMap())
)
```
or kept on one line. One line seems fine here.
There are many other cases in this file. I won't mention them.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/TargetAssignmentBuilderTest.java:
##########
@@ -168,20 +170,24 @@ public TargetAssignmentBuilder.TargetAssignmentResult build() {
});
// Prepare the expected topic metadata.
- Map<Uuid, AssignmentTopicMetadata> topicMetadata = new HashMap<>();
- subscriptionMetadata.forEach((topicName, metadata) -> {
- topicMetadata.put(metadata.id(), new AssignmentTopicMetadata(metadata.numPartitions()));
- });
+ Map<Uuid, TopicMetadata> topicMetadataMap = new HashMap<>();
+ subscriptionMetadata.forEach((topicName, topicMetadata) ->
+ topicMetadataMap.put(topicMetadata.id(), topicMetadata));
// Prepare the expected assignment spec.
AssignmentSpec assignmentSpec = new AssignmentSpec(
- memberSpecs,
- topicMetadata
+ memberSpecs
+ );
+
+ SubscribedTopicMetadata assignmentTopicMetadata = new SubscribedTopicMetadata(
+ topicMetadataMap
);
Review Comment:
nit: Could you put this right after the `topicMetadataMap` preparation. They seems to go together.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/TargetAssignmentBuilderTest.java:
##########
@@ -222,8 +231,25 @@ public void testCreateAssignmentMemberSpec() {
Map<String, TopicMetadata> subscriptionMetadata = new HashMap<String, TopicMetadata>() {
{
- put("foo", new TopicMetadata(fooTopicId, "foo", 5));
- put("bar", new TopicMetadata(barTopicId, "bar", 5));
+ put("foo", new TopicMetadata(fooTopicId, "foo", 5,
+ mkMap(
+ mkEntry(0, Collections.emptySet()),
+ mkEntry(1, Collections.emptySet()),
+ mkEntry(2, Collections.emptySet()),
+ mkEntry(3, Collections.emptySet()),
+ mkEntry(4, Collections.emptySet())
+ )
Review Comment:
As the racks are empty, we don't have to define the partitions, right? Could we just used `Collections.emptyMap()`? There are other cases in this file.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/TargetAssignmentBuilderTest.java:
##########
@@ -203,7 +209,10 @@ public TargetAssignmentBuilder.TargetAssignmentResult build() {
// Verify that the assignor was called once with the expected
// assignment spec.
- verify(assignor, times(1)).assign(assignmentSpec);
+ verify(assignor, times(1))
+ .assign(
+ assignmentSpec, assignmentTopicMetadata
+ );
Review Comment:
nit: This code format is weird. If you really want to break the line, I would format it as follow:
```
verify(assignor, times(1)).assign(
assignmentSpec,
assignmentTopicMetadata
);
```
or as follow (like you did at L189)
```
verify(assignor, times(1))
.assign(assignmentSpec, assignmentTopicMetadata);
```
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/TargetAssignmentBuilderTest.java:
##########
@@ -624,8 +745,27 @@ public void testDeleteMember() {
20
);
- Uuid fooTopicId = context.addTopicMetadata("foo", 6);
- Uuid barTopicId = context.addTopicMetadata("bar", 6);
+ Uuid fooTopicId = context.addTopicMetadata("foo", 6,
+ mkMap(
+ mkEntry(0, Collections.emptySet()),
+ mkEntry(1, Collections.emptySet()),
+ mkEntry(2, Collections.emptySet()),
+ mkEntry(3, Collections.emptySet()),
+ mkEntry(4, Collections.emptySet()),
+ mkEntry(5, Collections.emptySet())
+ )
+ );
+
+ Uuid barTopicId = context.addTopicMetadata("bar", 6,
Review Comment:
Should we have at least one test where we set racks for some partitions?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/SubscribedTopicMetadata.java:
##########
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.consumer;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.coordinator.group.assignor.PartitionAssignor;
+import org.apache.kafka.coordinator.group.assignor.SubscribedTopicDescriber;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * The subscribed topic metadata class is used by the {@link PartitionAssignor} to obtain
+ * topic and partition metadata for the topics that the consumer group is subscribed to.
+ */
+public class SubscribedTopicMetadata implements SubscribedTopicDescriber {
Review Comment:
Should we add basic unit tests for this class?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/TopicMetadataTest.java:
##########
@@ -28,24 +30,24 @@ public class TopicMetadataTest {
@Test
public void testAttributes() {
Uuid topicId = Uuid.randomUuid();
- TopicMetadata topicMetadata = new TopicMetadata(topicId, "foo", 15);
+ TopicMetadata topicMetadata = new TopicMetadata(topicId, "foo", 15, Collections.emptyMap());
Review Comment:
Should we set a map and verify that we get the correct one back?
##########
reviewers.py:
##########
@@ -28,7 +28,7 @@
def prompt_for_user():
while True:
try:
- user_input = input("\nName or email (case insensitive): ")
+ user_input = input("\nName or email (case insensitive): ")
Review Comment:
This file is still here :)
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/PartitionAssignor.java:
##########
@@ -26,18 +26,17 @@
*/
@InterfaceStability.Unstable
public interface PartitionAssignor {
-
/**
* Unique name for this assignor.
*/
String name();
/**
- * Perform the group assignment given the current members and
- * topic metadata.
+ * Assigns partitions to group members based on the given assignment specification and topic metadata.
*
- * @param assignmentSpec The assignment spec.
+ * @param assignmentSpec The assignment spec which included member metadata.
+ * @param subscribedTopicDescriber The topic and partition metadata describer {@link SubscribedTopicDescriber}.
Review Comment:
nit: I think that we don't have to put `{@link SubscribedTopicDescriber}` here as the javadoc will automatically link the type of `subscribedTopicDescriber`.
--
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
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1275809515
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/RecordHelpersTest.java:
##########
@@ -756,4 +759,24 @@ public void testNewOffsetCommitTombstoneRecord() {
Record record = RecordHelpers.newOffsetCommitTombstoneRecord("group-id", "foo", 1);
assertEquals(expectedRecord, record);
}
+
Review Comment:
yes adding 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
[GitHub] [kafka] dajac commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1276668936
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/RecordHelpersTest.java:
##########
@@ -612,15 +616,14 @@ public void testNewGroupMetadataRecordThrowsWhenEmptyAssignment() {
MetadataVersion.IBP_3_5_IV2
));
}
-
Review Comment:
It is not really about the empty line bit more about the unnecessary spaces on 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
[GitHub] [kafka] dajac commented on pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on PR #14099:
URL: https://github.com/apache/kafka/pull/14099#issuecomment-1651739655
@rreddy-22 The build does not look good. Could you check 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
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1275224303
##########
reviewers.py:
##########
@@ -28,7 +28,7 @@
def prompt_for_user():
while True:
try:
- user_input = input("\nName or email (case insensitive): ")
+ user_input = input("\nName or email (case insensitive): ")
Review Comment:
Ugh I removed it twice Idk why it keeps coming back, I promise I literally reverted the changes twice
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1275242714
##########
checkstyle/suppressions.xml:
##########
@@ -321,13 +321,13 @@
<!-- group coordinator -->
<suppress checks="CyclomaticComplexity"
+ files="(ConsumerGroupMember).java"/>
+ <suppress checks="ParameterNumber"
files="(ConsumerGroupMember|GroupMetadataManager).java"/>
<suppress checks="(NPathComplexity|MethodLength)"
files="(GroupMetadataManager|ConsumerGroupTest|GroupMetadataManagerTest).java"/>
<suppress checks="ClassFanOutComplexity"
files="(GroupMetadataManager|GroupMetadataManagerTest).java"/>
- <suppress checks="ParameterNumber"
- files="(ConsumerGroupMember|GroupMetadataManager).java"/>
Review Comment:
Hum.. I asked in the other PR if the change was necessary: https://github.com/apache/kafka/pull/13998/files#r1272466045. You said that you removed it but it is still here. I am confused.
--
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
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1275307169
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/RecordHelpers.java:
##########
@@ -131,13 +131,24 @@ public static Record newGroupSubscriptionMetadataRecord(
Map<String, TopicMetadata> newSubscriptionMetadata
) {
ConsumerGroupPartitionMetadataValue value = new ConsumerGroupPartitionMetadataValue();
- newSubscriptionMetadata.forEach((topicName, topicMetadata) ->
+ newSubscriptionMetadata.forEach((topicName, topicMetadata) -> {
+ List<ConsumerGroupPartitionMetadataValue.PartitionMetadata> partitionMetadata = new ArrayList<>();
+ if (!topicMetadata.partitionRacks().isEmpty()) {
+ topicMetadata.partitionRacks().forEach((partition, racks) -> {
+ partitionMetadata.add(new ConsumerGroupPartitionMetadataValue.PartitionMetadata()
+ .setPartition(partition)
+ .setRacks(new ArrayList<>(racks))
+ );
+ });
+ }
+ // If partition rack information is empty, store an empty list in the record.
value.topics().add(new ConsumerGroupPartitionMetadataValue.TopicMetadata()
.setTopicId(topicMetadata.id())
.setTopicName(topicMetadata.name())
.setNumPartitions(topicMetadata.numPartitions())
- )
- );
+ .setPartitionMetadata(partitionMetadata.isEmpty() ? Collections.emptyList() : partitionMetadata)
Review Comment:
Okay makes sense!
--
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
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1275362470
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/TargetAssignmentBuilderTest.java:
##########
@@ -222,8 +231,25 @@ public void testCreateAssignmentMemberSpec() {
Map<String, TopicMetadata> subscriptionMetadata = new HashMap<String, TopicMetadata>() {
{
- put("foo", new TopicMetadata(fooTopicId, "foo", 5));
- put("bar", new TopicMetadata(barTopicId, "bar", 5));
+ put("foo", new TopicMetadata(fooTopicId, "foo", 5,
+ mkMap(
+ mkEntry(0, Collections.emptySet()),
+ mkEntry(1, Collections.emptySet()),
+ mkEntry(2, Collections.emptySet()),
+ mkEntry(3, Collections.emptySet()),
+ mkEntry(4, Collections.emptySet())
+ )
Review Comment:
I kept it just in case we wanted to add the rack information, also wanted some tests with a map with information rather than just an empty map
--
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
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1276710063
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/RecordHelpersTest.java:
##########
@@ -764,4 +813,39 @@ public void testNewOffsetCommitTombstoneRecord() {
Record record = RecordHelpers.newOffsetCommitTombstoneRecord("group-id", "foo", 1);
assertEquals(expectedRecord, record);
}
+
+ // For the purpose of testing let:
+ // a) Number of replicas for each partition = 2
+ // b) Number of racks available in the cluster = 4
Review Comment:
Its applicable to both the mkMapOfPartitionRacks and mkListOfPartitionRacks methods to show why there's two rackIds in each set and why the mod is 4. I got confused when I saw the old tests and didn't understand why a certain calculation was made and what the assumptions behind it were. I initially added it to the javadoc for the first method but realised it probably doesn't belong there. I'm not sure if I should place it inside the method for both or where the correct placement would be
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #14099: KAFKA-14702: Extend server side assignor to support rack aware replica placement
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14099:
URL: https://github.com/apache/kafka/pull/14099#discussion_r1276739854
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/RecordHelpersTest.java:
##########
@@ -764,4 +813,39 @@ public void testNewOffsetCommitTombstoneRecord() {
Record record = RecordHelpers.newOffsetCommitTombstoneRecord("group-id", "foo", 1);
assertEquals(expectedRecord, record);
}
+
+ // For the purpose of testing let:
+ // a) Number of replicas for each partition = 2
+ // b) Number of racks available in the cluster = 4
Review Comment:
The javadoc is definitely a better place than this.
--
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