[
https://issues.apache.org/jira/browse/KAFKA-21188?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18121174#comment-18121174
]
sanghyeok An commented on KAFKA-21188:
--------------------------------------
For the reproducing,
{code:java}
public static void main(String[] args) {
final Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "ASH-HELLO1111100");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, "streams");
props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 1);
props.put(StreamsConfig.POLL_MS_CONFIG, 10);
props.put(StreamsConfig.REPLICATION_FACTOR_CONFIG, 1);
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG,
Serdes.String().getClass().getName());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,
Serdes.String().getClass().getName());
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
final StreamsBuilder builder = new StreamsBuilder();
KStream<Object, Object> streamA = builder.stream("A");
KStream<Object, Object> streamB = builder.stream("B");
streamA.merge(streamB).to("output");
final Topology topology = builder.build();
System.out.println(topology.describe());
final KafkaStreams kafkaStreams = new KafkaStreams(topology, props);
Runtime.getRuntime().addShutdownHook(new Thread(kafkaStreams::close));
kafkaStreams.start();
} {code}
Use this scripts to create topics.
{code:java}
$ bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic A
--partitions 1 --replication-factor 1
$ bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic B
--partitions 3 --replication-factor 1{code}
Use this scripts to get output from broker.
{code:java}
$ kcat -b localhost:9092 -C -t output -o beginning -f '%s\n' {code}
Use this scripts to producer messages.
{code:java}
$ printf 'probe1-from-B-1\n' | kcat -b localhost:9092 -P -t B -p 1{code}
Then, there is no observed record at all from "output" topic.
> Merging topics with different partition counts leaves some records
> unprocessed under the Streams Rebalance Protocol
> -------------------------------------------------------------------------------------------------------------------
>
> Key: KAFKA-21188
> URL: https://issues.apache.org/jira/browse/KAFKA-21188
> Project: Kafka
> Issue Type: Bug
> Reporter: sanghyeok An
> Assignee: sanghyeok An
> Priority: Major
>
> With {*}group.protocol=streams{*}, merging two topics with different
> partition counts includes nonexistent partitions in the tasks’ inputs,
> leaving some records unprocessed.
> *Steps to Reproduce*
> 1. Create topic A with 1 partition({*}0{*}) and topic B with 3
> partitions({*}0, 1, 2{*}).
> 2. Run the following topology with {*}group.protocol=streams{*}:
> {code:java}
> KStream<String, String> streamA = builder.stream("A");
> KStream<String, String> streamB = builder.stream("B");
> streamA.merge(streamB).to("output");
> {code}
> 3. Produce records to {*}A-0{*}, {*}B-0{*}, {*}B-1{*}, and {*}B-2{*}.
> * *Expected:* All input records appear in `output`.
> * *Actual:* Records from `B-1` and `B-2` do not appear in `output`, and the
> following messages repeatedly reference nonexistent partitions:
> {code:java}
> Requesting the log end offset for A-1 in order to compute lag
> Requesting the log end offset for A-2 in order to compute lag
> {code}
>
> *Cause*
> StreamsMembershipManager#topicPartitionsForActiveTasks and
> DefaultStreamsRebalanceListener#toTopicPartitions apply the task’s partition
> ID to every input topic without checking each topic’s partition count.
> As a result, a task that should include only *B-1* also includes {*}A-1{*}.
> The same issue occurs for partition ID 2.
> The classic protocol’s PartitionGrouper checks each topic’s partition count
> and includes only existing partitions. The new protocol’s conversion paths
> need equivalent handling.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
