You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by da...@apache.org on 2024/12/09 08:39:29 UTC
(kafka) branch trunk updated: KAFKA-18134; Disallow group upgrades when custom assignors are used (#18046)
This is an automated email from the ASF dual-hosted git repository.
dajac pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 9ae1b0f0178 KAFKA-18134; Disallow group upgrades when custom assignors are used (#18046)
9ae1b0f0178 is described below
commit 9ae1b0f01784d314d21100e7d021467626b20137
Author: Sean Quah <sq...@confluent.io>
AuthorDate: Mon Dec 9 08:39:22 2024 +0000
KAFKA-18134; Disallow group upgrades when custom assignors are used (#18046)
Disallow upgrades from classic groups to consumer groups when any member's assignment has non-empty userData.
Reviewers: David Jacot <dj...@confluent.io>
---
.../server/ConsumerProtocolMigrationTest.scala | 120 ++++++++++++++++++++-
.../coordinator/group/GroupMetadataManager.java | 39 +++++--
.../group/modern/consumer/ConsumerGroup.java | 16 ++-
.../group/GroupMetadataManagerTest.java | 115 ++++++++++++++++++++
4 files changed, 272 insertions(+), 18 deletions(-)
diff --git a/core/src/test/scala/unit/kafka/server/ConsumerProtocolMigrationTest.scala b/core/src/test/scala/unit/kafka/server/ConsumerProtocolMigrationTest.scala
index 31821ceb3a3..666557fe15c 100644
--- a/core/src/test/scala/unit/kafka/server/ConsumerProtocolMigrationTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ConsumerProtocolMigrationTest.scala
@@ -486,6 +486,121 @@ class ConsumerProtocolMigrationTest(cluster: ClusterInstance) extends GroupCoord
)
}
+ /**
+ * The test method checks the following scenario:
+ * 1. Creating a classic group with member 1, whose assignment has non-empty user data.
+ * 2. Member 2 using consumer protocol joins. The group cannot be upgraded and the join is
+ * rejected.
+ * 3. Member 1 leaves.
+ * 4. Member 2 using consumer protocol joins. The group is upgraded.
+ */
+ @ClusterTest(
+ serverProperties = Array(
+ new ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG, value = "1"),
+ new ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1"),
+ new ClusterConfigProperty(key = GroupCoordinatorConfig.CONSUMER_GROUP_MIGRATION_POLICY_CONFIG, value = "bidirectional")
+ )
+ )
+ def testOnlineMigrationWithNonEmptyUserDataInAssignment(): Unit = {
+ // Creates the __consumer_offsets topics because it won't be created automatically
+ // in this test because it does not use FindCoordinator API.
+ createOffsetsTopic()
+
+ // Create the topic.
+ createTopic(
+ topic = "foo",
+ numPartitions = 3
+ )
+
+ // Classic member 1 joins the classic group.
+ val groupId = "grp"
+
+ val memberId1 = joinDynamicConsumerGroupWithOldProtocol(
+ groupId = groupId,
+ metadata = metadata(List.empty),
+ assignment = assignment(List(0, 1, 2), ByteBuffer.allocate(1))
+ )._1
+
+ // The joining request with a consumer group member 2 is rejected.
+ val errorMessage = consumerGroupHeartbeat(
+ groupId = groupId,
+ memberId = Uuid.randomUuid.toString,
+ rebalanceTimeoutMs = 5 * 60 * 1000,
+ subscribedTopicNames = List("foo"),
+ topicPartitions = List.empty,
+ expectedError = Errors.GROUP_ID_NOT_FOUND
+ ).errorMessage
+
+ assertEquals(
+ "Cannot upgrade classic group grp to consumer group because an unsupported custom assignor is in use. " +
+ "Please refer to the documentation or switch to a default assignor before re-attempting the upgrade.",
+ errorMessage
+ )
+
+ // The group is still a classic group.
+ assertEquals(
+ List(
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId(groupId)
+ .setProtocolType("consumer")
+ .setGroupState(ClassicGroupState.STABLE.toString)
+ .setGroupType(Group.GroupType.CLASSIC.toString)
+ ),
+ listGroups(
+ statesFilter = List.empty,
+ typesFilter = List(Group.GroupType.CLASSIC.toString)
+ )
+ )
+
+ // Classic member 1 leaves the group.
+ leaveGroup(
+ groupId = groupId,
+ memberId = memberId1,
+ useNewProtocol = false,
+ version = ApiKeys.LEAVE_GROUP.latestVersion(isUnstableApiEnabled)
+ )
+
+ // Verify that the group is empty.
+ assertEquals(
+ List(
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId(groupId)
+ .setProtocolType("consumer")
+ .setGroupState(ClassicGroupState.EMPTY.toString)
+ .setGroupType(Group.GroupType.CLASSIC.toString)
+ ),
+ listGroups(
+ statesFilter = List.empty,
+ typesFilter = List(Group.GroupType.CLASSIC.toString)
+ )
+ )
+
+ // The joining request with a consumer group member is accepted.
+ consumerGroupHeartbeat(
+ groupId = groupId,
+ memberId = Uuid.randomUuid.toString,
+ rebalanceTimeoutMs = 5 * 60 * 1000,
+ subscribedTopicNames = List("foo"),
+ topicPartitions = List.empty,
+ expectedError = Errors.NONE
+ )
+
+ // The group has become a consumer group.
+ assertEquals(
+ List(
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId(groupId)
+ .setProtocolType("consumer")
+ .setGroupState(ConsumerGroupState.STABLE.toString)
+ .setGroupType(Group.GroupType.CONSUMER.toString)
+ ),
+ listGroups(
+ statesFilter = List.empty,
+ typesFilter = List(Group.GroupType.CONSUMER.toString)
+ )
+ )
+ }
+
private def testUpgradeFromEmptyClassicToConsumerGroup(): Unit = {
// Creates the __consumer_offsets topics because it won't be created automatically
// in this test because it does not use FindCoordinator API.
@@ -1262,10 +1377,11 @@ class ConsumerProtocolMigrationTest(cluster: ClusterInstance) extends GroupCoord
).array
}
- private def assignment(assignedPartitions: List[Int]): Array[Byte] = {
+ private def assignment(assignedPartitions: List[Int], userData: ByteBuffer = null): Array[Byte] = {
ConsumerProtocol.serializeAssignment(
new ConsumerPartitionAssignor.Assignment(
- assignedPartitions.map(new TopicPartition("foo", _)).asJava
+ assignedPartitions.map(new TopicPartition("foo", _)).asJava,
+ userData
)
).array
}
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
index 90d8db4d320..dd0c6954088 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
@@ -34,6 +34,7 @@ import org.apache.kafka.common.errors.UnknownMemberIdException;
import org.apache.kafka.common.errors.UnknownServerException;
import org.apache.kafka.common.errors.UnreleasedInstanceIdException;
import org.apache.kafka.common.errors.UnsupportedAssignorException;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
import org.apache.kafka.common.message.ConsumerGroupDescribeResponseData;
import org.apache.kafka.common.message.ConsumerGroupHeartbeatRequestData;
import org.apache.kafka.common.message.ConsumerGroupHeartbeatResponseData;
@@ -186,6 +187,7 @@ import static org.apache.kafka.coordinator.group.classic.ClassicGroupState.PREPA
import static org.apache.kafka.coordinator.group.classic.ClassicGroupState.STABLE;
import static org.apache.kafka.coordinator.group.metrics.GroupCoordinatorMetrics.CLASSIC_GROUP_COMPLETED_REBALANCES_SENSOR_NAME;
import static org.apache.kafka.coordinator.group.metrics.GroupCoordinatorMetrics.CONSUMER_GROUP_REBALANCES_SENSOR_NAME;
+import static org.apache.kafka.coordinator.group.modern.ModernGroupMember.hasAssignedPartitionsChanged;
import static org.apache.kafka.coordinator.group.modern.consumer.ConsumerGroupMember.hasAssignedPartitionsChanged;
/**
@@ -674,7 +676,8 @@ public class GroupMetadataManager {
} else {
if (group.type() == CONSUMER) {
return (ConsumerGroup) group;
- } else if (createIfNotExists && group.type() == CLASSIC && validateOnlineUpgrade((ClassicGroup) group)) {
+ } else if (createIfNotExists && group.type() == CLASSIC) {
+ validateOnlineUpgrade((ClassicGroup) group);
return convertToConsumerGroup((ClassicGroup) group, records);
} else {
throw new GroupIdNotFoundException(String.format("Group %s is not a consumer group.", groupId));
@@ -1033,23 +1036,28 @@ public class GroupMetadataManager {
* Validates the online upgrade if the Classic Group receives a ConsumerGroupHeartbeat request.
*
* @param classicGroup A ClassicGroup.
- * @return A boolean indicating whether it's valid to online upgrade the classic group.
+ * @throws GroupIdNotFoundException if the group cannot be upgraded.
*/
- private boolean validateOnlineUpgrade(ClassicGroup classicGroup) {
+ private void validateOnlineUpgrade(ClassicGroup classicGroup) {
if (!config.consumerGroupMigrationPolicy().isUpgradeEnabled()) {
- log.info("Cannot upgrade classic group {} to consumer group because the online upgrade is disabled.",
+ log.info("Cannot upgrade classic group {} to consumer group because online upgrade is disabled.",
classicGroup.groupId());
- return false;
+ throw new GroupIdNotFoundException(
+ String.format("Cannot upgrade classic group %s to consumer group because online upgrade is disabled.", classicGroup.groupId())
+ );
} else if (!classicGroup.usesConsumerGroupProtocol()) {
log.info("Cannot upgrade classic group {} to consumer group because the group does not use the consumer embedded protocol.",
classicGroup.groupId());
- return false;
+ throw new GroupIdNotFoundException(
+ String.format("Cannot upgrade classic group %s to consumer group because the group does not use the consumer embedded protocol.", classicGroup.groupId())
+ );
} else if (classicGroup.numMembers() > config.consumerGroupMaxSize()) {
log.info("Cannot upgrade classic group {} to consumer group because the group size exceeds the consumer group maximum size.",
classicGroup.groupId());
- return false;
+ throw new GroupIdNotFoundException(
+ String.format("Cannot upgrade classic group %s to consumer group because the group size exceeds the consumer group maximum size.", classicGroup.groupId())
+ );
}
- return true;
}
/**
@@ -1078,12 +1086,21 @@ public class GroupMetadataManager {
metadataImage.topics()
);
} catch (SchemaException e) {
- log.warn("Cannot upgrade the classic group " + classicGroup.groupId() +
+ log.warn("Cannot upgrade classic group " + classicGroup.groupId() +
" to consumer group because the embedded consumer protocol is malformed: "
+ e.getMessage() + ".", e);
- throw new GroupIdNotFoundException("Cannot upgrade the classic group " + classicGroup.groupId() +
- " to consumer group because the embedded consumer protocol is malformed.");
+ throw new GroupIdNotFoundException(
+ String.format("Cannot upgrade classic group %s to consumer group because the embedded consumer protocol is malformed.", classicGroup.groupId())
+ );
+ } catch (UnsupportedVersionException e) {
+ log.warn("Cannot upgrade classic group " + classicGroup.groupId() +
+ " to consumer group: " + e.getMessage() + ".", e);
+
+ throw new GroupIdNotFoundException(
+ String.format("Cannot upgrade classic group %s to consumer group because an unsupported custom assignor is in use. " +
+ "Please refer to the documentation or switch to a default assignor before re-attempting the upgrade.", classicGroup.groupId())
+ );
}
consumerGroup.createConsumerGroupRecords(records);
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroup.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroup.java
index fbd259dc332..f8c7b250903 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroup.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroup.java
@@ -24,8 +24,10 @@ import org.apache.kafka.common.errors.StaleMemberEpochException;
import org.apache.kafka.common.errors.UnknownMemberIdException;
import org.apache.kafka.common.errors.UnsupportedVersionException;
import org.apache.kafka.common.message.ConsumerGroupDescribeResponseData;
+import org.apache.kafka.common.message.ConsumerProtocolAssignment;
import org.apache.kafka.common.message.ConsumerProtocolSubscription;
import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.protocol.types.SchemaException;
import org.apache.kafka.common.requests.JoinGroupRequest;
import org.apache.kafka.coordinator.common.runtime.CoordinatorRecord;
import org.apache.kafka.coordinator.group.GroupCoordinatorRecordHelpers;
@@ -1131,6 +1133,9 @@ public class ConsumerGroup extends ModernGroup<ConsumerGroupMember> {
* @param classicGroup The converted classic group.
* @param topicsImage The TopicsImage for topic id and topic name conversion.
* @return The created ConsumerGroup.
+ *
+ * @throws SchemaException if any member's subscription or assignment cannot be deserialized.
+ * @throws UnsupportedVersionException if userData from a custom assignor would be lost.
*/
public static ConsumerGroup fromClassicGroup(
SnapshotRegistry snapshotRegistry,
@@ -1150,12 +1155,13 @@ public class ConsumerGroup extends ModernGroup<ConsumerGroupMember> {
if (Arrays.equals(classicGroupMember.assignment(), EMPTY_ASSIGNMENT)) {
assignedPartitions = Collections.emptyMap();
} else {
- assignedPartitions = toTopicPartitionMap(
- ConsumerProtocol.deserializeConsumerProtocolAssignment(
- ByteBuffer.wrap(classicGroupMember.assignment())
- ),
- topicsImage
+ ConsumerProtocolAssignment assignment = ConsumerProtocol.deserializeConsumerProtocolAssignment(
+ ByteBuffer.wrap(classicGroupMember.assignment())
);
+ if (assignment.userData() != null && assignment.userData().hasRemaining()) {
+ throw new UnsupportedVersionException("userData from a custom assignor would be lost");
+ }
+ assignedPartitions = toTopicPartitionMap(assignment, topicsImage);
}
// Every member is guaranteed to have metadata set when it joins,
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
index b92ce049c3e..18540253487 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
@@ -45,6 +45,8 @@ import org.apache.kafka.common.message.DescribeGroupsResponseData;
import org.apache.kafka.common.message.HeartbeatRequestData;
import org.apache.kafka.common.message.HeartbeatResponseData;
import org.apache.kafka.common.message.JoinGroupRequestData;
+import org.apache.kafka.common.message.JoinGroupRequestData.JoinGroupRequestProtocol;
+import org.apache.kafka.common.message.JoinGroupRequestData.JoinGroupRequestProtocolCollection;
import org.apache.kafka.common.message.JoinGroupResponseData;
import org.apache.kafka.common.message.JoinGroupResponseData.JoinGroupResponseMember;
import org.apache.kafka.common.message.LeaveGroupRequestData;
@@ -96,8 +98,11 @@ import org.apache.kafka.server.common.MetadataVersion;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
import org.junit.jupiter.params.provider.ValueSource;
+import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
@@ -10112,6 +10117,116 @@ public class GroupMetadataManagerTest {
assertEquals(group, context.groupMetadataManager.getOrMaybeCreateClassicGroup("group-id", false));
}
+ /**
+ * Supplies the {@link Arguments} to {@link #testConsumerGroupHeartbeatWithCustomAssignorClassicGroup(ByteBuffer, boolean)}.
+ */
+ private static Stream<Arguments> testConsumerGroupHeartbeatWithCustomAssignorClassicGroupSource() {
+ return Stream.of(
+ Arguments.of(null, true),
+ Arguments.of(ByteBuffer.allocate(0), true),
+ Arguments.of(ByteBuffer.allocate(1), false)
+ );
+ }
+
+ @ParameterizedTest
+ @MethodSource("testConsumerGroupHeartbeatWithCustomAssignorClassicGroupSource")
+ public void testConsumerGroupHeartbeatWithCustomAssignorClassicGroup(ByteBuffer userData, boolean expectUpgrade) {
+ String groupId = "group-id";
+ String memberId1 = "member-id-1";
+ String memberId2 = "member-id-2";
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+ Uuid barTopicId = Uuid.randomUuid();
+ String barTopicName = "bar";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+ assignor.prepareGroupAssignment(new GroupAssignment(Map.of(
+ memberId1, new MemberAssignmentImpl(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0)
+ )),
+ memberId2, new MemberAssignmentImpl(mkAssignment(
+ mkTopicAssignment(barTopicId, 0)
+ ))
+ )));
+
+ MetadataImage metadataImage = new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 1)
+ .addTopic(barTopicId, barTopicName, 1)
+ .addRacks()
+ .build();
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_MIGRATION_POLICY_CONFIG, ConsumerGroupMigrationPolicy.UPGRADE.toString())
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
+ .withMetadataImage(metadataImage)
+ .build();
+
+ JoinGroupRequestData.JoinGroupRequestProtocolCollection protocols = new JoinGroupRequestData.JoinGroupRequestProtocolCollection(1);
+ protocols.add(new JoinGroupRequestData.JoinGroupRequestProtocol()
+ .setName("range")
+ .setMetadata(Utils.toArray(ConsumerProtocol.serializeSubscription(new ConsumerPartitionAssignor.Subscription(
+ List.of(fooTopicName, barTopicName),
+ null,
+ List.of(
+ new TopicPartition(fooTopicName, 0),
+ new TopicPartition(barTopicName, 0)
+ )
+ ))))
+ );
+
+ Map<String, byte[]> assignments = Map.of(
+ memberId1,
+ Utils.toArray(ConsumerProtocol.serializeAssignment(new ConsumerPartitionAssignor.Assignment(List.of(
+ new TopicPartition(fooTopicName, 0),
+ new TopicPartition(barTopicName, 0)
+ ), userData)))
+ );
+
+ // Create a stable classic group with member 1.
+ ClassicGroup group = context.createClassicGroup(groupId);
+ group.setProtocolName(Optional.of("range"));
+ group.add(
+ new ClassicGroupMember(
+ memberId1,
+ Optional.empty(),
+ "client-id",
+ "client-host",
+ 10000,
+ 5000,
+ "consumer",
+ protocols,
+ assignments.get(memberId1)
+ )
+ );
+
+ group.transitionTo(PREPARING_REBALANCE);
+ group.transitionTo(COMPLETING_REBALANCE);
+ group.transitionTo(STABLE);
+
+ context.replay(GroupCoordinatorRecordHelpers.newGroupMetadataRecord(group, assignments, metadataImage.features().metadataVersion()));
+ context.commit();
+ group = context.groupMetadataManager.getOrMaybeCreateClassicGroup(groupId, false);
+
+ // A new member 2 with new protocol joins the classic group, triggering the upgrade.
+ ConsumerGroupHeartbeatRequestData consumerGroupHeartbeatRequestData =
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId2)
+ .setRebalanceTimeoutMs(5000)
+ .setServerAssignor("range")
+ .setSubscribedTopicNames(List.of(fooTopicName, barTopicName))
+ .setTopicPartitions(Collections.emptyList());
+
+ if (expectUpgrade) {
+ context.consumerGroupHeartbeat(consumerGroupHeartbeatRequestData);
+ } else {
+ Exception ex = assertThrows(GroupIdNotFoundException.class, () -> context.consumerGroupHeartbeat(consumerGroupHeartbeatRequestData));
+ assertEquals(
+ "Cannot upgrade classic group group-id to consumer group because an unsupported custom assignor is in use. " +
+ "Please refer to the documentation or switch to a default assignor before re-attempting the upgrade.", ex.getMessage());
+ }
+ }
+
@Test
public void testConsumerGroupHeartbeatToClassicGroupFromExistingStaticMember() {
String groupId = "group-id";