[Producer] respect context cancellation in Flush

[jayshrivastava]

Motivation

The producer's Flush method does not respect context cancellation. If the caller's context get's cancelled, it will have to wait for the producer to finish flushing.

Modifications

This change adds a FlushWithCtx method which takes a context and selects on two channels.

Verifying this change

• Make sure that the change passes the CI checks.

This change is already covered by existing tests. See code changes for relevant tests.

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

If yes was chosen, please highlight the changes

• 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
• If yes, how is the feature documented? not applicable
• If a feature is not applicable for documentation, explain why? It's a very trivial change. It adds a common pattern in Go.
• If a feature is not documented yet in this PR, please create a followup issue for adding the documentation

[jayshrivastava]

@RobertIndie Would you mind taking a look at this please?

[merlimat]

While I agree that we should have taking the context in this method, right now this would be a breaking API change.

I'd say to add a new FlushWithContext() method and, perhaps, mark the other one as deprecated.

[jayshrivastava]

Just pushed an update. Lmk how it looks!

Btw, do you know if the library will support shutting down via context cancellation in the future? There's some examples of where we don't do it:

pulsar-client-go/pulsar/producer_partition.go

Line 491 in efd9806

func (p *partitionProducer) runEventsLoop() {

pulsar-client-go/pulsar/producer_partition.go

Lines 1470 to 1481 in efd9806

	func (p *partitionProducer) Close() {
	if p.getProducerState() != producerReady {
	// Producer is closing
	return
	}
	cp := &closeProducer{doneCh: make(chan struct{})}
	p.cmdChan <- cp
	// wait for close producer request to complete
	<-cp.doneCh
	}

As long as things don't take too long to shut down, I suppose it's fine. Ideally, if the context is cancelled, you don't want to wait for a clean/graceful shutdown where we wait for the done channel. We want all the goroutines to terminate right where they are and immediately returned.

[jayshrivastava]

I would be happy to file an issue if there is not one already.

[jayshrivastava]

I can't implement CloseWithCtx such as in this PR because goroutines might leak (the call to close would terminate before the signal is given on the doneCh, indicating that things have shut down). I think it would be a bigger change.