KAFKA-20183: Share consumer group fails when group ID contains colon character

[fvaleri]

This patch rejects share consumer group IDs containing ':' to prevent
SharePartitionKey parsing failure.

The SharePartitionKey format (groupId:topicId:partition) breaks when
the group ID itself contains ':', causing silent consumption failure. We
apply a minimal targeted fix (colon-only restriction) to avoid breaking
existing group naming conventions. This is validated on both client and
broker to cover old clients without the check.

EDIT:

The original approach restricted group IDs containing colon, which would
require a KIP. Instead, we now fix the issue in SharePartitionKey
parsing the key from the right since partition (int) and topicId (base64
UUID) have known formats and are always the last two segments, while
groupId (everything before) may contain colons.

Reviewers: Mickael Maison mickael.maison@gmail.com, Andrew Schofield
aschofield@confluent.io, Sushant Mahajan smahajan@confluent.io

[showuon]

@fvaleri , please help add tests. Thanks.

[smjn]

@fvaleri While the problem exists, there is no need to restrict the groupId name. The fix should be in SharePartitionKey (in both cases if you plan to use the restriction or fix the split logic).

[AndrewJSchofield]

This seems like the most practical fix to me. Topic IDs already cannot contain : so the problem is restricted to group IDs only.

[smjn]

Thanks for the PR, however restriction to group name is not required.

[mimaison]

While this is the easy fix, I'm not sure it's the best option. Changing the allowed characters in a group name would typically require a KIP. Can't we adjust the split logic to start from the end? The topic name cannot contain a colon.

[smjn]

The change should be in https://github.com/apache/kafka/blob/trunk/server-common/src/main/java/org/apache/kafka/server/share/SharePartitionKey.java#L75

something like:

public static SharePartitionKey getInstance(String key) {
        validate(key);
        String[] tokens = key.split(":");
        int last =  tokens.length - 1;
        //String groupId = String.join(":", Arrays.copyOfRange(tokens, 0, last-1));
       // OR
      // String groupId = key.substring(0, key.lastIndexOf(":", key.lastIndexOf(":")-1));
        return new SharePartitionKey(
                groupId,
                Uuid.fromString(tokens[last-1]),
                Integer.parseInt(tokens[last])
        );
    }

as well as update in validate in same class

[mimaison]

We need to update validate() as well as it checks for exactly 3 tokens and then does further checks based on this wrong assumption

[fvaleri]

Right. Let me try this approach. Thanks.

[smjn]

Thanks for the changes, LGTM

[AndrewJSchofield]

Thanks for the PR. Looks good to me.

[mimaison]

LGTM