KAFKA-17747: [7/N] Add consumer group integration test for rack aware assignment #19856

FrankYang0529 · 2025-05-30T05:46:26Z

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 correctly.

Reviewers: David Jacot djacot@confluent.io

Signed-off-by: PoAn Yang <payang@apache.org>

dajac

@FrankYang0529 Thanks for the PR. I left a few comments.

dajac · 2025-06-02T13:02:33Z

...c/main/java/org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManager.java

+            // RackId - set if present
+            membershipManager.rackId().ifPresent(data::setRackId);


I think that it should only be set if it did not change since the last request or if we send a full request.

Updated it. Thanks.

dajac · 2025-06-02T13:03:00Z

...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) {


Let's extend unit tests to cover the rack field.

dajac · 2025-06-02T13:09:42Z

clients/src/main/java/org/apache/kafka/clients/GroupRebalanceConfig.java

+        // 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();
+        }


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;

I change to do this when protocol type is CONSUMER or SHARE, because they all support the field.

dajac · 2025-06-02T13:14:34Z

...tegration-tests/src/test/java/org/apache/kafka/clients/consumer/ConsumerIntegrationTest.java

@@ -216,6 +222,92 @@ public void testLeaderEpoch(ClusterInstance clusterInstance) throws Exception {
        }
    }

+    @ClusterTest(


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?

Yes, I added it to the same case. It uses alterPartitionReassignments to change partitions to different brokers and make sure consumers get new assignments.

Signed-off-by: PoAn Yang <payang@apache.org>

dajac

@FrankYang0529 Thanks for the update. A few more comments.

dajac · 2025-06-03T09:55:10Z

clients/src/main/java/org/apache/kafka/clients/GroupRebalanceConfig.java

+            // 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));


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.

Update ConsumerCoordinator to get rack id from GroupRebalanceConfig. Thanks for the suggestion.

dajac · 2025-06-03T09:55:28Z

...c/main/java/org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManager.java

+            // 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;
+            }


Could we also test this?

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.

Signed-off-by: PoAn Yang <payang@apache.org>

dajac

@FrankYang0529 Thanks for the update. I left a few nits.

dajac · 2025-06-04T12:39:41Z

clients/src/main/java/org/apache/kafka/clients/GroupRebalanceConfig.java

+            // 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.


nit: We can remove this comment now.

Removed it. Thanks.

dajac · 2025-06-04T12:43:26Z

...st/java/org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManagerTest.java

+        when(membershipManager.rackId()).thenReturn(Optional.empty());
+        data = heartbeatState.buildRequestData();
+        assertNull(data.rackId());
+    }


Should we verify that it is sent if all fields are sent again?

I add a case: if member state changes to JOINING again, it sends rackId again. Thanks.

…state is joining Signed-off-by: PoAn Yang <payang@apache.org>

dajac

lgtm, thanks

chia7712

@FrankYang0529 thanks for this patch. it is wonderful, and only two small comments are left. PTAL

chia7712 · 2025-06-04T19:27:25Z

clients/src/test/java/org/apache/kafka/clients/GroupRebalanceConfigTest.java

+            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"


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?

Followup PR: #19906

Thanks for the suggestion.

chia7712 · 2025-06-04T19:28:53Z

...nts-integration-tests/src/test/java/org/apache/kafka/clients/consumer/RackAwareAssignor.java

+/**
+ * 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.


It also requires all member to subscribe same topics, right?

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?

Do we want to make it support heterogeneous group?

no, it is used by testing only so it should be fine.

… assignment (apache#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 correctly. Reviewers: David Jacot <djacot@confluent.io> --------- Signed-off-by: PoAn Yang <payang@apache.org>

KAFKA-17747: Add integration test for rack aware assignment

392b850

Signed-off-by: PoAn Yang <payang@apache.org>

FrankYang0529 changed the title ~~KAFKA-17747: Add integration test for rack aware assignment (wip)~~ KAFKA-17747: [7/N] Add integration test for rack aware assignment (wip) May 30, 2025

github-actions bot added consumer build clients labels May 30, 2025

FrankYang0529 mentioned this pull request May 30, 2025

KAFKA-17747: [4/N] Replace subscription metadata with metadata hash in consumer group #19761

Merged

FrankYang0529 added 2 commits May 30, 2025 21:47

ignore undefined client.rack key and empty value

af5a620

Signed-off-by: PoAn Yang <payang@apache.org>

Merge branch 'trunk' into KAFKA-17747-7

af08e87

FrankYang0529 changed the title ~~KAFKA-17747: [7/N] Add integration test for rack aware assignment (wip)~~ KAFKA-17747: [7/N] Add integration test for rack aware assignment May 30, 2025

FrankYang0529 requested review from dajac and lianetm May 30, 2025 23:46

dajac reviewed Jun 2, 2025

View reviewed changes

FrankYang0529 added 2 commits June 2, 2025 23:01

Merge branch 'trunk' into KAFKA-17747-7

cf58ce6

address comment

a3104f8

Signed-off-by: PoAn Yang <payang@apache.org>

FrankYang0529 mentioned this pull request Jun 3, 2025

KAFKA-17747: [6/N] Replace subscription metadata with metadata hash in share group #19796

Merged

FrankYang0529 requested a review from dajac June 3, 2025 09:27

dajac reviewed Jun 3, 2025

View reviewed changes

FrankYang0529 added 2 commits June 3, 2025 18:18

Merge branch 'trunk' into KAFKA-17747-7

1b92ef7

address comment

55daa01

Signed-off-by: PoAn Yang <payang@apache.org>

github-actions bot added the connect label Jun 3, 2025

FrankYang0529 requested a review from dajac June 3, 2025 14:09

Merge branch 'trunk' into KAFKA-17747-7

fe83294

FrankYang0529 changed the title ~~KAFKA-17747: [7/N] Add integration test for rack aware assignment~~ KAFKA-17747: [7/N] Add consumer group integration test for rack aware assignment Jun 4, 2025

dajac reviewed Jun 4, 2025

View reviewed changes

FrankYang0529 added 2 commits June 4, 2025 22:53

Merge branch 'trunk' into KAFKA-17747-7

a979564

remove comment and add test case about sending rackId only if member …

157d171

…state is joining Signed-off-by: PoAn Yang <payang@apache.org>

FrankYang0529 mentioned this pull request Jun 4, 2025

KAFKA-19369: Add group.share.assignors config and integration test #19900

Merged

dajac approved these changes Jun 4, 2025

View reviewed changes

dajac merged commit 949617b into apache:trunk Jun 4, 2025
25 checks passed

chia7712 reviewed Jun 4, 2025

View reviewed changes

FrankYang0529 deleted the KAFKA-17747-7 branch June 5, 2025 02:41

		// RackId - set if present
		membershipManager.rackId().ifPresent(data::setRackId);

		@@ -142,7 +142,7 @@ private ConsumerMembershipManager createMembershipManagerJoiningGroup(String gro

		private ConsumerMembershipManager createMembershipManager(String groupInstanceId) {

KAFKA-17747: [7/N] Add consumer group integration test for rack aware assignment #19856

KAFKA-17747: [7/N] Add consumer group integration test for rack aware assignment #19856

Uh oh!

Conversation

FrankYang0529 commented May 30, 2025 • edited by dajac Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

dajac left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dajac left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dajac left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dajac left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

chia7712 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

FrankYang0529 commented May 30, 2025 •

edited by dajac

Loading