Make long waiting consumer and producer API wakeable #2126

k-raina · 2025-11-07T15:53:58Z

Summary

This PR fixes a critical usability issue where Consumer.poll(), Consumer.consume(), Producer.poll(), and Producer.flush() would block indefinitely and not respond to Ctrl+C (KeyboardInterrupt) signals.

Fixes: #209 and #807

Problem

When calling blocking operations with infinite or very long timeouts:

Consumer.poll() / Consumer.consume() - would block indefinitely waiting for messages
Producer.poll() / Producer.flush() - would block indefinitely waiting for delivery callbacks or queue flushing

Because these operations release the Python Global Interpreter Lock (GIL) during blocking calls to the underlying librdkafka C library, Python's signal handling mechanism couldn't detect Ctrl+C signals, making it impossible to gracefully interrupt these operations.

Solution

The fix implements a "wakeable poll" pattern that:

Chunks long timeouts: Instead of making a single blocking call with the full timeout, the implementation breaks it into smaller 200ms chunks
Periodic signal checking: Between chunks, the code re-acquires the Python GIL and calls PyErr_CheckSignals() to detect pending KeyboardInterrupt signals
Immediate interruption: If a signal is detected, the operation returns immediately, allowing the KeyboardInterrupt to propagate to the Python code

Impact

This fix significantly improves the developer and operational experience for applications using Python client.

Before this fix:

Developers had to forcefully kill processes during development and testing
No way to gracefully stop consumer/producer loops during debugging
Producer flush() operations couldn't be interrupted, causing issues during shutdown

After this fix:

Developers can use standard Ctrl+C to interrupt all blocking operations
Clean shutdown during development and testing

Testing

Consumer Tests (`tests/test_Consumer.py`)

Utility function tests:
- test_calculate_chunk_timeout_utility_function(): Tests chunk timeout calculation logic
- test_check_signals_between_chunks_utility_function(): Tests signal detection between chunks
- test_wakeable_poll_utility_functions_interaction(): Tests interaction between both utilities
Poll interruptibility tests:
- test_wakeable_poll_interruptibility_and_messages(): Tests poll() can be interrupted and still handles messages correctly
- test_wakeable_poll_edge_cases(): Tests edge cases (zero timeout, closed consumer, short timeouts)
Consume interruptibility tests:
- test_wakeable_consume_interruptibility_and_messages(): Tests consume() can be interrupted and still handles messages correctly
- test_wakeable_consume_edge_cases(): Tests edge cases (zero timeout, invalid parameters, short timeouts)

Producer Tests (`tests/test_Producer.py`)

Utility function tests:
- test_wakeable_poll_utility_functions_interaction(): Tests chunk calculation and signal checking work together
Poll interruptibility tests:
- test_wakeable_poll_interruptibility_and_messages(): Tests poll() can be interrupted and still handles delivery callbacks correctly
- test_wakeable_poll_edge_cases(): Tests edge cases (zero timeout, closed producer, short timeouts)
Flush interruptibility tests:
- test_wakeable_flush_interruptibility_and_messages(): Tests flush() can be interrupted and still delivers messages correctly
- test_wakeable_flush_edge_cases(): Tests edge cases (zero timeout, closed producer, short timeouts, empty queue)

Shared Utility Tests (`tests/test_wakeable_utilities.py`)

Tests for newly added utility functions

Integration Tests

tests/integration/consumer/test_consumer_wakeable_poll_consume.py: Verifies wakeable pattern doesn't interfere with normal message consumption
tests/integration/producer/test_producer_wakeable_poll_flush.py: Verifies wakeable pattern doesn't interfere with normal message production and delivery callbacks

All tests use a helper function TestUtils.send_sigint_after_delay() to simulate Ctrl+C in automated tests.

Performance Impact

Minimal overhead: Only adds signal checks between 200ms chunks
No impact on short timeouts: Timeouts < 200ms bypass chunking entirely

Manual Testing

Replicable code

test_wakeable_consume_interrupt.py
test_wakeable_poll_interrupt.py
test_wakeable_producer_flush_interrupt.py
test_wakeable_producer_poll_interrupt.py

Example Run

 Queue before flush: 101000 messages
  (Many messages are in queue and in-flight, waiting for acknowledgment)

[20:45:54] Calling flush() with infinite timeout...
    (This will block until all messages are flushed or Ctrl+C)
    Background thread is continuously adding messages to the queue
    Current queue length: 101000 messages
    Press Ctrl+C to test interruptibility...

^C
======================================================================
✓ KeyboardInterrupt caught!
  Interrupted after 1.07 seconds
  Background thread produced 1167722 messages during this time
  (Had 53 production errors)
  With wakeable pattern, interruption should occur within ~200ms of Ctrl+C
  ⚠ Interruption took 1.07s (may indicate wakeable pattern issue)
======================================================================

Stopping background producer thread...
  Final stats: 1167722 messages produced
  (53 production errors)

Closing producer...
Producer closed.
Traceback (most recent call last):
  File "/Users/kaushikraina/projects/njc/confluent-kafka-python/test_wakeable_producer_flush_interrupt.py", line 292, in <module>
    main()
  File "/Users/kaushikraina/projects/njc/confluent-kafka-python/test_wakeable_producer_flush_interrupt.py", line 223, in main
    remaining = producer.flush(timeout=-1.0)  # Infinite timeout - will block
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
KeyboardInterrupt

confluent-cla-assistant · 2025-11-07T15:54:10Z

🎉 All Contributor License Agreements have been signed. Ready to merge.
_{Please push an empty commit if you would like to re-run the checks to verify CLA status for all contributors.}

Copilot

Pull Request Overview

This PR fixes critical usability issues where Consumer.poll() and Consumer.consume() would block indefinitely and not respond to Ctrl+C (KeyboardInterrupt) signals. The solution implements a "wakeable poll" pattern that chunks long timeouts into 200ms intervals and periodically checks for pending signals between chunks, allowing proper signal handling and graceful interruption.

Key Changes:

Implemented helper functions calculate_chunk_timeout() and check_signals_between_chunks() in C code to enable interruptible polling
Modified Consumer.poll() and Consumer.consume() to use chunked polling with periodic signal checks
Added comprehensive test coverage for utility functions, interruptibility, edge cases, and message handling

Reviewed Changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated 2 comments.

File	Description
tests/test_Consumer.py	Added utility helper `send_sigint_after_delay()` and 7 new test functions covering chunk timeout calculation, signal detection, utility function interaction, poll/consume interruptibility, and edge cases
src/confluent_kafka/src/Consumer.c	Implemented wakeable poll pattern with helper functions and refactored `Consumer_poll()` and `Consumer_consume()` to use chunked polling with signal checking
CHANGELOG.md	Documented the fix for blocking poll/consume operations

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Copilot · 2025-11-07T15:54:40Z

src/confluent_kafka/src/Consumer.c

+ * 
+ * Instead of a single blocking call to rd_kafka_consumer_poll() with the
+ * full timeout, this function:
+ * 1. Splits the timeout into 200ms chunks


[nitpick] The function documentation would benefit from documenting the CHUNK_TIMEOUT_MS constant value (200ms) in the description to match the implementation details mentioned in comment lines.

Suggested change

* 1. Splits the timeout into 200ms chunks

* 1. Splits the timeout into 200ms chunks (CHUNK_TIMEOUT_MS = 200ms)

Copilot · 2025-11-07T15:54:41Z

src/confluent_kafka/src/Consumer.c

-                return NULL;
-        }
-
+        /* Create Python list from messages  */


Extra whitespace before closing comment marker.

Suggested change

/* Create Python list from messages */

/* Create Python list from messages */

MSeal

One minor comment, tested out things locally with a 30s timeout with this vs master... much better experience of always being able to interrupt.

MSeal · 2025-11-07T23:10:23Z

tests/test_Consumer.py

+    time.sleep(delay_seconds)
+    try:
+        os.kill(os.getpid(), signal.SIGINT)
+    except Exception:


Should probably catch the KeyboardInterrupt here instead so you don't mask other errors being raised instead

Thanks
os.kill() raises OSError, not KeyboardInterrupt, i have removed try catch block as any exception raised in this function should not be masked

sonarqube-confluent · 2025-11-11T09:52:35Z

Quality Gate failed

Failed conditions
0.0% Coverage on New Code (required ≥ 80%)

See analysis details on SonarQube

sonarqube-confluent · 2025-11-27T10:48:30Z

Quality Gate failed

Failed conditions
0.0% Coverage on New Code (required ≥ 80%)

See analysis details on SonarQube

MSeal

Minor changes requested but up to you if you feel they are worth it. Test blocks were a bit dense/difficult to read after a long weekend losing mental context. I'm not sure how to make them easier to maintain but the next person coming along who has to fix those for future change will need to reread them a few times I think.

MSeal · 2025-12-01T21:45:09Z

src/confluent_kafka/src/confluent_kafka.h

+                        /* CallState_end detected signal and cleaned up */
+                        return 1; /* Signal detected */
+                }
+                return 1;


We simply always return 1 -- was this intentional? Seems like we'd either have a return 0 in one call path or cleanup the code to just call CallState_end and then return no matter what

MSeal · 2025-12-01T21:46:53Z

src/confluent_kafka/src/confluent_kafka.h

+        }
+
+        /* Re-release GIL for next iteration */
+        cs->thread_state = PyEval_SaveThread();


Maybe put this in an else block so you have clear control paths for calling PyEval_SaveThread appropriately if things get refactored. I've found those explicit conditional blocks help a lot down the line for clarity of intent after additional changes

MSeal · 2025-12-01T21:47:42Z

src/confluent_kafka/src/Consumer.c

+/****************************************************************************
+ *
+ *
+ * Consumer Methods


If we're going to have a giant comment block it maybe should be more descriptive. Though I'm not sure it's highly valuable here to do this anyway

MSeal · 2025-12-01T21:53:56Z

src/confluent_kafka/src/Consumer.c

+        /* Final GIL restore and signal check */
        if (!CallState_end(self, &cs)) {
                if (rkm)
                        rd_kafka_message_destroy(rkm);


I always found C code less risky to refactor errors with single line if's include wrapping brackets, but I don't believe we follow that convention here as of now. If we're doing that convention within a function let's be consistent to the rest of the calls in the same scope imo

MSeal · 2025-12-01T21:55:37Z

src/confluent_kafka/src/Consumer.c

+/**
+ * @brief Consume a batch of messages from the subscribed topics.
+ *
+ * Instead of a single blocking call to rd_kafka_consume_batch_queue() with the


Nothing to change but commenting that just enough code differences that it's probably best left separate even though it looks similar enough to share more code.

Copilot AI review requested due to automatic review settings November 7, 2025 15:53

k-raina requested review from a team and MSeal as code owners November 7, 2025 15:53

Copilot AI reviewed Nov 7, 2025

View reviewed changes

MSeal previously approved these changes Nov 7, 2025

View reviewed changes

airlock-confluentinc bot dismissed MSeal’s stale review via 56dc957 November 10, 2025 15:36

k-raina requested a review from MSeal November 10, 2025 15:55

MSeal previously approved these changes Nov 10, 2025

View reviewed changes

airlock-confluentinc bot dismissed MSeal’s stale review via 0df67a5 November 10, 2025 19:28

airlock-confluentinc bot force-pushed the fix/wakeable-poll-issues-209-807 branch from dcc1639 to 6180f01 Compare November 26, 2025 14:48

k-raina added 12 commits November 26, 2025 20:38

Make consumer - consumer and poll wakeable

e064d90

Dont mask error in send signal function

ed2095d

Fix flake8

cd44692

Skip lock release reacquire for small timeouts

28ef10f

Move real broker tests to integration tests

d7200d8

Fix flake8

1e10109

Standardarize assert timings for wakeble poll and consume

ffb8c2e

Minor comments update

447a6ce

Standardarize assert timings for wakeble poll and consume

49a6a6c

Add Wakeable pattern to poll and flush in python

10cf864

Add tests

ea444a3

Apply style fix

5a86d02

airlock-confluentinc bot force-pushed the fix/wakeable-poll-issues-209-807 branch from 6180f01 to 5a86d02 Compare November 26, 2025 15:08

k-raina changed the title ~~Make consumer - consumer and poll wakeable~~ Make long waiting consumer and producer API wakeable Nov 26, 2025

k-raina added 2 commits November 26, 2025 21:07

Fix CI

5aca67c

Fix CI

2ff45b1

k-raina requested a review from MSeal November 26, 2025 16:52

Update changelog

5be7933

MSeal approved these changes Dec 1, 2025

View reviewed changes

	* 1. Splits the timeout into 200ms chunks
	* 1. Splits the timeout into 200ms chunks (CHUNK_TIMEOUT_MS = 200ms)

	/* Create Python list from messages */
	/* Create Python list from messages */

Make long waiting consumer and producer API wakeable #2126

Are you sure you want to change the base?

Make long waiting consumer and producer API wakeable #2126

Uh oh!

Conversation

k-raina commented Nov 7, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Problem

Solution

Impact

Testing

Consumer Tests (tests/test_Consumer.py)

Producer Tests (tests/test_Producer.py)

Shared Utility Tests (tests/test_wakeable_utilities.py)

Integration Tests

Performance Impact

Manual Testing

Replicable code

Example Run

Uh oh!

confluent-cla-assistant bot commented Nov 7, 2025

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Reviewed Changes

Uh oh!

Copilot AI Nov 7, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Nov 7, 2025

Choose a reason for hiding this comment

Uh oh!

MSeal left a comment

Choose a reason for hiding this comment

Uh oh!

MSeal Nov 7, 2025

Choose a reason for hiding this comment

Uh oh!

k-raina Nov 10, 2025 • 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!

sonarqube-confluent bot commented Nov 11, 2025

Quality Gate failed

Uh oh!

sonarqube-confluent bot commented Nov 27, 2025

Quality Gate failed

Uh oh!

MSeal left a comment

Choose a reason for hiding this comment

Uh oh!

MSeal Dec 1, 2025

Choose a reason for hiding this comment

Uh oh!

MSeal Dec 1, 2025

Choose a reason for hiding this comment

Uh oh!

MSeal Dec 1, 2025

Choose a reason for hiding this comment

Uh oh!

MSeal Dec 1, 2025

Choose a reason for hiding this comment

Uh oh!

MSeal Dec 1, 2025

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

k-raina commented Nov 7, 2025 •

edited

Loading

Consumer Tests (`tests/test_Consumer.py`)

Producer Tests (`tests/test_Producer.py`)

Shared Utility Tests (`tests/test_wakeable_utilities.py`)

k-raina Nov 10, 2025 •

edited

Loading