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 2026/06/24 08:55:05 UTC
(kafka) branch 4.3 updated: KAFKA-20570: Catch RuntimeException in ConsumerProtocol deserialization in group coordinator (#22264)
This is an automated email from the ASF dual-hosted git repository.
dajac pushed a commit to branch 4.3
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/4.3 by this push:
new e7070ec3a0f KAFKA-20570: Catch RuntimeException in ConsumerProtocol deserialization in group coordinator (#22264)
e7070ec3a0f is described below
commit e7070ec3a0ffbc67dd9ff2b0f14f7f7abd042c1d
Author: Dongnuo Lyu <13...@users.noreply.github.com>
AuthorDate: Thu May 21 11:40:32 2026 -0400
KAFKA-20570: Catch RuntimeException in ConsumerProtocol deserialization in group coordinator (#22264)
The `deserialize*` methods in `ConsumerProtocol` only catch
`BufferUnderflowException` and re-wrap it as `SchemaException`.
Malformed bytes can also surface as other `RuntimeException`s (e.g.
`IllegalArgumentException` from negative array lengths in
`ByteBufferAccessor`), which escape these methods. Callers that only
guard against `SchemaException` then propagate the failure, which can
destabilize the group coordinator.
### Changes
In `ConsumerProtocol`, broaden the `catch` clause from
`BufferUnderflowException` to `RuntimeException` in all four
deserialization entry points, re-wrapping as `SchemaException`:
- `deserializeSubscription`
- `deserializeConsumerProtocolSubscription`
- `deserializeAssignment`
- `deserializeConsumerProtocolAssignment`
Reviewers: Sean Quah <sq...@confluent.io>, David Jacot
<da...@gmail.com>
---
.../consumer/internals/ConsumerProtocol.java | 16 +--
.../consumer/internals/ConsumerProtocolTest.java | 85 +++++++++++++
.../group/GroupMetadataManagerTest.java | 140 +++++++++++++++++++++
.../group/classic/ClassicGroupTest.java | 31 +++++
4 files changed, 264 insertions(+), 8 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerProtocol.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerProtocol.java
index aa9d0cc40e4..b334824e8e3 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerProtocol.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerProtocol.java
@@ -116,8 +116,8 @@ public class ConsumerProtocol {
ownedPartitions,
data.generationId(),
data.rackId() == null || data.rackId().isEmpty() ? Optional.empty() : Optional.of(data.rackId()));
- } catch (BufferUnderflowException e) {
- throw new SchemaException("Buffer underflow while parsing consumer protocol's subscription", e);
+ } catch (RuntimeException e) {
+ throw new SchemaException("Malformed consumer protocol subscription", e);
}
}
@@ -133,8 +133,8 @@ public class ConsumerProtocol {
try {
return new ConsumerProtocolSubscription(new ByteBufferAccessor(buffer), version);
- } catch (BufferUnderflowException e) {
- throw new SchemaException("Buffer underflow while parsing consumer protocol's subscription", e);
+ } catch (RuntimeException e) {
+ throw new SchemaException("Malformed consumer protocol subscription", e);
}
}
@@ -186,8 +186,8 @@ public class ConsumerProtocol {
return new Assignment(
assignedPartitions,
data.userData() != null ? data.userData().duplicate() : null);
- } catch (BufferUnderflowException e) {
- throw new SchemaException("Buffer underflow while parsing consumer protocol's assignment", e);
+ } catch (RuntimeException e) {
+ throw new SchemaException("Malformed consumer protocol assignment", e);
}
}
@@ -203,8 +203,8 @@ public class ConsumerProtocol {
try {
return new ConsumerProtocolAssignment(new ByteBufferAccessor(buffer), version);
- } catch (BufferUnderflowException e) {
- throw new SchemaException("Buffer underflow while parsing consumer protocol's assignment", e);
+ } catch (RuntimeException e) {
+ throw new SchemaException("Malformed consumer protocol assignment", e);
}
}
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerProtocolTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerProtocolTest.java
index b73576757e8..4d516d78c0c 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerProtocolTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerProtocolTest.java
@@ -24,6 +24,7 @@ import org.apache.kafka.common.message.ConsumerProtocolSubscription;
import org.apache.kafka.common.protocol.types.ArrayOf;
import org.apache.kafka.common.protocol.types.Field;
import org.apache.kafka.common.protocol.types.Schema;
+import org.apache.kafka.common.protocol.types.SchemaException;
import org.apache.kafka.common.protocol.types.Struct;
import org.apache.kafka.common.protocol.types.Type;
@@ -39,9 +40,11 @@ import java.util.Optional;
import java.util.Set;
import static org.apache.kafka.clients.consumer.internals.AbstractStickyAssignor.DEFAULT_GENERATION;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
public class ConsumerProtocolTest {
@@ -356,4 +359,86 @@ public class ConsumerProtocolTest {
return buffer;
}
+
+ @Test
+ public void deserializeSubscriptionThrowsSchemaExceptionForEveryTruncation() {
+ Subscription subscription = new Subscription(
+ Arrays.asList("foo", "bar"),
+ ByteBuffer.wrap(new byte[]{0x01, 0x02}),
+ Arrays.asList(new TopicPartition("foo", 0), new TopicPartition("bar", 1)),
+ DEFAULT_GENERATION,
+ Optional.of("rack"));
+ ByteBuffer serialized = ConsumerProtocol.serializeSubscription(subscription,
+ ConsumerProtocolSubscription.HIGHEST_SUPPORTED_VERSION);
+ byte[] serializedBytes = new byte[serialized.remaining()];
+ serialized.duplicate().get(serializedBytes);
+
+ for (int len = 0; len < serializedBytes.length; len++) {
+ byte[] truncated = Arrays.copyOf(serializedBytes, len);
+ assertThrows(SchemaException.class,
+ () -> ConsumerProtocol.deserializeSubscription(ByteBuffer.wrap(truncated)),
+ "Expected SchemaException for subscription truncated to length " + len);
+ }
+ assertDoesNotThrow(() -> ConsumerProtocol.deserializeSubscription(ByteBuffer.wrap(serializedBytes)));
+ }
+
+ @Test
+ public void deserializeAssignmentThrowsSchemaExceptionForEveryTruncation() {
+ Assignment assignment = new Assignment(
+ Arrays.asList(new TopicPartition("foo", 0), new TopicPartition("bar", 1)),
+ ByteBuffer.wrap(new byte[]{0x01, 0x02}));
+ ByteBuffer serialized = ConsumerProtocol.serializeAssignment(assignment,
+ ConsumerProtocolAssignment.HIGHEST_SUPPORTED_VERSION);
+ byte[] serializedBytes = new byte[serialized.remaining()];
+ serialized.duplicate().get(serializedBytes);
+
+ for (int len = 0; len < serializedBytes.length; len++) {
+ byte[] truncated = Arrays.copyOf(serializedBytes, len);
+ assertThrows(SchemaException.class,
+ () -> ConsumerProtocol.deserializeAssignment(ByteBuffer.wrap(truncated)),
+ "Expected SchemaException for assignment truncated to length " + len);
+ }
+ assertDoesNotThrow(() -> ConsumerProtocol.deserializeAssignment(ByteBuffer.wrap(serializedBytes)));
+ }
+
+ @Test
+ public void deserializeConsumerProtocolSubscriptionThrowsSchemaExceptionForEveryTruncation() {
+ Subscription subscription = new Subscription(
+ Arrays.asList("foo", "bar"),
+ ByteBuffer.wrap(new byte[]{0x01, 0x02}),
+ Arrays.asList(new TopicPartition("foo", 0), new TopicPartition("bar", 1)),
+ DEFAULT_GENERATION,
+ Optional.of("rack"));
+ ByteBuffer serialized = ConsumerProtocol.serializeSubscription(subscription,
+ ConsumerProtocolSubscription.HIGHEST_SUPPORTED_VERSION);
+ byte[] serializedBytes = new byte[serialized.remaining()];
+ serialized.duplicate().get(serializedBytes);
+
+ for (int len = 0; len < serializedBytes.length; len++) {
+ byte[] truncated = Arrays.copyOf(serializedBytes, len);
+ assertThrows(SchemaException.class,
+ () -> ConsumerProtocol.deserializeConsumerProtocolSubscription(ByteBuffer.wrap(truncated)),
+ "Expected SchemaException for ConsumerProtocolSubscription truncated to length " + len);
+ }
+ assertDoesNotThrow(() -> ConsumerProtocol.deserializeConsumerProtocolSubscription(ByteBuffer.wrap(serializedBytes)));
+ }
+
+ @Test
+ public void deserializeConsumerProtocolAssignmentThrowsSchemaExceptionForEveryTruncation() {
+ Assignment assignment = new Assignment(
+ Arrays.asList(new TopicPartition("foo", 0), new TopicPartition("bar", 1)),
+ ByteBuffer.wrap(new byte[]{0x01, 0x02}));
+ ByteBuffer serialized = ConsumerProtocol.serializeAssignment(assignment,
+ ConsumerProtocolAssignment.HIGHEST_SUPPORTED_VERSION);
+ byte[] serializedBytes = new byte[serialized.remaining()];
+ serialized.duplicate().get(serializedBytes);
+
+ for (int len = 0; len < serializedBytes.length; len++) {
+ byte[] truncated = Arrays.copyOf(serializedBytes, len);
+ assertThrows(SchemaException.class,
+ () -> ConsumerProtocol.deserializeConsumerProtocolAssignment(ByteBuffer.wrap(truncated)),
+ "Expected SchemaException for ConsumerProtocolAssignment truncated to length " + len);
+ }
+ assertDoesNotThrow(() -> ConsumerProtocol.deserializeConsumerProtocolAssignment(ByteBuffer.wrap(serializedBytes)));
+ }
}
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 99e22bbeac5..dc83f47a31b 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
@@ -12163,6 +12163,146 @@ public class GroupMetadataManagerTest {
}
}
+ @Test
+ public void testConsumerGroupHeartbeatWithStableClassicGroupFailsOnMalformedProtocol() {
+ String groupId = "group-id";
+ String memberId1 = "member-id-1";
+ String memberId2 = "member-id-2";
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+
+ CoordinatorMetadataImage metadataImage = new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 1)
+ .addRacks()
+ .buildCoordinatorMetadataImage();
+
+ 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();
+
+ // Throws RuntimeException when read
+ byte[] poisonMetadata = new byte[]{
+ 0, 1, // version (int16) = 1
+ (byte) 0xFF, (byte) 0xFF, (byte) 0xFF, (byte) 0xFF // topics array length (int32) = -1
+ };
+
+ JoinGroupRequestData.JoinGroupRequestProtocolCollection protocols = new JoinGroupRequestData.JoinGroupRequestProtocolCollection(1);
+ protocols.add(new JoinGroupRequestData.JoinGroupRequestProtocol()
+ .setName("range")
+ .setMetadata(poisonMetadata));
+
+ Map<String, byte[]> assignments = Map.of(
+ memberId1,
+ Utils.toArray(ConsumerProtocol.serializeAssignment(
+ new ConsumerPartitionAssignor.Assignment(List.of(new TopicPartition(fooTopicName, 0)))))
+ );
+
+ 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));
+ context.commit();
+
+ // A new member 2 with the new protocol joins the classic group, triggering the upgrade.
+ ConsumerGroupHeartbeatRequestData request = new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId2)
+ .setRebalanceTimeoutMs(5000)
+ .setServerAssignor("range")
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setTopicPartitions(List.of());
+
+ Exception ex = assertThrows(GroupIdNotFoundException.class,
+ () -> context.consumerGroupHeartbeat(request));
+ assertEquals(
+ "Cannot upgrade classic group group-id to consumer group because the embedded consumer protocol is malformed.",
+ ex.getMessage()
+ );
+ }
+
+ @Test
+ public void testClassicGroupJoinToConsumerGroupFailsOnMalformedSubscriptionMetadata() {
+ String groupId = "group-id";
+ String existingMemberId = Uuid.randomUuid().toString();
+ String newMemberId = Uuid.randomUuid().toString();
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+
+ CoordinatorMetadataImage metadataImage = new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 1)
+ .addRacks()
+ .buildCoordinatorMetadataImage();
+
+ ConsumerGroupMember existingMember = new ConsumerGroupMember.Builder(existingMemberId)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId(DEFAULT_CLIENT_ID)
+ .setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setAssignedPartitions(toAssignmentWithEpochs(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0)), 10))
+ .build();
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_MIGRATION_POLICY_CONFIG, ConsumerGroupMigrationPolicy.UPGRADE.toString())
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(new MockPartitionAssignor("range")))
+ .withMetadataImage(metadataImage)
+ .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
+ .withMember(existingMember)
+ .withAssignment(existingMemberId, mkAssignment(mkTopicAssignment(fooTopicId, 0)))
+ .withAssignmentEpoch(10)
+ .withMetadataHash(computeGroupHash(Map.of(
+ fooTopicName, computeTopicHash(fooTopicName, metadataImage)
+ ))))
+ .build();
+
+ // Throws RuntimeException when read.
+ byte[] poisonMetadata = new byte[]{
+ 0, 1, // version (int16) = 1
+ (byte) 0xFF, (byte) 0xFF, (byte) 0xFF, (byte) 0xFF // topics array length (int32) = -1
+ };
+ JoinGroupRequestData.JoinGroupRequestProtocolCollection protocols = new JoinGroupRequestData.JoinGroupRequestProtocolCollection(1);
+ protocols.add(new JoinGroupRequestData.JoinGroupRequestProtocol()
+ .setName("range")
+ .setMetadata(poisonMetadata));
+
+ JoinGroupRequestData joinRequest = new JoinGroupRequestData()
+ .setGroupId(groupId)
+ .setMemberId(newMemberId)
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ .setProtocols(protocols)
+ .setSessionTimeoutMs(5000)
+ .setRebalanceTimeoutMs(45000);
+
+ IllegalStateException ex = assertThrows(IllegalStateException.class,
+ () -> context.sendClassicGroupJoin(joinRequest));
+ assertEquals("Malformed embedded consumer protocol in subscription deserialization.", ex.getMessage());
+ }
+
@Test
public void testConsumerGroupHeartbeatToClassicGroupFromExistingStaticMember() {
String groupId = "group-id";
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/classic/ClassicGroupTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/classic/ClassicGroupTest.java
index 5c096a11b1d..6d3416f59f5 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/classic/ClassicGroupTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/classic/ClassicGroupTest.java
@@ -1264,6 +1264,37 @@ public class ClassicGroupTest {
assertTrue(group.isSubscribedToTopic("topic"));
}
+ @Test
+ public void testComputeSubscribedTopicsHandlesMalformedMemberMetadata() {
+ ClassicGroup group = new ClassicGroup(logContext, "groupId", EMPTY, Time.SYSTEM);
+
+ JoinGroupRequestProtocolCollection protocols = new JoinGroupRequestProtocolCollection();
+ protocols.add(new JoinGroupRequestProtocol()
+ .setName("range")
+ .setMetadata(new byte[]{
+ 0, 1, // version (int16) = 1
+ (byte) 0xFF, (byte) 0xFF, (byte) 0xFF, (byte) 0xFF // topics array length (int32) = -1
+ }));
+
+ ClassicGroupMember poisonMember = new ClassicGroupMember(
+ "poisonMember",
+ Optional.empty(),
+ clientId,
+ clientHost,
+ rebalanceTimeoutMs,
+ sessionTimeoutMs,
+ "consumer",
+ protocols
+ );
+
+ group.add(poisonMember);
+ group.transitionTo(PREPARING_REBALANCE);
+ group.initNextGeneration();
+
+ // RuntimeException should not propagate; falls through to Optional.empty().
+ assertEquals(Optional.empty(), group.computeSubscribedTopics());
+ }
+
@Test
public void testIsInStates() {
ClassicGroup group = new ClassicGroup(new LogContext(), "groupId", EMPTY, Time.SYSTEM);