KAFKA-21302: Sort copartition groups and indexes for consistent Streams topology comparison by chickenchickenlove · Pull Request #23855 · apache/kafka

chickenchickenlove · 2026-10-18T03:21:12Z

With group.protocol=streams, a member with identical copartition
constraints can be rejected because of differences in source topic
ordering. For example, constructing the same join with source topics
["A", "B"] and ["B", "A"] can produce heartbeat copartition indexes
of [0, 1, 2] and [1, 0, 2], respectively.

The coordinator compares these lists in an order-sensitive manner and
treats the second member’s request as a topology change. It returns
INVALID_REQUEST, causing the client to transition to FATAL and
preventing an otherwise valid instance from joining or replacing an
existing member

Reviewers: Lucas Brutschy lbrutschy@confluent.io

…ms topology comparison

Copilot

🟢 Approval recommended

Normalization preserves index targets and inputs, with focused regression coverage and no unresolved findings.

No open issues 🎉

What changed in this PR

Makes Streams topology comparison insensitive to copartition group and index ordering, preventing rejection of members with equivalent constraints.

Changes:

Normalizes copartition constraints during record replay and heartbeat processing without changing referenced topic lists.
Adds regression coverage for equality, input preservation, and successful member joins.

File	Description
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsTopologyTest.java	Tests normalization, equality, and input preservation.
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java	Verifies joins with reordered copartition constraints succeed.
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsTopology.java	Normalizes copartition groups and indexes across both ingestion paths.

_{Review effort: Balanced}

lucasbru · 2026-10-20T09:33:23Z

+                .setCopartitionGroups(List.of(
+                    copartitionGroup(topicType, List.of((short) 0)),
+                    copartitionGroup(topicType, List.of((short) 0, (short) 1)),
+                    copartitionGroup(topicType, List.of((short) 2, (short) 3))


Could we also cover a group that mixes source and repartition indexes, e.g. {source: [1], repartition: [0]}? That's the typical join case, and it's where the thenComparing tie-breaks actually matter.

lucasbru · 2026-10-20T09:33:24Z

+                .setWarmupTasks(List.of()));
+
+        assertEquals(Errors.NONE.code(), result.response().data().errorCode());
+        assertTrue(result.response().data().memberEpoch() > 0);


Maybe also assert that result.records() contains no new topology record, so we know the request was matched against the existing topology?

lucasbru · 2026-10-20T09:37:53Z

+        // The first step sorts indexes within each group; the second sorts the groups.
+        // The referenced topic lists stay unchanged, so each index still identifies the same topic.
+        // Copy before sorting to avoid mutating a replayed record or the heartbeat request.
+        Subtopology copy = subtopology.duplicate();


This deep-copies the whole subtopology, including all the TopicInfos and their configs, just to sort the copartition groups, and fromHeartbeatRequest now copies twice (convert + duplicate). Could we only copy the copartition groups? If we sort in convertToStreamsGroupTopologyRecord as suggested in the review body, the heartbeat path doesn't need a copy at all.

Also, convertToTopicInfo maps a null topicConfigs to null, and duplicate() would NPE on that. A deserialized request can't have null there since the field isn't nullable, so maybe that null branch is just dead code. But it's another reason not to duplicate fields we don't touch.

Thinking about this more: since the join heartbeat can be hot on large clusters (e.g. during rejoin storms), could we avoid building a StreamsTopology from the request at all when the group already has one? maybeUpdateTopology already converts the full request on every topology-carrying heartbeat, and with the normalization we copy it a second time.

One option: only call fromHeartbeatRequest when initializing the topology, and otherwise compare the request directly against the stored topology with something like StreamsTopology#matches(StreamsGroupHeartbeatRequestData.Topology), matching copartition groups regardless of order. The stored topology is already normalized in fromRecord, and the groups are tiny, so this can be done without allocating. It would need a test that fails when a field is added to Subtopology, so the comparison doesn't silently miss it. I'd rather not override equals for this, since hashCode would also need to be made order-independent.

What do you think?

@lucasbru

Thank you for the thoughtful suggestion. I agree that, considering cases such as rejoin storms, optimizing this path with matches() is important for the broker.

Would you be open to first applying your initial suggestion to copy only the copartition groups, then landing and backporting the correctness fix before moving on to matches()? (I'm happy to take on the matches() optimization separately.)

My concern is that introducing a new comparison implementation covering the current schema could inadvertently introduce new correctness issues. With that in mind, I would prefer to keep this PR narrowly focused on handling the ordering differences between client requests and the broker’s stored topology, and complete the backports first(even client side!).

If you’re comfortable with that approach, could we address the matches() optimization in a separate ticket, with dedicated review and testing, including the schema field-coverage test you suggested?

I think matches() would actually replace the normalization rather than add to it: with an order-insensitive comparison we don't need to sort or copy the copartition groups at all, and fromRecord/fromHeartbeatRequest stay as they were. The current version also lists the Subtopology fields explicitly, so the risk of missing a field is the same either way and needs the same reflection test. So I'd lean towards doing it in this PR. What do you think?

Definitely agree.
I think this is an area where regressions could easily be introduced if new fields are added in the future.

lucasbru · 2026-10-20T09:37:54Z

+        return copy;
+    }
+
+    private static int compareTopicIndexes(List<Short> first, List<Short> second) {


nit: this is a hand-rolled lexicographic compare. Not a big deal, but since the lists are sorted at this point, would something like converting to short[] and using Arrays.compare be simpler? Fine to keep as is if you prefer.

Sorry, my nit made this slightly worse: toArray(Short[]::new) allocates two arrays on every comparison. The previous loop didn't allocate, so if we keep the normalization I'd go back to it.

lucasbru · 2026-10-20T09:40:56Z

Thanks for the fix! I think we should also fix this on the client. StreamsGroupHeartbeatRequestManager#getCopartitionGroupsFromStreams builds the index lists and the list of groups by iterating a Set, which is where the nondeterministic order comes from. Could you sort the indexes within each group and then the groups there too, the same way we already sort source/sink/repartition topics in fromStreamsToHeartbeatRequest? Clients would then always send a canonical topology. We'd still keep the broker-side normalization permanently, both on join and on replay, because of older clients and already-persisted unsorted records. A test in StreamsGroupHeartbeatRequestManagerTest that checks the sorted output would be good too.

It might also be worth sorting in convertToStreamsGroupTopologyRecord, so that newly written topology records are canonical too. The record format doesn't change, and it means a broker downgrade won't reject newer, sorted clients against a stored record. fromRecord would still normalize for groups created before this change.

chickenchickenlove · 2026-10-20T12:12:37Z

@lucasbru
Thanks for taking the time to review!

I agree that we should fix the client side as well. Given the backport implications, would it make sense to track that change in a separate ticket and PR?

AFAIK, Copartition groups are persisted, including their ordering. If an existing client is upgraded to send sorted groups while the broker remains unpatched, the broker could reject the topology because its ordering differs from the persisted record, even though the topology is semantically unchanged.

Separating the client-side change would let us backport the broker normalization independently and evaluate the client-side backport with this compatibility concern in mind.

What do you think?

lucasbru · 2026-10-20T12:47:03Z

Good point, I hadn't thought about the unpatched-broker case. Agreed, let's keep this PR broker-only so it can be backported on its own, and handle the client side in a separate ticket.

The same argument applies to my suggestion of writing sorted records: if a patched broker writes sorted records and is later downgraded, existing clients sending the old order would be rejected. So let's drop that and keep persisting what the client sent.

chickenchickenlove · 2026-10-20T12:56:27Z

Thanks for the clarification!
First of all, I've created new ticket for client side. (https://issues.apache.org/jira/browse/KAFKA-21320)

Currently, I'm working on addressing your comments in this PR.
I'll be back ASAP with new commit!

lucasbru · 2026-10-20T13:00:25Z

Small addition after looking at how the client builds the groups: the order comes from HashSet iteration in InternalTopologyBuilder#copartitionGroups, which is deterministic for the same topology code. So instances running the same code always send the same order, and the mismatch only happens when equivalent topologies are written differently (e.g. join order swapped) and the topic names collide in the same hash bucket.

That also means existing records generally match what current clients send, which makes your point about sorting on the client even stronger. A sorting client would usually differ from the persisted order on an unpatched broker.

chickenchickenlove · 2026-10-20T13:55:53Z

@lucasbru

I've pushed a new commit that incorporates most of your suggestions!

I left out the matches() optimization for now, as I think it would be better to address it in a separate issue. However, if you consider it a minor change, I'm also happy to include it in this PR. Just let me know what you think!

I also added a comment at the end of KAFKA-21302. Even when running the exact same code, the ordering of copartition groups can vary across JVM runs if the source nodes are defined using Set.of(...).

I wanted to highlight this again, as I thought it might be helpful when determining the priority of this issue.

When you get a chance, please take another look! 🙇‍♂️

lucasbru · 2026-10-21T10:24:38Z

Thanks for the update and for checking the Set.of case, you are right. This can happen without code changes, in some cases (more detail in the ticket).

lucasbru · 2026-10-21T10:24:32Z

+        // The first step sorts indexes within each group; the second sorts the groups.
+        // The referenced topic lists stay unchanged, so each index still identifies the same topic.
+        // Copy before sorting to avoid mutating a replayed record or the heartbeat request.
+        Subtopology copy = subtopology.duplicate();


I think matches() would actually replace the normalization rather than add to it: with an order-insensitive comparison we don't need to sort or copy the copartition groups at all, and fromRecord/fromHeartbeatRequest stay as they were. The current version also lists the Subtopology fields explicitly, so the risk of missing a field is the same either way and needs the same reflection test. So I'd lean towards doing it in this PR. What do you think?

lucasbru · 2026-10-21T10:24:33Z

+        return copy;
+    }
+
+    private static int compareTopicIndexes(List<Short> first, List<Short> second) {


Sorry, my nit made this slightly worse: toArray(Short[]::new) allocates two arrays on every comparison. The previous loop didn't allocate, so if we keep the normalization I'd go back to it.

lucasbru · 2026-10-21T10:24:36Z

+                .thenComparing(CopartitionGroup::repartitionSourceTopics, StreamsTopology::compareTopicIndexes)
+        );
+
+        return new Subtopology()


Since this lists the fields explicitly, a field added to Subtopology later (it would have to be a tagged field) would be silently dropped here. Could we add a small test that counts the declared instance fields of StreamsGroupTopologyValue.Subtopology via reflection (8 today, including _unknownTaggedFields)? A schema field count wouldn't catch tagged fields. Or at least a comment saying this has to list every field. The same applies to matches() if we go that way.

Yes, you are right!
I've addressed it.

chickenchickenlove · 2026-10-21T12:15:28Z

@lucasbru
Thanks for taking the time to review this carefully!
I reverted change for the normalization and implemented matches method and some test cases.
When you get a chance, please take another look. 🙇‍♂️

KAFKA-21302: Sort copartition groups and indexes for consistent Strea…

f20def5

…ms topology comparison

github-actions Bot added triage PRs from the community group-coordinator labels Oct 18, 2026

lucasbru requested review from lucasbru and a balanced review from Copilot October 20, 2026 09:12

lucasbru self-assigned this Oct 20, 2026

Copilot started reviewing on behalf of lucasbru October 20, 2026 09:13

View session

Copilot AI reviewed Oct 20, 2026

View reviewed changes

lucasbru reviewed Oct 20, 2026

View reviewed changes

lucasbru added the ci-approved label Oct 20, 2026

Addressing review.

b18728e

github-actions Bot removed the triage PRs from the community label Oct 21, 2026

lucasbru reviewed Oct 21, 2026

View reviewed changes

Addressing review.

2b2398e

chickenchickenlove requested a review from lucasbru October 21, 2026 12:14

Conversation

chickenchickenlove commented Oct 18, 2026 • edited by github-actions Bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

🟢 Approval recommended

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!

chickenchickenlove Oct 20, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

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!

lucasbru commented Oct 20, 2026

Uh oh!

chickenchickenlove commented Oct 20, 2026

Uh oh!

lucasbru commented Oct 20, 2026

Uh oh!

chickenchickenlove commented Oct 20, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

lucasbru commented Oct 20, 2026

Uh oh!

chickenchickenlove commented Oct 20, 2026

Uh oh!

lucasbru commented Oct 21, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

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!

chickenchickenlove commented Oct 21, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

chickenchickenlove commented Oct 18, 2026 •

edited by github-actions Bot

Loading

chickenchickenlove Oct 20, 2026 •

edited

Loading

chickenchickenlove commented Oct 20, 2026 •

edited

Loading

lucasbru commented Oct 21, 2026 •

edited

Loading