Apply style fix

Author: k-raina

src/confluent_kafka/src/Consumer.c

@@ -116,8 +116,8 @@ static int Consumer_traverse(Handle *self, visitproc visit, void *arg) {
  ****************************************************************************/
 
 
-static PyObject *Consumer_subscribe(Handle *self, PyObject *args,
-                                    PyObject *kwargs) {
+static PyObject *
+Consumer_subscribe(Handle *self, PyObject *args, PyObject *kwargs) {
 
         rd_kafka_topic_partition_list_t *topics;
         static char *kws[] = {"topics", "on_assign", "on_revoke", "on_lost",
@@ -970,7 +970,7 @@ Consumer_offsets_for_times(Handle *self, PyObject *args, PyObject *kwargs) {
 
 /**
  * @brief Poll for a single message from the subscribed topics.
- * 
+ *
  * Instead of a single blocking call to rd_kafka_consumer_poll() with the
  * full timeout, this function:
  * 1. Splits the timeout into 200ms chunks
@@ -988,14 +988,13 @@ Consumer_offsets_for_times(Handle *self, PyObject *args, PyObject *kwargs) {
  * @return PyObject* Message object, None if timeout, or NULL on error
  *         (raises KeyboardInterrupt if signal detected)
  */
-static PyObject *Consumer_poll(Handle *self, PyObject *args,
-                                PyObject *kwargs) {
-        double tmout = -1.0f;
-        static char *kws[] = {"timeout", NULL};
+static PyObject *Consumer_poll(Handle *self, PyObject *args, PyObject *kwargs) {
+        double tmout            = -1.0f;
+        static char *kws[]      = {"timeout", NULL};
         rd_kafka_message_t *rkm = NULL;
         PyObject *msgobj;
         CallState cs;
-        const int CHUNK_TIMEOUT_MS = 200;  /* 200ms chunks for signal checking */
+        const int CHUNK_TIMEOUT_MS = 200; /* 200ms chunks for signal checking */
         int total_timeout_ms;
         int chunk_timeout_ms;
         int chunk_count = 0;
@@ -1021,15 +1020,16 @@ static PyObject *Consumer_poll(Handle *self, PyObject *args,
         } else {
                 while (1) {
                         /* Calculate timeout for this chunk */
-                        chunk_timeout_ms = calculate_chunk_timeout(total_timeout_ms, chunk_count,
-                                                                  CHUNK_TIMEOUT_MS);
+                        chunk_timeout_ms = calculate_chunk_timeout(
+                            total_timeout_ms, chunk_count, CHUNK_TIMEOUT_MS);
                         if (chunk_timeout_ms == 0) {
                                 /* Timeout expired */
                                 break;
                         }
 
                         /* Poll with chunk timeout */
-                        rkm = rd_kafka_consumer_poll(self->rk, chunk_timeout_ms);
+                        rkm =
+                            rd_kafka_consumer_poll(self->rk, chunk_timeout_ms);
 
                         /* If we got a message, exit the loop */
                         if (rkm) {
@@ -1099,7 +1099,7 @@ Consumer_memberid(Handle *self, PyObject *args, PyObject *kwargs) {
  * Instead of a single blocking call to rd_kafka_consume_batch_queue() with the
  * full timeout, this function:
  * 1. Splits the timeout into 200ms chunks
- * 2. Calls rd_kafka_consume_batch_queue() with chunk timeout 
+ * 2. Calls rd_kafka_consume_batch_queue() with chunk timeout
  * 3. Between chunks, re-acquires GIL and calls PyErr_CheckSignals()
  * 4. If signal detected, returns NULL (raises KeyboardInterrupt)
  * 5. Continues until messages received, timeout expired, or signal detected.
@@ -1111,11 +1111,11 @@ Consumer_memberid(Handle *self, PyObject *args, PyObject *kwargs) {
  *                consume per call. Default: 1. Maximum: 1000000.
  *              - timeout (float, optional): Timeout in seconds.
  *                Default: -1.0 (infinite timeout)
- * @return PyObject* List of Message objects, empty list if timeout, or NULL on error
- *         (raises KeyboardInterrupt if signal detected)
+ * @return PyObject* List of Message objects, empty list if timeout, or NULL on
+ * error (raises KeyboardInterrupt if signal detected)
  */
-static PyObject *Consumer_consume(Handle *self, PyObject *args,
-                                  PyObject *kwargs) {
+static PyObject *
+Consumer_consume(Handle *self, PyObject *args, PyObject *kwargs) {
         unsigned int num_messages = 1;
         double tmout              = -1.0f;
         static char *kws[]        = {"num_messages", "timeout", NULL};
@@ -1124,7 +1124,7 @@ static PyObject *Consumer_consume(Handle *self, PyObject *args,
         rd_kafka_queue_t *rkqu = self->u.Consumer.rkqu;
         CallState cs;
         Py_ssize_t i, n = 0;
-        const int CHUNK_TIMEOUT_MS = 200;  /* 200ms chunks for signal checking */
+        const int CHUNK_TIMEOUT_MS = 200; /* 200ms chunks for signal checking */
         int total_timeout_ms;
         int chunk_timeout_ms;
         int chunk_count = 0;
@@ -1160,37 +1160,40 @@ static PyObject *Consumer_consume(Handle *self, PyObject *args,
          * ThreadPool. Only use wakeable poll for
          * blocking calls that need to be interruptible. */
         if (total_timeout_ms >= 0 && total_timeout_ms < CHUNK_TIMEOUT_MS) {
-                n = (Py_ssize_t)rd_kafka_consume_batch_queue(rkqu, total_timeout_ms,
-                                                             rkmessages, num_messages);
+                n = (Py_ssize_t)rd_kafka_consume_batch_queue(
+                    rkqu, total_timeout_ms, rkmessages, num_messages);
 
                 if (n < 0) {
                         /* Error - need to restore GIL before setting error */
                         PyEval_RestoreThread(cs.thread_state);
                         free(rkmessages);
-                        cfl_PyErr_Format(rd_kafka_last_error(),
-                                         "%s", rd_kafka_err2str(rd_kafka_last_error()));
+                        cfl_PyErr_Format(
+                            rd_kafka_last_error(), "%s",
+                            rd_kafka_err2str(rd_kafka_last_error()));
                         return NULL;
                 }
         } else {
                 while (1) {
                         /* Calculate timeout for this chunk */
-                        chunk_timeout_ms = calculate_chunk_timeout(total_timeout_ms, chunk_count,
-                                                                  CHUNK_TIMEOUT_MS);
+                        chunk_timeout_ms = calculate_chunk_timeout(
+                            total_timeout_ms, chunk_count, CHUNK_TIMEOUT_MS);
                         if (chunk_timeout_ms == 0) {
                                 /* Timeout expired */
                                 break;
                         }
 
                         /* Consume with chunk timeout */
-                        n = (Py_ssize_t)rd_kafka_consume_batch_queue(rkqu, chunk_timeout_ms,
-                                                                       rkmessages, num_messages);
+                        n = (Py_ssize_t)rd_kafka_consume_batch_queue(
+                            rkqu, chunk_timeout_ms, rkmessages, num_messages);
 
                         if (n < 0) {
-                                /* Error - need to restore GIL before setting error */
+                                /* Error - need to restore GIL before setting
+                                 * error */
                                 PyEval_RestoreThread(cs.thread_state);
                                 free(rkmessages);
-                                cfl_PyErr_Format(rd_kafka_last_error(),
-                                                 "%s", rd_kafka_err2str(rd_kafka_last_error()));
+                                cfl_PyErr_Format(
+                                    rd_kafka_last_error(), "%s",
+                                    rd_kafka_err2str(rd_kafka_last_error()));
                                 return NULL;
                         }
 

src/confluent_kafka/src/Producer.c

@@ -339,7 +339,7 @@ Producer_produce(Handle *self, PyObject *args, PyObject *kwargs) {
 
 /**
  * @brief Poll for producer events with wakeable pattern for interruptibility.
- * 
+ *
  * This function:
  * 1. Splits the timeout into 200ms chunks
  * 2. Calls rd_kafka_poll() with chunk timeout
@@ -349,14 +349,14 @@ Producer_produce(Handle *self, PyObject *args, PyObject *kwargs) {
  *
  * @param self Producer handle
  * @param tmout Timeout in milliseconds (-1 for infinite)
- * @returns -1 if callback crashed, signal detected, or poll() failed, else the number
- * of events served.
+ * @returns -1 if callback crashed, signal detected, or poll() failed, else the
+ * number of events served.
  */
 static int Producer_poll0(Handle *self, int tmout) {
         int r = 0;
         CallState cs;
-        const int CHUNK_TIMEOUT_MS = 200;  /* 200ms chunks for signal checking */
-        int total_timeout_ms = tmout;
+        const int CHUNK_TIMEOUT_MS = 200; /* 200ms chunks for signal checking */
+        int total_timeout_ms       = tmout;
         int chunk_timeout_ms;
         int chunk_count = 0;
 
@@ -371,27 +371,28 @@ static int Producer_poll0(Handle *self, int tmout) {
         } else {
                 while (1) {
                         /* Calculate timeout for this chunk */
-                        chunk_timeout_ms = calculate_chunk_timeout(total_timeout_ms, chunk_count,
-                                                                  CHUNK_TIMEOUT_MS);
+                        chunk_timeout_ms = calculate_chunk_timeout(
+                            total_timeout_ms, chunk_count, CHUNK_TIMEOUT_MS);
                         if (chunk_timeout_ms == 0) {
                                 /* Timeout expired */
                                 break;
                         }
 
                         /* Poll with chunk timeout */
-                        int chunk_result = rd_kafka_poll(self->rk, chunk_timeout_ms);
+                        int chunk_result =
+                            rd_kafka_poll(self->rk, chunk_timeout_ms);
                         /* Error from poll */
                         if (chunk_result < 0) {
                                 r = chunk_result;
                                 break;
                         }
-                        r += chunk_result;  /* Accumulate events processed */
+                        r += chunk_result; /* Accumulate events processed */
 
                         chunk_count++;
 
                         /* Check for signals between chunks */
                         if (check_signals_between_chunks(self, &cs)) {
-                                return -1;  /* Signal detected */
+                                return -1; /* Signal detected */
                         }
                 }
         }
@@ -426,8 +427,9 @@ static PyObject *Producer_poll(Handle *self, PyObject *args, PyObject *kwargs) {
 
 
 /**
- * @brief Flush all messages in the producer queue with wakeable pattern for interruptibility.
- * 
+ * @brief Flush all messages in the producer queue with wakeable pattern for
+ * interruptibility.
+ *
  * Instead of a single blocking call to rd_kafka_flush() with the
  * full timeout, this function:
  * 1. Splits the timeout into 200ms chunks
@@ -451,7 +453,7 @@ Producer_flush(Handle *self, PyObject *args, PyObject *kwargs) {
         static char *kws[] = {"timeout", NULL};
         rd_kafka_resp_err_t err;
         CallState cs;
-        const int CHUNK_TIMEOUT_MS = 200;  /* 200ms chunks for signal checking */
+        const int CHUNK_TIMEOUT_MS = 200; /* 200ms chunks for signal checking */
         int total_timeout_ms;
         int chunk_timeout_ms;
         int chunk_count = 0;
@@ -474,15 +476,16 @@ Producer_flush(Handle *self, PyObject *args, PyObject *kwargs) {
         if (total_timeout_ms >= 0 && total_timeout_ms < CHUNK_TIMEOUT_MS) {
                 err = rd_kafka_flush(self->rk, total_timeout_ms);
         } else {
-                /* For infinite timeout, we need to keep looping and checking for signals.
-                 * rd_kafka_flush() waits for messages that were in the queue when it's called.
-                 * When flush() returns NO_ERROR, it means all messages that were queued at
-                 * that point have been delivered.
-                 * Note: Messages produced after flush() starts are not included in the current flush. */
+                /* For infinite timeout, we need to keep looping and checking
+                 * for signals. rd_kafka_flush() waits for messages that were in
+                 * the queue when it's called. When flush() returns NO_ERROR, it
+                 * means all messages that were queued at that point have been
+                 * delivered. Note: Messages produced after flush() starts are
+                 * not included in the current flush. */
                 while (1) {
                         /* Calculate timeout for this chunk */
-                        chunk_timeout_ms = calculate_chunk_timeout(total_timeout_ms, chunk_count,
-                                                                  CHUNK_TIMEOUT_MS);
+                        chunk_timeout_ms = calculate_chunk_timeout(
+                            total_timeout_ms, chunk_count, CHUNK_TIMEOUT_MS);
                         if (chunk_timeout_ms == 0) {
                                 /* Timeout expired */
                                 err = RD_KAFKA_RESP_ERR__TIMED_OUT;
@@ -492,10 +495,11 @@ Producer_flush(Handle *self, PyObject *args, PyObject *kwargs) {
                         /* Flush with chunk timeout */
                         err = rd_kafka_flush(self->rk, chunk_timeout_ms);
 
-                        /* Always check for signals between chunks (critical for interruptibility) */
+                        /* Always check for signals between chunks (critical for
+                         * interruptibility) */
                         chunk_count++;
                         if (check_signals_between_chunks(self, &cs)) {
-                                return NULL;  /* Signal detected */
+                                return NULL; /* Signal detected */
                         }
 
                         if (err == RD_KAFKA_RESP_ERR_NO_ERROR) {

src/confluent_kafka/src/confluent_kafka.h

@@ -545,26 +545,29 @@ static CFL_UNUSED CFL_INLINE int cfl_timeout_ms(double tmout) {
  * - Infinite timeouts (-1) use the chunk size repeatedly
  * - Finite timeouts are properly divided and don't exceed the total
  * - The final chunk uses any remaining time (may be < chunk_size)
- 
+
  *
  * @param total_timeout_ms Total timeout in milliseconds (-1 for infinite)
  * @param chunk_count Current chunk iteration count (0-based)
  * @param chunk_timeout_ms Chunk size in milliseconds (200ms by default)
  * @return int Chunk timeout in milliseconds, or 0 if total timeout expired
  */
-static CFL_UNUSED CFL_INLINE int calculate_chunk_timeout(int total_timeout_ms, int chunk_count,
-                                                          int chunk_timeout_ms) {
+static CFL_UNUSED CFL_INLINE int calculate_chunk_timeout(int total_timeout_ms,
+                                                         int chunk_count,
+                                                         int chunk_timeout_ms) {
         if (total_timeout_ms < 0) {
                 /* Infinite timeout - use chunk size */
                 return chunk_timeout_ms;
         } else {
                 /* Finite timeout - calculate remaining */
-                int remaining_ms = total_timeout_ms - (chunk_count * chunk_timeout_ms);
+                int remaining_ms =
+                    total_timeout_ms - (chunk_count * chunk_timeout_ms);
                 if (remaining_ms <= 0) {
                         /* Timeout expired */
                         return 0;
                 }
-                return (remaining_ms < chunk_timeout_ms) ? remaining_ms : chunk_timeout_ms;
+                return (remaining_ms < chunk_timeout_ms) ? remaining_ms
+                                                         : chunk_timeout_ms;
         }
 }
 
@@ -581,27 +584,31 @@ static CFL_UNUSED CFL_INLINE int calculate_chunk_timeout(int total_timeout_ms, i
  * - SIGINT (Ctrl+C): Raises KeyboardInterrupt exception
  * - SIGTERM: Can raise SystemExit or be handled by user code
  * - Other signals: If the user has registered handlers via Python's `signal`
- *   module, those will also be checked (e.g., signal.signal(signal.SIGUSR1, handler)).
- *   User code will need to handle these signals accordingly.
+ *   module, those will also be checked (e.g., signal.signal(signal.SIGUSR1,
+ * handler)). User code will need to handle these signals accordingly.
  *
  *
  * @param self Handle (Producer or Consumer)
  * @param cs CallState structure (thread state will be updated)
- * @return int 0 if no signal detected (continue), 1 if signal detected (should return NULL)
+ * @return int 0 if no signal detected (continue), 1 if signal detected (should
+ * return NULL)
  */
-static CFL_UNUSED CFL_INLINE int check_signals_between_chunks(Handle *self, CallState *cs) {
+static CFL_UNUSED CFL_INLINE int check_signals_between_chunks(Handle *self,
+                                                              CallState *cs) {
         /* Re-acquire GIL */
         PyEval_RestoreThread(cs->thread_state);
 
         /* Check for pending signals (KeyboardInterrupt, etc.) */
         /* PyErr_CheckSignals() already set the exception */
         if (PyErr_CheckSignals() == -1) {
-                /* Note: GIL is already held, but CallState_end expects to restore it */
-                /* Save thread state again so CallState_end can restore it properly */
+                /* Note: GIL is already held, but CallState_end expects to
+                 * restore it */
+                /* Save thread state again so CallState_end can restore it
+                 * properly */
                 cs->thread_state = PyEval_SaveThread();
                 if (!CallState_end(self, cs)) {
                         /* CallState_end detected signal and cleaned up */
-                        return 1;  /* Signal detected */
+                        return 1; /* Signal detected */
                 }
                 return 1;
         }

tests/integration/consumer/test_consumer_wakeable_poll_consume.py

@@ -35,12 +35,14 @@ def test_poll_message_delivery_with_wakeable_pattern(kafka_cluster):
     producer.flush(timeout=1.0)
 
     # Create consumer with wakeable poll pattern settings
-    consumer_conf = kafka_cluster.client_conf({
-        'group.id': 'test-poll-message-available',
-        'socket.timeout.ms': 100,
-        'session.timeout.ms': 6000,
-        'auto.offset.reset': 'earliest'
-    })
+    consumer_conf = kafka_cluster.client_conf(
+        {
+            'group.id': 'test-poll-message-available',
+            'socket.timeout.ms': 100,
+            'session.timeout.ms': 6000,
+            'auto.offset.reset': 'earliest',
+        }
+    )
     consumer = TestConsumer(consumer_conf)
     consumer.subscribe([topic])
 
@@ -77,12 +79,14 @@ def test_consume_message_delivery_with_wakeable_pattern(kafka_cluster):
     producer.flush(timeout=1.0)
 
     # Create consumer with wakeable poll pattern settings
-    consumer_conf = kafka_cluster.client_conf({
-        'group.id': 'test-consume-messages-available',
-        'socket.timeout.ms': 100,
-        'session.timeout.ms': 6000,
-        'auto.offset.reset': 'earliest'
-    })
+    consumer_conf = kafka_cluster.client_conf(
+        {
+            'group.id': 'test-consume-messages-available',
+            'socket.timeout.ms': 100,
+            'session.timeout.ms': 6000,
+            'auto.offset.reset': 'earliest',
+        }
+    )
     consumer = TestConsumer(consumer_conf)
     consumer.subscribe([topic])
 
@@ -106,8 +110,7 @@ def test_consume_message_delivery_with_wakeable_pattern(kafka_cluster):
         assert msg.value() is not None, f"Message {i} has no value"
         # Verify we got the expected messages
         expected_value = f'test-message-{i}'.encode()
-        expected_msg = (f"Message {i} value mismatch: expected {expected_value}, "
-                        f"got {msg.value()}")
+        expected_msg = f"Message {i} value mismatch: expected {expected_value}, " f"got {msg.value()}"
         assert msg.value() == expected_value, expected_msg
 
     consumer.close()