Adjusted to reuse exit logic

Author: MSeal

DEVELOPER.md

@@ -178,26 +178,6 @@ Integration tests (may require local/CI Kafka cluster; see tests/README.md):
 pytest -q tests/integration
 ```
 
-## Local Setup with UV
-
-Tested with python 3.11
-
-```bash
-# Modify pyproject.toml to require python version >=3.11
-# This fixes the cel-python dependency conflict
-uv venv --python 3.11
-source .venv/bin/activate
-
-uv sync --extra dev --extra tests
-uv pip install trivup setuptools
-pytest tests/
-
-# When making changes, change project.version in pyproject.toml before re-running:
-uv sync --extra dev --extra tests
-
-```
-
-
 ## Tests
 
 See [tests/README.md](tests/README.md) for instructions on how to run tests.

src/confluent_kafka/src/Producer.c

@@ -422,37 +422,32 @@ static PyObject *Producer_flush (Handle *self, PyObject *args,
 
 
 static PyObject *Producer_close(Handle *self, PyObject *args, PyObject *kwargs) {
-
-        double tmout = 1; // Default timeout is 1 second for close to clear rather than indefinitely
-        static char *kws[] = { "timeout", NULL };
         rd_kafka_resp_err_t err;
         CallState cs;
 
-        if (!PyArg_ParseTupleAndKeywords(args, kwargs, "|d", kws, &tmout))
-            return NULL;
-
         if (!self->rk)
             Py_RETURN_TRUE;
 
         CallState_begin(self, &cs);
 
-        // Flush any remaining messages before closing if possible
-        err = rd_kafka_flush(self->rk, cfl_timeout_ms(tmout));
-        if (err != RD_KAFKA_RESP_ERR_NO_ERROR) {
-            PyErr_WarnFormat(PyExc_RuntimeWarning, 1,
-                             "Producer flush failed during close: %s",
-                             rd_kafka_err2str(err));
-        }
-        rd_kafka_destroy(self->rk);
-        rd_kafka_log_print(self->rk, CK_LOG_INFO, "CLOSEINF", "Producer destroy requested");
+		/* Flush any pending messages (wait indefinitely to ensure delivery) */
+		err = rd_kafka_flush(self->rk, -1);
 
-        self->rk = NULL;
+		/* Destroy the producer (even if flush had issues) */
+		rd_kafka_destroy(self->rk);
+		self->rk = NULL;
 
-        if (!CallState_end(self, &cs))
-            return NULL;
+		if (!CallState_end(self, &cs))
+			return NULL;
 
-        Py_RETURN_TRUE;
+		/* If flush failed, warn but don't suppress original exception */
+		if (err != RD_KAFKA_RESP_ERR_NO_ERROR) {
+			PyErr_WarnFormat(PyExc_RuntimeWarning, 1,
+					"Producer flush failed during close: %s",
+					rd_kafka_err2str(err));
+		}
 
+        Py_RETURN_TRUE;
 }
 
 
@@ -942,27 +937,7 @@ static PyObject *Producer_exit (Handle *self, PyObject *args) {
 			      &exc_type, &exc_value, &exc_traceback))
 		return NULL;
 
-	/* Cleanup: flush pending messages and destroy producer */
-	if (self->rk) {
-		CallState_begin(self, &cs);
-
-		/* Flush any pending messages (wait indefinitely to ensure delivery) */
-		err = rd_kafka_flush(self->rk, -1);
-
-		/* Destroy the producer (even if flush had issues) */
-		rd_kafka_destroy(self->rk);
-		self->rk = NULL;
-
-		if (!CallState_end(self, &cs))
-			return NULL;
-
-		/* If flush failed, warn but don't suppress original exception */
-		if (err != RD_KAFKA_RESP_ERR_NO_ERROR) {
-			PyErr_WarnFormat(PyExc_RuntimeWarning, 1,
-					"Producer flush failed during context exit: %s",
-					rd_kafka_err2str(err));
-		}
-	}
+    Producer_close(self, (PyObject *)NULL, (PyObject *)NULL);
 
 	/* Return None to propagate any exceptions from the with block */
 	Py_RETURN_NONE;
@@ -1020,11 +995,10 @@ static PyMethodDef Producer_methods[] = {
 	  "\n"
 	},
 	{ "close", (PyCFunction)Producer_close, METH_VARARGS|METH_KEYWORDS,
-          ".. py:function:: close([timeout])\n"
+          ".. py:function:: close()\n"
           "\n"
-	  "   Request to close the producer on demand with an optional timeout.\n"
+	  "   Request to close the producer on demand.\n"
 	  "\n"
-      "  :param: float timeout: Maximum time to block (default 1 second). (Seconds)\n"
       "  :rtype: bool\n"
       "  :returns: True if producer close requested successfully, False otherwise\n"
       "\n"

tests/test_Producer.py

@@ -1414,24 +1414,9 @@ def test_producer_close():
         'message.timeout.ms': 10
     }
     producer = Producer(conf)
-    producer.produce('mytopic', value='somedata', key='a key')
+    cb_detector = {"on_delivery_called": False}
+    def on_delivery(err, msg):
+        cb_detector["on_delivery_called"] = True
+    producer.produce('mytopic', value='somedata', key='a key', callback=on_delivery)
     assert producer.close(), "The producer could not be closed on demand"
-    # Ensure no messages remain in the flush buffer after close
-    assert len(producer) == 0
-
-
-def test_producer_close_with_timeout():
-    """
-    Ensures the producer close can be requested on demand
-    """
-    conf = {
-        'debug': 'all',
-        'socket.timeout.ms': 10,
-        'error_cb': error_cb,
-        'message.timeout.ms': 10
-    }
-    producer = Producer(conf)
-    producer.produce('mytopic', value='somedata', key='a key')
-    assert producer.close(0.1), "The producer could not be closed on demand with timeout"
-    # Ensure no messages remain in the flush buffer after close
-    assert len(producer) == 0
+    assert cb_detector["on_delivery_called"], "The delivery callback should have been called by flushing during close"