[Bug]: Kafka timeout before the position for partition could be determined in one topic

[i-emek]

What happened?

'm having a Google dataflow job, I'm using apache beam ReadFromKafka to consume topic messages. I'm consuming 4 topics. The pipeline used to work fine, after we added a new broker to our kafka cluster and triggered a rebalancing, the consumer started failing on one specific topic but successfully kept working for the other 3 topics.
Same behaviour noticed when creating a new consumer group also

The error is: org.apache.kafka.common.errors.TimeoutException: Timeout of 60000ms expired before the position for partition topic_name-0 could be determined

Here is the code:

( pcoll | f"Read From Kafka {self.transf_label}" >> ReadFromKafka( consumer_config=self.kafka_consumer_config, topics=_topic_list, with_metadata=True, ) | f"Extract KVT {self.transf_label}" >> Map(extract_key_value_topic) | f"Decode messages {self.transf_label}" >> ParDo(DecodeKafkaMessage()) )

My config

{
        "bootstrap.servers": BROKERS,
        "security.protocol": SECURITY_PROTOCOL,
        "sasl.mechanism": SASL_MECHANISM,
        "group.id": "consumer-group-dummy",
        "session.timeout.ms": "60000",
        "sasl.jaas.config": f'org.apache.kafka.common.security.scram.ScramLoginModule required username="{SASL_USERNAME}" password="{SASL_PASSWORD}";',
        "auto.offset.reset": "latest",

    }

One thing to consider is our other Kafka consumer apps are not throwing any issue, this behavior is noticed only with the apache beam pipeline. Also when I tried to trigger locally a consumer with the same config and same consumer group using Kafka confluent library and simple python script, it seems to be working fine and pulling the messages.

I'm wondering if this is a bug from the expansion server which is using Kafka IO

Issue Priority

Priority: 3 (minor)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam YAML
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner