[Fix] Multiple calls to client.Close causes panic

[crossoverJie]

Fixes #1026

Motivation

Multiple calls to client.Close causes panic.

Modifications

Add closeOnce sync.Once

Verifying this change

• Make sure that the change passes the CI checks.

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): (no)
• The public API: (no)
• The schema: (no)
• The default values of configurations: (no)
• The wire protocol: (no)

Documentation

• Does this pull request introduce a new feature? (no)

[hezhangjian]

This PR enables the 'close' method to be re-entrant. However, I'm not quite seeing the value in making the 'close' method re-entrant. Could you provide any scenarios or use cases where this would be necessary?

[crossoverJie]

This is just defensive programming, the consumer's close logic also has similar usage.

pulsar-client-go/pulsar/consumer_multitopic.go

Lines 256 to 270 in 5f8df27

	c.closeOnce.Do(func() {
	var wg sync.WaitGroup
	wg.Add(len(c.consumers))
	for _, con := range c.consumers {
	go func(consumer Consumer) {
	defer wg.Done()
	consumer.Close()
	}(con)
	}
	wg.Wait()
	close(c.closeCh)
	c.client.handlers.Del(c)
	c.dlq.close()
	c.rlq.close()
	})

[tisonkun]

Generally looks good.

My concern can be that it may not be an idiom among the code. So if we change the flavor randomly it can introduce noise. But I think this direction is good to go.

[tisonkun]

What if we cause the internal closes idempotent instead of hacky wrapping an Once outer?

[crossoverJie]

What if we cause the internal closes idempotent instead of hacky wrapping an Once outer?

https://github.com/apache/pulsar/blob/a3bca685a4ef0916462228e035f77fdbf295653d/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java#L735-L739

I looked at the Java SDK's close logic, it uses an internal state to maintain the client state, and the Once principle used here is similar.

[YanshuoH]

This PR enables the 'close' method to be re-entrant. However, I'm not quite seeing the value in making the 'close' method re-entrant. Could you provide any scenarios or use cases where this would be necessary?

I might add a scenario to this improvement.

For example, our app may instantiate multiple instances that implement io.Closer, we might do something like:

• create a server that will create multiple rpc instances (including pulsar.Client) inside.
• server.Run func that has a defer cleanup() within. That cleanup would try to close the rpc instances.
• a defer server.Close() func that calls cleanup underling that will be used in the main file.

In Go, we expect Close that can be called multiple times, so the above workflow is designed to make sure all rpc instances are closed properly without leaving any openfd or occupying other resources.

That is the opportunity that I've found out #1026.

[RobertIndie]

@tisonkun @Shoothzj Could you take a look at this? Does the explanation make sense to you?

[hezhangjian]

I believe this modification is feasible. Regardless of whether or not it supports reentrancy, and whether reentrancy supports concurrency, we can maintain consistent behavior across the entire code repository.

[tisonkun]

Thanks for your contribution!