pubsub: client cause memory leak

[Arc-Jung]

Thanks for stopping by to let us know something could be better!

PLEASE READ: If you have a support contract with Google, please create an issue in the support console instead of filing on GitHub. This will ensure a timely response.

Is this a client library issue or a product issue? We will only be able to assist with issues that pertain to the behaviors of this library. If the issue you're experiencing is due to the behavior of the product itself, please visit the Support page to reach the most relevant engineers.

If the support paths suggested above still do not result in a resolution, please provide the following details.

Environment details

• Programming language: Go language
• OS: Mac OS
• Language runtime version: 1.17rc2
• Package version: cloud.google.com/go/pubsub v1.17.1 ( https://pkg.go.dev/cloud.google.com/go/pubsub )

Steps to reproduce

1. I try coding to this example.

// https://cloud.google.com/pubsub/docs/samples/pubsub-subscriber-concurrency-control

func pullMsgsConcurrenyControl(w io.Writer, projectID, subID string) error {
        // projectID := "my-project-id"
        // subID := "my-sub"
        ctx := context.Background()
        client, err := pubsub.NewClient(ctx, projectID)
        if err != nil {
                return fmt.Errorf("pubsub.NewClient: %v", err)
        }
        defer client.Close()

        sub := client.Subscription(subID)
        // Must set ReceiveSettings.Synchronous to false (or leave as default) to enable
        // concurrency settings. Otherwise, NumGoroutines will be set to 1.
        sub.ReceiveSettings.Synchronous = false
        // NumGoroutines determines the number of goroutines sub.Receive will spawn to pull
        // messages.
        sub.ReceiveSettings.NumGoroutines = 16
        // MaxOutstandingMessages limits the number of concurrent handlers of messages.
        // In this case, up to 8 unacked messages can be handled concurrently.
        sub.ReceiveSettings.MaxOutstandingMessages = 8

        // Receive messages for 30 seconds.
        ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
        defer cancel()

        var counter int32

        // Receive blocks until the context is cancelled or an error occurs.
        err = sub.Receive(ctx, func(_ context.Context, msg *pubsub.Message) {
                // The message handler passed to Receive may be called concurrently
                // so it's okay to process the messages concurrently but make sure
                // to synchronize access to shared memory.
                atomic.AddInt32(&counter, 1)
                msg.Ack()
        })
        if err != nil {
                return fmt.Errorf("pubsub: Receive returned error: %v", err)
        }
        fmt.Fprintf(w, "Received %d messages\n", counter)

        return nil
}

2. I founded at this example that memory leak accident.

3. Therefore i inserted code that use garbage collection when GCP PubSub Client pull message.

4. Even after doing this solution, memory leak occur. I don't know that cause memory leak when subscription client receive GCP PubSub message.

Making sure to follow these steps will guarantee the quickest resolution possible.

Thanks!

[codyoss]

Transferring this issue to the correct repo.

[hongalex]

Thanks for reporting. What is your publish rate? Also, is there anything underneath the last line, msg.Ack, in your last screenshot?

[Arc-Jung]

I am so sorry. I checked this situation. Solution is here.

func PullMessage(w io.Writer, projectID string, subID string) error {
	ctx := context.Background()
	client, err := pubsub.NewClient(ctx, projectID, option.WithCredentialsFile("./auth_env.json"))

	if err != nil {
		log.Errorf("PubSub Credentials Error : ", err)
		return err
	}
	defer client.Close()

	sub := client.Subscription(subID)
	sub.ReceiveSettings.Synchronous = true
	sub.ReceiveSettings.NumGoroutines = 32
	sub.ReceiveSettings.MaxOutstandingMessages = 32

	ctx, cancel := context.WithCancel(ctx)
	ctx, _ = context.WithTimeout(ctx, 60*time.Second) // Receive messages for 60 seconds. If timeout then Deadline Except occur.
	defer cancel()

As my code write, difference occur with example. difference is timeout function.
I change my code that modify "ctx, _" to "ctx, cancel"

BEFORE

	sub := client.Subscription(subID)
	sub.ReceiveSettings.Synchronous = true
	sub.ReceiveSettings.NumGoroutines = 32
	sub.ReceiveSettings.MaxOutstandingMessages = 32

	ctx, cancel := context.WithCancel(ctx)
	ctx, _ = context.WithTimeout(ctx, 60*time.Second) // Receive messages for 60 seconds. If timeout then Deadline Except occur.
	defer cancel()

AFTER

	sub := client.Subscription(subID)
	sub.ReceiveSettings.Synchronous = true
	sub.ReceiveSettings.NumGoroutines = 32
	sub.ReceiveSettings.MaxOutstandingMessages = 32

	ctx, cancel = context.WithTimeout(ctx, 60*time.Second) // Receive messages for 60 seconds. If timeout then Deadline Except occur.
	defer cancel()

However, if a similar error occurs again, I will write an issue again.
Thanks.