Added a flush method for python producer (apache#3685)

pulsar-client-cpp/python/pulsar/__init__.py

@@ -808,6 +808,15 @@ def callback(res, msg):
                               replication_clusters, disable_replication, event_timestamp)
         self._producer.send_async(msg, callback)
 
+
+    def flush(self):
+        """
+        Flush all the messages buffered in the client and wait until all messages have been
+        successfully persisted
+        """
+        self._producer.flush()
+
+
     def close(self):
         """
         Close the producer.

pulsar-client-cpp/python/src/producer.cc

@@ -56,6 +56,15 @@ void Producer_sendAsync(Producer& producer, const Message& message, py::object c
     Py_END_ALLOW_THREADS
 }
 
+void Producer_flush(Producer& producer) {
+    Result res;
+    Py_BEGIN_ALLOW_THREADS
+    res = producer.flush();
+    Py_END_ALLOW_THREADS
+
+    CHECK_RESULT(res);
+}
+
 void Producer_close(Producer& producer) {
     Result res;
     Py_BEGIN_ALLOW_THREADS
@@ -89,6 +98,9 @@ void export_producer() {
                          "\n"
                          "@param msg message to publish\n")
             .def("send_async", &Producer_sendAsync)
+            .def("flush", &Producer_flush,
+                 "Flush all the messages buffered in the client and wait until all messages have been\n"
+                         "successfully persisted\n")
             .def("close", &Producer_close)
             ;
 }

pulsar-client-cpp/python/test_producer.py

@@ -40,4 +40,5 @@
     except Exception as e:
         print("Failed to send message: %s", e)
 
+producer.flush()
 producer.close()