[
https://issues.apache.org/jira/browse/KAFKA-21302?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18129154#comment-18129154
]
sanghyeok An edited comment on KAFKA-21302 at 10/20/26 1:34 PM:
----------------------------------------------------------------
[~lucasbru]
By the way, how about this?
Run the same code repeatedily, but in a fresh JVM each time.
{code:java}
var builder = new StreamsBuilder();KStream<String, String> a =
builder.stream(Set.of("Aa", "BB"));
KStream<String, String> b = builder.stream("right");a.join(
b,
(x, y) -> x + y,
JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofSeconds(1))
);
{code}
Test Result here on client side.
{code:java}
001: sourceTopics=[Aa, BB, right]; copartitionGroups=[[0, 1, 2]]
002: sourceTopics=[Aa, BB, right]; copartitionGroups=[[0, 1, 2]]
003: sourceTopics=[Aa, BB, right]; copartitionGroups=[[0, 1, 2]]
004: sourceTopics=[Aa, BB, right]; copartitionGroups=[[1, 0, 2]]
...
097: sourceTopics=[Aa, BB, right]; copartitionGroups=[[1, 0, 2]]
098: sourceTopics=[Aa, BB, right]; copartitionGroups=[[0, 1, 2]]
099: sourceTopics=[Aa, BB, right]; copartitionGroups=[[1, 0, 2]]
100: sourceTopics=[Aa, BB, right]; copartitionGroups=[[0, 1, 2]]
Distinct outputs: 2
47/100: sourceTopics=[Aa, BB, right]; copartitionGroups=[[0, 1, 2]]
53/100: sourceTopics=[Aa, BB, right]; copartitionGroups=[[1, 0, 2]] {code}
was (Author: JIRAUSER303328):
[~lucasbru]
By the way, how about this?
Run the same code repeatedily, but in a fresh JVM each time.
{code:java}
var builder = new StreamsBuilder();KStream<String, String> a =
builder.stream(Set.of("Aa", "BB"));
KStream<String, String> b = builder.stream("right");a.join(
b,
(x, y) -> x + y,
JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofSeconds(1))
);
{code}
Test Result here
{code:java}
001: sourceTopics=[Aa, BB, right]; copartitionGroups=[[0, 1, 2]]
002: sourceTopics=[Aa, BB, right]; copartitionGroups=[[0, 1, 2]]
003: sourceTopics=[Aa, BB, right]; copartitionGroups=[[0, 1, 2]]
004: sourceTopics=[Aa, BB, right]; copartitionGroups=[[1, 0, 2]]
...
097: sourceTopics=[Aa, BB, right]; copartitionGroups=[[1, 0, 2]]
098: sourceTopics=[Aa, BB, right]; copartitionGroups=[[0, 1, 2]]
099: sourceTopics=[Aa, BB, right]; copartitionGroups=[[1, 0, 2]]
100: sourceTopics=[Aa, BB, right]; copartitionGroups=[[0, 1, 2]]
Distinct outputs: 2
47/100: sourceTopics=[Aa, BB, right]; copartitionGroups=[[0, 1, 2]]
53/100: sourceTopics=[Aa, BB, right]; copartitionGroups=[[1, 0, 2]] {code}
> Streams group joins rejected due to copartition topic ordering differences
> --------------------------------------------------------------------------
>
> Key: KAFKA-21302
> URL: https://issues.apache.org/jira/browse/KAFKA-21302
> Project: Kafka
> Issue Type: Bug
> Components: streams
> Reporter: sanghyeok An
> Assignee: sanghyeok An
> Priority: Minor
>
> With {{{}group.protocol=streams{}}}, a new application instance can fail to
> join its group even when it uses the same logical topology, depending on the
> order in which source topics are supplied.
> For example, two instances with the same {{application.id}} construct their
> topologies as follows:
> * Instance A joins {{stream(List.of("Aa", "BB"))}} with a stream from topic
> {{{}z{}}}.
> * Instance B joins {{stream(List.of("BB", "Aa"))}} with a stream from topic
> {{{}z{}}}.
> All three topics have the same partition count. Node names, join settings,
> and all other configuration are identical. Only the source topic order
> differs, so both topologies have the same copartition constraints.
> The client sorts the source topic list in the heartbeat, producing {{[Aa, BB,
> z]}} for both instances. However, copartition indexes are collected in Set
> iteration order, so A can send {{[0, 1, 2]}} while B sends {{{}[1, 0, 2]{}}}.
> After A joins and its topology is stored, the coordinator treats B’s
> different index ordering as a topology change and returns:
>
> {{INVALID_REQUEST: Topology updates are not supported yet.}}
> B’s client treats this error as fatal, transitions from {{JOINING}} to
> {{{}FATAL{}}}, and stops sending heartbeats. As a result, adding or replacing
> an instance can fail despite the instances using logically identical
> topologies.
> This occurs when the copartition list ordering actually differs. Reversing an
> arbitrary source topic list does not necessarily trigger the issue. In this
> example, {{Aa}} and {{BB}} have the same Java string hash code, allowing
> their input order to affect Set iteration order.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
