You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "FrankYang0529 (via GitHub)" <gi...@apache.org> on 2025/05/30 05:46:29 UTC
[PR] KAFKA-17747: Add integration test for rack aware assignment (wip) [kafka]
FrankYang0529 opened a new pull request, #19856:
URL: https://github.com/apache/kafka/pull/19856
* Add `RackAwareAssignor`. It uses `racksForPartition` to check the rack id of a partition and assign it to a member which has the same rack id.
* Add `ConsumerIntegrationTest#testRackAwareAssignment` to check `racksForPartition` works expected.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [7/N] Add consumer group integration test for rack aware assignment [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #19856:
URL: https://github.com/apache/kafka/pull/19856#discussion_r2126508197
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManagerTest.java:
##########
@@ -1009,6 +1009,28 @@ public void testRegexInJoiningHeartbeat() {
assertNull(data.subscribedTopicRegex());
}
+ @Test
+ public void testRackIdInHeartbeatLifecycle() {
+ heartbeatState = new HeartbeatState(subscriptions, membershipManager, DEFAULT_MAX_POLL_INTERVAL_MS);
+ createHeartbeatRequestStateWithZeroHeartbeatInterval();
+
+ // Initial heartbeat with rackId
+ mockJoiningMemberData(null);
+ when(membershipManager.rackId()).thenReturn(Optional.of("rack1"));
+ ConsumerGroupHeartbeatRequestData data = heartbeatState.buildRequestData();
+ assertEquals("rack1", data.rackId());
+
+ // RackId not included in HB if not updated
+ when(membershipManager.state()).thenReturn(MemberState.STABLE);
+ data = heartbeatState.buildRequestData();
+ assertNull(data.rackId());
+
+ // Empty rackId not included in HB
+ when(membershipManager.rackId()).thenReturn(Optional.empty());
+ data = heartbeatState.buildRequestData();
+ assertNull(data.rackId());
+ }
Review Comment:
Should we verify that it is sent if all fields are sent again?
##########
clients/src/main/java/org/apache/kafka/clients/GroupRebalanceConfig.java:
##########
@@ -53,8 +54,14 @@ public GroupRebalanceConfig(AbstractConfig config, ProtocolType protocolType) {
// Consumer and Connect use different config names for defining rebalance timeout
if ((protocolType == ProtocolType.CONSUMER) || (protocolType == ProtocolType.SHARE)) {
this.rebalanceTimeoutMs = config.getInt(CommonClientConfigs.MAX_POLL_INTERVAL_MS_CONFIG);
+ // Connect doesn't support rack id.
+ // The GroupCoordinatorService throws error if the rackId is empty. The default value of client.rack is empty string.
+ // Skip empty rackId to avoid InvalidRequestException.
Review Comment:
nit: We can remove this comment now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [7/N] Add consumer group integration test for rack aware assignment [kafka]
Posted by "stanislavkozlovski (via GitHub)" <gi...@apache.org>.
stanislavkozlovski commented on code in PR #19856:
URL: https://github.com/apache/kafka/pull/19856#discussion_r2265427192
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/RackAwareAssignor.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.clients.consumer;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.coordinator.group.api.assignor.ConsumerGroupPartitionAssignor;
+import org.apache.kafka.coordinator.group.api.assignor.GroupAssignment;
+import org.apache.kafka.coordinator.group.api.assignor.GroupSpec;
+import org.apache.kafka.coordinator.group.api.assignor.MemberAssignment;
+import org.apache.kafka.coordinator.group.api.assignor.PartitionAssignorException;
+import org.apache.kafka.coordinator.group.api.assignor.SubscribedTopicDescriber;
+import org.apache.kafka.coordinator.group.modern.MemberAssignmentImpl;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * The RackAwareAssignor is a consumer group partition assignor that takes into account the rack
+ * information of the members when assigning partitions to them.
+ * It needs all brokers and members to have rack information available.
+ */
+public class RackAwareAssignor implements ConsumerGroupPartitionAssignor {
Review Comment:
yeah I ack this is a test one, I was wondering if you had any context regarding the rack aware assignor. If the rack info had to be removed from 4.0 due to memory issues, I understand why it wasn't included
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [7/N] Add consumer group integration test for rack aware assignment [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #19856:
URL: https://github.com/apache/kafka/pull/19856#discussion_r2267150942
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/RackAwareAssignor.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.clients.consumer;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.coordinator.group.api.assignor.ConsumerGroupPartitionAssignor;
+import org.apache.kafka.coordinator.group.api.assignor.GroupAssignment;
+import org.apache.kafka.coordinator.group.api.assignor.GroupSpec;
+import org.apache.kafka.coordinator.group.api.assignor.MemberAssignment;
+import org.apache.kafka.coordinator.group.api.assignor.PartitionAssignorException;
+import org.apache.kafka.coordinator.group.api.assignor.SubscribedTopicDescriber;
+import org.apache.kafka.coordinator.group.modern.MemberAssignmentImpl;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * The RackAwareAssignor is a consumer group partition assignor that takes into account the rack
+ * information of the members when assigning partitions to them.
+ * It needs all brokers and members to have rack information available.
+ */
+public class RackAwareAssignor implements ConsumerGroupPartitionAssignor {
Review Comment:
We already have configuration `group.consumer.assignors` to support customized assignors. For KAFKA-19387, we will add rack-aware support to built-in assignors `RangeAssignor` and `UniformAssignor`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [7/N] Add consumer group integration test for rack aware assignment [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #19856:
URL: https://github.com/apache/kafka/pull/19856#discussion_r2266723624
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/RackAwareAssignor.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.clients.consumer;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.coordinator.group.api.assignor.ConsumerGroupPartitionAssignor;
+import org.apache.kafka.coordinator.group.api.assignor.GroupAssignment;
+import org.apache.kafka.coordinator.group.api.assignor.GroupSpec;
+import org.apache.kafka.coordinator.group.api.assignor.MemberAssignment;
+import org.apache.kafka.coordinator.group.api.assignor.PartitionAssignorException;
+import org.apache.kafka.coordinator.group.api.assignor.SubscribedTopicDescriber;
+import org.apache.kafka.coordinator.group.modern.MemberAssignmentImpl;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * The RackAwareAssignor is a consumer group partition assignor that takes into account the rack
+ * information of the members when assigning partitions to them.
+ * It needs all brokers and members to have rack information available.
+ */
+public class RackAwareAssignor implements ConsumerGroupPartitionAssignor {
Review Comment:
We will add rack aware support to Range and Uniform assignors in 4.2. It's tracked in https://issues.apache.org/jira/browse/KAFKA-19387.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [7/N] Add consumer group integration test for rack aware assignment [kafka]
Posted by "stanislavkozlovski (via GitHub)" <gi...@apache.org>.
stanislavkozlovski commented on code in PR #19856:
URL: https://github.com/apache/kafka/pull/19856#discussion_r2265381736
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/RackAwareAssignor.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.clients.consumer;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.coordinator.group.api.assignor.ConsumerGroupPartitionAssignor;
+import org.apache.kafka.coordinator.group.api.assignor.GroupAssignment;
+import org.apache.kafka.coordinator.group.api.assignor.GroupSpec;
+import org.apache.kafka.coordinator.group.api.assignor.MemberAssignment;
+import org.apache.kafka.coordinator.group.api.assignor.PartitionAssignorException;
+import org.apache.kafka.coordinator.group.api.assignor.SubscribedTopicDescriber;
+import org.apache.kafka.coordinator.group.modern.MemberAssignmentImpl;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * The RackAwareAssignor is a consumer group partition assignor that takes into account the rack
+ * information of the members when assigning partitions to them.
+ * It needs all brokers and members to have rack information available.
+ */
+public class RackAwareAssignor implements ConsumerGroupPartitionAssignor {
Review Comment:
@dajac @FrankYang0529 @chia7712 is there a reason this assignor doesn't exist by default in the non-testing code? I commented here https://issues.apache.org/jira/browse/KAFKA-14450?focusedCommentId=18007549&page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel#comment-18007549 that I believe KIP-848 introduced a feature regression by not supporting rack-aware assignment out of the box. It seems like quite the obvious miss, so I presume I may be missing something. Please let me know
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [7/N] Add integration test for rack aware assignment [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #19856:
URL: https://github.com/apache/kafka/pull/19856#discussion_r2121604977
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ConsumerIntegrationTest.java:
##########
@@ -216,6 +222,92 @@ public void testLeaderEpoch(ClusterInstance clusterInstance) throws Exception {
}
}
+ @ClusterTest(
Review Comment:
Yes, I added it to the same case. It uses `alterPartitionReassignments` to change partitions to different brokers and make sure consumers get new assignments.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [7/N] Add consumer group integration test for rack aware assignment [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19856:
URL: https://github.com/apache/kafka/pull/19856#discussion_r2267175543
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/RackAwareAssignor.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.clients.consumer;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.coordinator.group.api.assignor.ConsumerGroupPartitionAssignor;
+import org.apache.kafka.coordinator.group.api.assignor.GroupAssignment;
+import org.apache.kafka.coordinator.group.api.assignor.GroupSpec;
+import org.apache.kafka.coordinator.group.api.assignor.MemberAssignment;
+import org.apache.kafka.coordinator.group.api.assignor.PartitionAssignorException;
+import org.apache.kafka.coordinator.group.api.assignor.SubscribedTopicDescriber;
+import org.apache.kafka.coordinator.group.modern.MemberAssignmentImpl;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * The RackAwareAssignor is a consumer group partition assignor that takes into account the rack
+ * information of the members when assigning partitions to them.
+ * It needs all brokers and members to have rack information available.
+ */
+public class RackAwareAssignor implements ConsumerGroupPartitionAssignor {
Review Comment:
Got it. Thanks for your explanation!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [7/N] Add integration test for rack aware assignment [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #19856:
URL: https://github.com/apache/kafka/pull/19856#discussion_r2121092739
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerMembershipManagerTest.java:
##########
@@ -142,7 +142,7 @@ private ConsumerMembershipManager createMembershipManagerJoiningGroup(String gro
private ConsumerMembershipManager createMembershipManager(String groupInstanceId) {
Review Comment:
Let's extend unit tests to cover the rack field.
##########
clients/src/main/java/org/apache/kafka/clients/GroupRebalanceConfig.java:
##########
@@ -73,6 +74,16 @@ public GroupRebalanceConfig(AbstractConfig config, ProtocolType protocolType) {
this.groupInstanceId = Optional.empty();
}
+ // The WorkerGroupMember in connect module also uses this class, but there is no client.rack in DistributedConfig.
+ // Ignore the rackId in that case to avoid ConfigException.
+ // The GroupCoordinatorService throws error if the rackId is empty. The default value of client.rack is empty string.
+ // Skip empty rackId to avoid InvalidRequestException.
+ if (config.values().containsKey(CommonClientConfigs.CLIENT_RACK_CONFIG) && !config.getString(CommonClientConfigs.CLIENT_RACK_CONFIG).isEmpty()) {
+ this.rackId = Optional.ofNullable(config.getString(CommonClientConfigs.CLIENT_RACK_CONFIG));
+ } else {
+ this.rackId = Optional.empty();
+ }
Review Comment:
- Should we only do this if the protocolType is CONSUMER?
- It may be simpler to get the config in a local variable and to set this.rack if non-null and non-empty;
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManager.java:
##########
@@ -262,6 +262,9 @@ public ConsumerGroupHeartbeatRequestData buildRequestData() {
// InstanceId - set if present
membershipManager.groupInstanceId().ifPresent(data::setInstanceId);
+ // RackId - set if present
+ membershipManager.rackId().ifPresent(data::setRackId);
Review Comment:
I think that it should only be set if it did not change since the last request or if we send a full request.
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ConsumerIntegrationTest.java:
##########
@@ -216,6 +222,92 @@ public void testLeaderEpoch(ClusterInstance clusterInstance) throws Exception {
}
}
+ @ClusterTest(
Review Comment:
I wonder whether we could add another test which verifies that a rebalance is triggered when the racks of a partition has changed. Have you considered it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [7/N] Add consumer group integration test for rack aware assignment [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19856:
URL: https://github.com/apache/kafka/pull/19856#discussion_r2266858774
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/RackAwareAssignor.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.clients.consumer;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.coordinator.group.api.assignor.ConsumerGroupPartitionAssignor;
+import org.apache.kafka.coordinator.group.api.assignor.GroupAssignment;
+import org.apache.kafka.coordinator.group.api.assignor.GroupSpec;
+import org.apache.kafka.coordinator.group.api.assignor.MemberAssignment;
+import org.apache.kafka.coordinator.group.api.assignor.PartitionAssignorException;
+import org.apache.kafka.coordinator.group.api.assignor.SubscribedTopicDescriber;
+import org.apache.kafka.coordinator.group.modern.MemberAssignmentImpl;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * The RackAwareAssignor is a consumer group partition assignor that takes into account the rack
+ * information of the members when assigning partitions to them.
+ * It needs all brokers and members to have rack information available.
+ */
+public class RackAwareAssignor implements ConsumerGroupPartitionAssignor {
Review Comment:
@FrankYang0529 Does it need a KIP? or the discussion is already included by another KIP?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [7/N] Add consumer group integration test for rack aware assignment [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19856:
URL: https://github.com/apache/kafka/pull/19856#discussion_r2267124755
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/RackAwareAssignor.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.clients.consumer;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.coordinator.group.api.assignor.ConsumerGroupPartitionAssignor;
+import org.apache.kafka.coordinator.group.api.assignor.GroupAssignment;
+import org.apache.kafka.coordinator.group.api.assignor.GroupSpec;
+import org.apache.kafka.coordinator.group.api.assignor.MemberAssignment;
+import org.apache.kafka.coordinator.group.api.assignor.PartitionAssignorException;
+import org.apache.kafka.coordinator.group.api.assignor.SubscribedTopicDescriber;
+import org.apache.kafka.coordinator.group.modern.MemberAssignmentImpl;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * The RackAwareAssignor is a consumer group partition assignor that takes into account the rack
+ * information of the members when assigning partitions to them.
+ * It needs all brokers and members to have rack information available.
+ */
+public class RackAwareAssignor implements ConsumerGroupPartitionAssignor {
Review Comment:
> I think we can have PRs first. If it's too complicated, we can write a KIP to discuss it.
The out-of-box assignor is a kind of public API, right? I mean, users could configure it in the server properties
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [7/N] Add consumer group integration test for rack aware assignment [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #19856:
URL: https://github.com/apache/kafka/pull/19856#discussion_r2126838780
##########
clients/src/main/java/org/apache/kafka/clients/GroupRebalanceConfig.java:
##########
@@ -53,8 +54,14 @@ public GroupRebalanceConfig(AbstractConfig config, ProtocolType protocolType) {
// Consumer and Connect use different config names for defining rebalance timeout
if ((protocolType == ProtocolType.CONSUMER) || (protocolType == ProtocolType.SHARE)) {
this.rebalanceTimeoutMs = config.getInt(CommonClientConfigs.MAX_POLL_INTERVAL_MS_CONFIG);
+ // Connect doesn't support rack id.
+ // The GroupCoordinatorService throws error if the rackId is empty. The default value of client.rack is empty string.
+ // Skip empty rackId to avoid InvalidRequestException.
Review Comment:
Removed it. Thanks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [7/N] Add consumer group integration test for rack aware assignment [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19856:
URL: https://github.com/apache/kafka/pull/19856#discussion_r2127290138
##########
clients/src/test/java/org/apache/kafka/clients/GroupRebalanceConfigTest.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * 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.clients;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class GroupRebalanceConfigTest {
+
+ @ParameterizedTest
+ @EnumSource(value = GroupRebalanceConfig.ProtocolType.class, names = {"CONSUMER", "SHARE"})
+ void testRackIdIsEmptyIfNoDefined(GroupRebalanceConfig.ProtocolType protocolType) {
+ GroupRebalanceConfig groupRebalanceConfig = new GroupRebalanceConfig(
+ new ConsumerConfig(Map.of(
+ ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer",
+ ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer"
+ )),
+ protocolType
+ );
+ assertTrue(groupRebalanceConfig.rackId.isEmpty());
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = GroupRebalanceConfig.ProtocolType.class, names = {"CONSUMER", "SHARE"})
+ void testRackIdIsNotEmptyIfDefined(GroupRebalanceConfig.ProtocolType protocolType) {
+ GroupRebalanceConfig groupRebalanceConfig = new GroupRebalanceConfig(
+ new ConsumerConfig(Map.of(
+ ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer",
+ ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer",
+ ConsumerConfig.CLIENT_RACK_CONFIG, "rack1"
Review Comment:
Since an empty rack is illegal in the heartbeat RPC, the empty rack should be handled as no rack. That is a critical behavior, so could you please add unit test for it?
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/RackAwareAssignor.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.clients.consumer;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.coordinator.group.api.assignor.ConsumerGroupPartitionAssignor;
+import org.apache.kafka.coordinator.group.api.assignor.GroupAssignment;
+import org.apache.kafka.coordinator.group.api.assignor.GroupSpec;
+import org.apache.kafka.coordinator.group.api.assignor.MemberAssignment;
+import org.apache.kafka.coordinator.group.api.assignor.PartitionAssignorException;
+import org.apache.kafka.coordinator.group.api.assignor.SubscribedTopicDescriber;
+import org.apache.kafka.coordinator.group.modern.MemberAssignmentImpl;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * The RackAwareAssignor is a consumer group partition assignor that takes into account the rack
+ * information of the members when assigning partitions to them.
+ * It needs all brokers and members to have rack information available.
Review Comment:
It also requires all member to subscribe same topics, right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [7/N] Add consumer group integration test for rack aware assignment [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac merged PR #19856:
URL: https://github.com/apache/kafka/pull/19856
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [7/N] Add consumer group integration test for rack aware assignment [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #19856:
URL: https://github.com/apache/kafka/pull/19856#discussion_r2127917487
##########
clients/src/test/java/org/apache/kafka/clients/GroupRebalanceConfigTest.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * 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.clients;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class GroupRebalanceConfigTest {
+
+ @ParameterizedTest
+ @EnumSource(value = GroupRebalanceConfig.ProtocolType.class, names = {"CONSUMER", "SHARE"})
+ void testRackIdIsEmptyIfNoDefined(GroupRebalanceConfig.ProtocolType protocolType) {
+ GroupRebalanceConfig groupRebalanceConfig = new GroupRebalanceConfig(
+ new ConsumerConfig(Map.of(
+ ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer",
+ ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer"
+ )),
+ protocolType
+ );
+ assertTrue(groupRebalanceConfig.rackId.isEmpty());
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = GroupRebalanceConfig.ProtocolType.class, names = {"CONSUMER", "SHARE"})
+ void testRackIdIsNotEmptyIfDefined(GroupRebalanceConfig.ProtocolType protocolType) {
+ GroupRebalanceConfig groupRebalanceConfig = new GroupRebalanceConfig(
+ new ConsumerConfig(Map.of(
+ ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer",
+ ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer",
+ ConsumerConfig.CLIENT_RACK_CONFIG, "rack1"
Review Comment:
Followup PR: https://github.com/apache/kafka/pull/19906
Thanks for the suggestion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [7/N] Add integration test for rack aware assignment [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #19856:
URL: https://github.com/apache/kafka/pull/19856#discussion_r2123571086
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManager.java:
##########
@@ -306,6 +307,13 @@ public ConsumerGroupHeartbeatRequestData buildRequestData() {
sentFields.localAssignment = local;
}
+ // RackId - sent when joining or if it has changed since the last heartbeat
+ String rackId = membershipManager.rackId().orElse(null);
+ if (sendAllFields || !Objects.equals(rackId, sentFields.rackId)) {
+ data.setRackId(rackId);
+ sentFields.rackId = rackId;
+ }
Review Comment:
Added related test `testRackIdInHeartbeatLifecycle`. BTW, I remove `rackId` from `sentFields`. Users cannot reconfigure it, so we don't need to keep last sent data in `sentFields`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [7/N] Add integration test for rack aware assignment [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #19856:
URL: https://github.com/apache/kafka/pull/19856#discussion_r2121601064
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManager.java:
##########
@@ -262,6 +262,9 @@ public ConsumerGroupHeartbeatRequestData buildRequestData() {
// InstanceId - set if present
membershipManager.groupInstanceId().ifPresent(data::setInstanceId);
+ // RackId - set if present
+ membershipManager.rackId().ifPresent(data::setRackId);
Review Comment:
Updated it. Thanks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [7/N] Add consumer group integration test for rack aware assignment [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19856:
URL: https://github.com/apache/kafka/pull/19856#discussion_r2128038915
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/RackAwareAssignor.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.clients.consumer;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.coordinator.group.api.assignor.ConsumerGroupPartitionAssignor;
+import org.apache.kafka.coordinator.group.api.assignor.GroupAssignment;
+import org.apache.kafka.coordinator.group.api.assignor.GroupSpec;
+import org.apache.kafka.coordinator.group.api.assignor.MemberAssignment;
+import org.apache.kafka.coordinator.group.api.assignor.PartitionAssignorException;
+import org.apache.kafka.coordinator.group.api.assignor.SubscribedTopicDescriber;
+import org.apache.kafka.coordinator.group.modern.MemberAssignmentImpl;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * The RackAwareAssignor is a consumer group partition assignor that takes into account the rack
+ * information of the members when assigning partitions to them.
+ * It needs all brokers and members to have rack information available.
Review Comment:
> Do we want to make it support heterogeneous group?
no, it is used by testing only so it should be fine.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [7/N] Add consumer group integration test for rack aware assignment [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #19856:
URL: https://github.com/apache/kafka/pull/19856#discussion_r2126840075
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManagerTest.java:
##########
@@ -1009,6 +1009,28 @@ public void testRegexInJoiningHeartbeat() {
assertNull(data.subscribedTopicRegex());
}
+ @Test
+ public void testRackIdInHeartbeatLifecycle() {
+ heartbeatState = new HeartbeatState(subscriptions, membershipManager, DEFAULT_MAX_POLL_INTERVAL_MS);
+ createHeartbeatRequestStateWithZeroHeartbeatInterval();
+
+ // Initial heartbeat with rackId
+ mockJoiningMemberData(null);
+ when(membershipManager.rackId()).thenReturn(Optional.of("rack1"));
+ ConsumerGroupHeartbeatRequestData data = heartbeatState.buildRequestData();
+ assertEquals("rack1", data.rackId());
+
+ // RackId not included in HB if not updated
+ when(membershipManager.state()).thenReturn(MemberState.STABLE);
+ data = heartbeatState.buildRequestData();
+ assertNull(data.rackId());
+
+ // Empty rackId not included in HB
+ when(membershipManager.rackId()).thenReturn(Optional.empty());
+ data = heartbeatState.buildRequestData();
+ assertNull(data.rackId());
+ }
Review Comment:
I add a case: if member state changes to JOINING again, it sends `rackId` again. Thanks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [7/N] Add consumer group integration test for rack aware assignment [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19856:
URL: https://github.com/apache/kafka/pull/19856#discussion_r2265386219
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/RackAwareAssignor.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.clients.consumer;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.coordinator.group.api.assignor.ConsumerGroupPartitionAssignor;
+import org.apache.kafka.coordinator.group.api.assignor.GroupAssignment;
+import org.apache.kafka.coordinator.group.api.assignor.GroupSpec;
+import org.apache.kafka.coordinator.group.api.assignor.MemberAssignment;
+import org.apache.kafka.coordinator.group.api.assignor.PartitionAssignorException;
+import org.apache.kafka.coordinator.group.api.assignor.SubscribedTopicDescriber;
+import org.apache.kafka.coordinator.group.modern.MemberAssignmentImpl;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * The RackAwareAssignor is a consumer group partition assignor that takes into account the rack
+ * information of the members when assigning partitions to them.
+ * It needs all brokers and members to have rack information available.
+ */
+public class RackAwareAssignor implements ConsumerGroupPartitionAssignor {
Review Comment:
The `RackAwareAssignor` is incomplete, as it does not support heterogeneous groups. Hence, it should not be considered production-ready.
That said, it would be valuable to provide a production-ready `RackAwareAssignor`. We could propose a KIP for it as a follow-up.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [7/N] Add integration test for rack aware assignment [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #19856:
URL: https://github.com/apache/kafka/pull/19856#discussion_r2121602544
##########
clients/src/main/java/org/apache/kafka/clients/GroupRebalanceConfig.java:
##########
@@ -73,6 +74,16 @@ public GroupRebalanceConfig(AbstractConfig config, ProtocolType protocolType) {
this.groupInstanceId = Optional.empty();
}
+ // The WorkerGroupMember in connect module also uses this class, but there is no client.rack in DistributedConfig.
+ // Ignore the rackId in that case to avoid ConfigException.
+ // The GroupCoordinatorService throws error if the rackId is empty. The default value of client.rack is empty string.
+ // Skip empty rackId to avoid InvalidRequestException.
+ if (config.values().containsKey(CommonClientConfigs.CLIENT_RACK_CONFIG) && !config.getString(CommonClientConfigs.CLIENT_RACK_CONFIG).isEmpty()) {
+ this.rackId = Optional.ofNullable(config.getString(CommonClientConfigs.CLIENT_RACK_CONFIG));
+ } else {
+ this.rackId = Optional.empty();
+ }
Review Comment:
I change to do this when protocol type is CONSUMER or SHARE, because they all support the field.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [7/N] Add integration test for rack aware assignment [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #19856:
URL: https://github.com/apache/kafka/pull/19856#discussion_r2123327253
##########
clients/src/main/java/org/apache/kafka/clients/GroupRebalanceConfig.java:
##########
@@ -53,8 +54,15 @@ public GroupRebalanceConfig(AbstractConfig config, ProtocolType protocolType) {
// Consumer and Connect use different config names for defining rebalance timeout
if ((protocolType == ProtocolType.CONSUMER) || (protocolType == ProtocolType.SHARE)) {
this.rebalanceTimeoutMs = config.getInt(CommonClientConfigs.MAX_POLL_INTERVAL_MS_CONFIG);
+ // Connect doesn't support rack id.
+ // The GroupCoordinatorService throws error if the rackId is empty. The default value of client.rack is empty string.
+ // Skip empty rackId to avoid InvalidRequestException.
+ this.rackId = config.getString(CommonClientConfigs.CLIENT_RACK_CONFIG).isEmpty() ?
+ Optional.empty() :
+ Optional.of(config.getString(CommonClientConfigs.CLIENT_RACK_CONFIG));
Review Comment:
In `ConsumerCoordinator`, we have the following code:
```
this.rackId = rackId == null || rackId.isEmpty() ? Optional.empty() : Optional.of(rackId);
```
Hence, we can use the same logic for the classic protocol. I wonder if we could just do the following here for all protocols.
```
String rackId = config.getString(CommonClientConfigs.CLIENT_RACK_CONFIG);
this.rackId = rackId == null || rackId.isEmpty() ? Optional.empty() : Optional.of(rackId);
```
We may also want to update `ConsumerCoordinator` to get it from `GroupRebalanceConfig` too.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManager.java:
##########
@@ -306,6 +307,13 @@ public ConsumerGroupHeartbeatRequestData buildRequestData() {
sentFields.localAssignment = local;
}
+ // RackId - sent when joining or if it has changed since the last heartbeat
+ String rackId = membershipManager.rackId().orElse(null);
+ if (sendAllFields || !Objects.equals(rackId, sentFields.rackId)) {
+ data.setRackId(rackId);
+ sentFields.rackId = rackId;
+ }
Review Comment:
Could we also test 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
Re: [PR] KAFKA-17747: [7/N] Add integration test for rack aware assignment [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #19856:
URL: https://github.com/apache/kafka/pull/19856#discussion_r2123564124
##########
clients/src/main/java/org/apache/kafka/clients/GroupRebalanceConfig.java:
##########
@@ -53,8 +54,15 @@ public GroupRebalanceConfig(AbstractConfig config, ProtocolType protocolType) {
// Consumer and Connect use different config names for defining rebalance timeout
if ((protocolType == ProtocolType.CONSUMER) || (protocolType == ProtocolType.SHARE)) {
this.rebalanceTimeoutMs = config.getInt(CommonClientConfigs.MAX_POLL_INTERVAL_MS_CONFIG);
+ // Connect doesn't support rack id.
+ // The GroupCoordinatorService throws error if the rackId is empty. The default value of client.rack is empty string.
+ // Skip empty rackId to avoid InvalidRequestException.
+ this.rackId = config.getString(CommonClientConfigs.CLIENT_RACK_CONFIG).isEmpty() ?
+ Optional.empty() :
+ Optional.of(config.getString(CommonClientConfigs.CLIENT_RACK_CONFIG));
Review Comment:
Update `ConsumerCoordinator` to get rack id from `GroupRebalanceConfig`. Thanks for the suggestion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [7/N] Add consumer group integration test for rack aware assignment [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #19856:
URL: https://github.com/apache/kafka/pull/19856#discussion_r2266973705
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/RackAwareAssignor.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.clients.consumer;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.coordinator.group.api.assignor.ConsumerGroupPartitionAssignor;
+import org.apache.kafka.coordinator.group.api.assignor.GroupAssignment;
+import org.apache.kafka.coordinator.group.api.assignor.GroupSpec;
+import org.apache.kafka.coordinator.group.api.assignor.MemberAssignment;
+import org.apache.kafka.coordinator.group.api.assignor.PartitionAssignorException;
+import org.apache.kafka.coordinator.group.api.assignor.SubscribedTopicDescriber;
+import org.apache.kafka.coordinator.group.modern.MemberAssignmentImpl;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * The RackAwareAssignor is a consumer group partition assignor that takes into account the rack
+ * information of the members when assigning partitions to them.
+ * It needs all brokers and members to have rack information available.
+ */
+public class RackAwareAssignor implements ConsumerGroupPartitionAssignor {
Review Comment:
I think we can have PRs first. If it's too complicated, we can write a KIP to discuss it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [7/N] Add consumer group integration test for rack aware assignment [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #19856:
URL: https://github.com/apache/kafka/pull/19856#discussion_r2128036238
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/RackAwareAssignor.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.clients.consumer;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.coordinator.group.api.assignor.ConsumerGroupPartitionAssignor;
+import org.apache.kafka.coordinator.group.api.assignor.GroupAssignment;
+import org.apache.kafka.coordinator.group.api.assignor.GroupSpec;
+import org.apache.kafka.coordinator.group.api.assignor.MemberAssignment;
+import org.apache.kafka.coordinator.group.api.assignor.PartitionAssignorException;
+import org.apache.kafka.coordinator.group.api.assignor.SubscribedTopicDescriber;
+import org.apache.kafka.coordinator.group.modern.MemberAssignmentImpl;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * The RackAwareAssignor is a consumer group partition assignor that takes into account the rack
+ * information of the members when assigning partitions to them.
+ * It needs all brokers and members to have rack information available.
Review Comment:
Yes, this only checks subscription of first member and assume all members use same subscription. This is a sample assignor under test module. Do we want to make it support heterogeneous group?
--
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