[Java Client] Use failPendingMessages to ensure proper cleanup

[michaeljmarshall]

Motivation

When calling ProducerImpl#closeAsync, the current cleanup relies on clearPendingMessagesWhenClose. This method does not fail any remaining messages in the batchMessageContainer. After inspecting the failPendingMessages method, it looks like it would be better to call this within the clearPendingMessagesWhenClose method. It also has more robust handling of failing the messages contained in the pendingMessages queue.

Modifications

• Update the implementation of clearPendingMessagesWhenClose to rely on failPendingMessages.
• Move already thread safe method calls, like setState outside of the synchronized block

Verifying this change

I added a test that will pass regardless of the outcome of the current mailing list thread discussing whether "close implies flush". There is also already a test that ensures that any outstanding pending messages are failed. Together, these tests ensure correct behavior.

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

This is not a breaking change.

Documentation

There is no need to update the docs for this change.

[eolivelli]

Overall looks good to me,
I left some minor comments

[eolivelli]

in the previous code "setState" was inside the synchronized block
so probably is is better to keep it inside that block.

At that point we can simply move the synchronized keyword in the signature of the method

[BewareMyPower]

I think it's okay to move setState outside the synchronized block. I think it's a problem of previous code. See connectionOpened and connectionFailed, setState is all called outside the synchronized block since it's an atomic operation. It's only called inside the synchronized block in closeAsync, which is unnecessary.

[michaeljmarshall]

It's not clear to me why the setState method was ever in a synchronized block. In looking at the git history, it has been that way since the initial import. Since setSet has its own concurrency control and since we call setState outside of synchronized blocks in many other parts of the ProducerImpl, I don't see why this update should be in the synchronized block. Let me know if you think I am missing something, @eolivelli.

[merlimat]

I don't remember the original reasoning, though if the variable state is changed within the sync block, it would appear atomic also with all the other changes in the sync block.

[michaeljmarshall]

it would appear atomic also with all the other changes in the sync block.

Yes, that is true. I'll move it back into the sync block. I wonder if these should be updated too?

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java

Lines 1504 to 1517 in da5bac9

	if (cause instanceof PulsarClientException.TopicTerminatedException) {
	setState(State.Terminated);
	synchronized (this) {
	failPendingMessages(cnx(), (PulsarClientException) cause);
	}
	producerCreatedFuture.completeExceptionally(cause);
	client.cleanupProducer(this);
	} else if (cause instanceof PulsarClientException.ProducerFencedException) {
	setState(State.ProducerFenced);
	synchronized (this) {
	failPendingMessages(cnx(), (PulsarClientException) cause);
	}
	producerCreatedFuture.completeExceptionally(cause);
	client.cleanupProducer(this);

I don't think it will matter too much because once the state is set to a failure/closed state, the producer won't accept new messages anyway.

[michaeljmarshall]

@merlimat @shoothzj - PTAL

[merlimat]

Would it make sense to also move client.cleanupProducer(this); to closeAndClearPendingMessages()?

[michaeljmarshall]

Yes, we can add that to closeAndClearPendingMessages.

[michaeljmarshall]

I put it in the sync block based on the same reasoning from the discussion on setState.

[merlimat]

I don't remember the original reasoning, though if the variable state is changed within the sync block, it would appear atomic also with all the other changes in the sync block.

[shoothzj]

Good Catch. LGTM

[eolivelli]

Lgtm

[michaeljmarshall]

/pulsarbot run-failure-checks

[michaeljmarshall]

/pulsarbot run-failure-checks

[michaeljmarshall]

/pulsarbot run-failure-checks