Hey,
we stumbled over the issue that in the current version of the package, a call (event.wait()) nested inside the KafkaConsumer.commit() method is ignoring the selected timeout-values and therefore can run indefinitely in certain conditions.
We only noticed it because in our case, it actually lead to a production issue, where multiple consumers stopped consuming any topics for 2h+
Method-Call-Path:
KafkaConsumer.commit()
-> ConsumerCoordinator.commit_offsets_sync()
-> NetworkSelector.run()
-> event.wait()
As can be reproduced in the example, the script stays/hangs on consumer.commit() and doesn't continue
Example for reproducing the problem:
#!/usr/bin/env python3
"""
Minimal reproduction example for the kafka-python-3.x sync wait.
No Kafka broker is required. After starting commit(), the process
intentionally hangs and must be manually terminated with Ctrl+C.
"""
import kafka
from kafka import KafkaConsumer, OffsetAndMetadata, TopicPartition
from kafka.coordinator.consumer import ConsumerCoordinator
from kafka.net.selector import NetworkSelector
if not kafka.__version__.startswith("3."):
raise RuntimeError(f"This example expects kafka-python 3.x, found: {kafka.__version__}")
selector = NetworkSelector()
selector._io_thread = object()
selector.call_soon_threadsafe = lambda _waiter: print(
"NetworkSelector.run(): waiter wasnt executed",
flush=True,
)
coordinator = object.__new__(ConsumerCoordinator)
coordinator.config = {"api_version": (2, 5)}
coordinator._net = selector
coordinator._invoke_completed_offset_commit_callbacks = lambda: None
consumer = KafkaConsumer(
# The explicit API version prevents a bootstrap network access.
bootstrap_servers=["127.0.0.1:1"],
api_version=(2, 5),
group_id="kafka-python-3-timeout-minimal",
enable_auto_commit=False,
request_timeout_ms=100,
fetch_max_wait_ms=10,
)
consumer._coordinator = coordinator
print(f"kafka-python={kafka.__version__}", flush=True)
print("consumer.commit(timeout_ms=100) was started", flush=True)
consumer.commit(
offsets={
TopicPartition("reproducer-topic", 0): OffsetAndMetadata(1, "", -1),
},
timeout_ms=100,
)
print("UNEXPECTED: consumer.commit() actually returned", flush=True)
Hey,
we stumbled over the issue that in the current version of the package, a call (event.wait()) nested inside the KafkaConsumer.commit() method is ignoring the selected timeout-values and therefore can run indefinitely in certain conditions.
We only noticed it because in our case, it actually lead to a production issue, where multiple consumers stopped consuming any topics for 2h+
Method-Call-Path:
KafkaConsumer.commit()
-> ConsumerCoordinator.commit_offsets_sync()
-> NetworkSelector.run()
-> event.wait()
As can be reproduced in the example, the script stays/hangs on consumer.commit() and doesn't continue
Example for reproducing the problem: