[go-client] support produce flush

pulsar-client-cpp/include/pulsar/c/producer.h

@@ -34,6 +34,7 @@ typedef struct _pulsar_producer pulsar_producer_t;
 
 typedef void (*pulsar_send_callback)(pulsar_result, pulsar_message_t *msg, void *ctx);
 typedef void (*pulsar_close_callback)(pulsar_result, void *ctx);
+typedef void (*pulsar_flush_callback)(pulsar_result, void *ctx);
 
 /**
  * @return the topic to which producer is publishing to
@@ -114,6 +115,11 @@ pulsar_result pulsar_producer_close(pulsar_producer_t *producer);
  */
 void pulsar_producer_close_async(pulsar_producer_t *producer, pulsar_close_callback callback, void *ctx);
 
+// Flush all the messages buffered in the client and wait until all messages have been successfully persisted.
+pulsar_result pulsar_producer_flush(pulsar_producer_t *producer);
+
+void pulsar_producer_flush_async(pulsar_producer_t *producer, pulsar_flush_callback callback, void *ctx);
+
 void pulsar_producer_free(pulsar_producer_t *producer);
 
 #pragma GCC visibility pop

pulsar-client-cpp/lib/c/c_Producer.cc

@@ -60,3 +60,11 @@ pulsar_result pulsar_producer_close(pulsar_producer_t *producer) {
 void pulsar_producer_close_async(pulsar_producer_t *producer, pulsar_close_callback callback, void *ctx) {
     producer->producer.closeAsync(boost::bind(handle_result_callback, _1, callback, ctx));
 }
+
+pulsar_result pulsar_producer_flush(pulsar_producer_t *producer) {
+    return (pulsar_result)producer->producer.flush();
+}
+
+void pulsar_producer_flush_async(pulsar_producer_t *producer, pulsar_close_callback callback, void *ctx) {
+    producer->producer.flushAsync(boost::bind(handle_result_callback, _1, callback, ctx));
+}

pulsar-client-go/pulsar/c_go_pulsar.h

@@ -49,6 +49,12 @@ static inline void _pulsar_client_create_producer_async(pulsar_client_t *client,
     pulsar_client_create_producer_async(client, topic, conf, pulsarCreateProducerCallbackProxy, ctx);
 }
 
+void pulsarProducerFlushCallbackProxy(pulsar_result result, void *ctx);
+
+static inline void _pulsar_producer_flush_async(pulsar_producer_t *producer, void *ctx){
+    pulsar_producer_flush_async(producer, pulsarProducerFlushCallbackProxy, ctx);
+}
+
 void pulsarProducerCloseCallbackProxy(pulsar_result result, void *ctx);
 
 static inline void _pulsar_producer_close_async(pulsar_producer_t *producer, void *ctx) {

pulsar-client-go/pulsar/c_producer.go

@@ -238,3 +238,28 @@ func pulsarProducerCloseCallbackProxy(res C.pulsar_result, ctx unsafe.Pointer) {
 		callback(nil)
 	}
 }
+
+func (p *producer) Flush() error {
+	f := make(chan error)
+	p.FlushAsync(func(err error) {
+		f <- err
+		close(f)
+	})
+	return <-f
+}
+
+func (p *producer) FlushAsync(callback func(error)) {
+	C._pulsar_producer_flush_async(p.ptr, savePointer(callback))
+}
+
+
+//export pulsarProducerFlushCallbackProxy
+func pulsarProducerFlushCallbackProxy(res C.pulsar_result, ctx unsafe.Pointer) {
+	callback := restorePointer(ctx).(func(error))
+
+	if res != C.pulsar_result_Ok {
+		callback(newError(res, "Failed to flush Producer"))
+	} else {
+		callback(nil)
+	}
+}

pulsar-client-go/pulsar/producer.go

@@ -176,6 +176,10 @@ type Producer interface {
 	// return the last sequence id published by this producer.
 	LastSequenceID() int64
 
+	// Flush all the messages buffered in the client and wait until all messages have been successfully
+	// persisted.
+	Flush() error
+
 	// Close the producer and releases resources allocated
 	// No more writes will be accepted from this producer. Waits until all pending write request are persisted. In case
 	// of errors, pending writes will not be retried.

pulsar-client-go/pulsar/producer_test.go

@@ -25,6 +25,9 @@ import (
 	"github.com/stretchr/testify/assert"
 	"testing"
 	"time"
+
+	log "github.com/apache/pulsar/pulsar-client-go/logutil"
+	"github.com/stretchr/testify/assert"
 )
 
 func TestInvalidURL(t *testing.T) {
@@ -205,3 +208,51 @@ func TestProducerZstd(t *testing.T) {
 		}
 	}
 }
+
+func TestProducer_Flush(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: "pulsar://localhost:6650",
+	})
+	assert.Nil(t, err)
+	defer client.Close()
+
+	topicName := "test-flush-in-producer"
+	subName := "subscription-name"
+
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic: topicName,
+		Properties: map[string]string{
+			"producer-name": "test-producer-name",
+			"producer-id":   "test-producer-id",
+		},
+	})
+	assert.Nil(t, err)
+	defer producer.Close()
+
+	consumer, err := client.Subscribe(ConsumerOptions{
+		Topic:            topicName,
+		SubscriptionName: subName,
+		Properties: map[string]string{
+			"consumer-name": "test-consumer-name",
+			"consumer-id":   "test-consumer-id",
+		},
+	})
+	assert.Nil(t, err)
+	defer consumer.Close()
+
+	ctx := context.Background()
+	for i := 0; i < 10; i++ {
+		// Create a different message to send asynchronously
+		asyncMsg := ProducerMessage{
+			Payload: []byte(fmt.Sprintf("async-message-%d", i)),
+		}
+		// Attempt to send the message asynchronously and handle the response
+		producer.SendAsync(ctx, asyncMsg, func(msg ProducerMessage, err error) {
+			if err != nil {
+				log.Fatal(err)
+			}
+			fmt.Printf("Message %s successfully published", msg.Payload)
+		})
+		producer.Flush()
+	}
+}

pulsar-client-go/pulsar/reader_test.go

@@ -46,7 +46,7 @@ func TestReaderConnectError(t *testing.T) {
 	assert.Nil(t, reader)
 	assert.NotNil(t, err)
 
-	assert.Equal(t, err.(*Error).Result(), ConnectError);
+	assert.Equal(t, err.(*Error).Result(), ConnectError)
 }
 
 func TestReader(t *testing.T) {