Test commit

Author: MSeal

src/confluent_kafka/src/Producer.c

@@ -438,6 +438,11 @@ static PyObject *Producer_close(Handle *self, PyObject *args, PyObject *kwargs)
 
         // 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");
 

tests/test_Producer.py

@@ -1,7 +1,6 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
 import gc
-import json
 import pytest
 import threading
 import time
@@ -1403,6 +1402,7 @@ def __init__(self, config):
     # Test __len__() - should return 0 for closed producer (safe, no crash)
     assert len(producer) == 0
 
+
 def test_producer_close():
     """
     Ensures the producer close can be requested on demand
@@ -1414,9 +1414,11 @@ def test_producer_close():
         'message.timeout.ms': 10
     }
     producer = Producer(conf)
-    msg = {"test": "test"}
-    producer.produce(json.dumps(msg))
+    producer.produce('mytopic', value='somedata', key='a key')
     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():
     """
@@ -1429,6 +1431,7 @@ def test_producer_close_with_timeout():
         'message.timeout.ms': 10
     }
     producer = Producer(conf)
-    msg = {"test": "test"}
-    producer.produce(json.dumps(msg))
-    assert producer.close(0.1), "The producer could not be closed on demand with timeout"
+    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