Golang Client: Async producer does not send messages

[paulm17]

hey all. New to pulsar.

I have the standalone version running on docker and now trying a basic pub/sub app in golang.

package main

import (
	"context"
	"log"
	"time"

	"github.com/apache/pulsar-client-go/pulsar"
)

func main() {
	client, err := pulsar.NewClient(pulsar.ClientOptions{
		URL:               "pulsar://localhost:6650",
		OperationTimeout:  30 * time.Second,
		ConnectionTimeout: 30 * time.Second,
	})

	if err != nil {
		log.Fatalf("Could not instantiate Pulsar client: %v", err)
	}

	defer client.Close()

	// Create a producer asynchronously
	producer, err := client.CreateProducer(pulsar.ProducerOptions{
		Topic: "my-topic",
	})

	if err != nil {
		log.Fatalf("Could not create producer: %v", err)
	}

	defer producer.Close()

	// create the message
	msg := &pulsar.ProducerMessage{
		Payload: []byte("my-sync-message3"),
		// 	Key:     "my-key3", // Set the message key
	}

	// Send the message asynchronously
	producer.SendAsync(context.Background(), msg, func(msgID pulsar.MessageID, producerMessage *pulsar.ProducerMessage, err error) {
		if err != nil {
			log.Fatalf("Failed to send message: %v", err)
		}
		log.Printf("Message ID: %v", msgID)
	})

         // this works just fine!
	// for i := 0; i < 10; i++ {
	// 	if msgId, err := producer.Send(context.Background(), &pulsar.ProducerMessage{
	// 		Payload: []byte(fmt.Sprintf("hello-%d", i)),
	// 	}); err != nil {
	// 		log.Fatal(err)
	// 	} else {
	// 		log.Println("Published message: ", msgId)
	// 	}
	// }
}

I’m able to use a sync publisher, but when I try an async pub. I don’t see any messages being sent, nor any response from the function (logging).

Can someone take a look at my code and let me know if I have missed anything out? I’ve taken a look at the producer_test.go from the go-client and can’t see anything.

https://github.com/apache/pulsar-client-go/blob/6d9cbd8ba3aa08b5285cceb0a4688dad57ae3171/pulsar/producer_test.go

Thanks

[david-streamlio]

By default, pulsar producers batch their messages together before sending them to the broker. The sending of the messages occurs only after a certain threshold in terms of message count, total message size, or time has been reached.

Since you are sending just one message, my guess is that none of these thresholds are met. Can you try wrapping the sendAsync command in a for loop as well?

[jak78]

Hey @paulm17 welcome to the community! Since sendAsync is asynchronous, your program might exit before the message has a chance to be sent. Before exiting the program, you might need to wait for the message to be sent. Using WaitGroups provided by the sync package might help. This is the solution implemented by the test code you shared. - disclaimer: I'm not a golang expert ☺️. Just doing my best to help.

[merlimat]

To add on the previous responses. SendAsync() will take a callback function that will be invoked when the send operation is completed, so you can track it.

	// SendAsync a message in asynchronous mode
	// This call is blocked when the `maxPendingMessages` becomes full (default: 1000)
	// The callback will report back the message being published and
	// the eventual error in publishing
	SendAsync(context.Context, *ProducerMessage, func(MessageID, *ProducerMessage, error))

Additionally, if you want to just send a bunch of messages and then wait for completion, you can use Flush():

for i := 0; i < 100; i++ {
    producer.SendAsync(context.Background(), &pulsar.ProducerMessage{
			Payload: []byte(fmt.Sprintf("hello-%d", i)),
 	}, nil)
}

// Wait for all messages to be published
err := producer.Flush()