[Client] Fix ConcurrentModificationException in sendAsync

[lhotari]

Fixes #11783

Motivation

See #11783 for details.

Modifications

Create a subclass for ArrayDeque that can postpone the adding of items that are added during a forEach iteration. To support all possible cases, this checks for nested calls too although that might not happen in this case.
The items added during a forEach loop will be added after the loop completes.

Todo

Find a way to create a failing test for the current issue.

[lhotari]

@merlimat @codelipenghui does this PR make sense? any recommendation how to add tests?

[shoothzj]

@merlimat @codelipenghui does this PR make sense? any recommendation how to add tests?

Can we do that?

1. init producer
2. close broker
3. send messages
4. start broker
5. wait for timeout, the timeout will call failPendingMessages too

and add msgs to pendingMessages does not check connect stats

[315157973]

We can't just guarantee the thread safety of the queue, so we can use a Concurrent queue. The reason why the ProducerImpl object was locked before is because there are other operations that must be guaranteed to be atomic, for example: ProducerImpl#ackReceived, we need to peek first

[lhotari]

We can't just guarantee the thread safety of the queue, so we can use a Concurrent queue. The reason why the ProducerImpl object was locked before is because there are other operations that must be guaranteed to be atomic, for example: ProducerImpl#ackReceived, we need to peek first

In this case it doesn't seem to be about thread safety and there aren't multiple threads. It's about adding an item to a queue while the same thread is in the forEach loop. That fails with a ConcurrentModificationException.

[eolivelli]

This solution looks a little bit hacky as it is fixing some specific case and sequence of method calls.

in order to fix the issue, isn't it better to handle correctly concurrent access to the pendingMessagesQueue?

if we want to go down this way I believe that at least we have to:

• create a new interface QueueThatCanBeAddedWhileExecutingForEachXX...
• create a specific subclass
• use QueueThatCanBeAddedWhileExecutingForEachXX and not Queue

otherwise it may be dangerous to modify the code that accesses the pendingMessagesQueue, as access patterns may change in the future

[shoothzj]

This solution looks a little bit hacky as it is fixing some specific case and sequence of method calls.

in order to fix the issue, isn't it better to handle correctly concurrent access to the pendingMessagesQueue?

if we want to go down this way I believe that at least we have to:

• create a new interface QueueThatCanBeAddedWhileExecutingForEachXX...
• create a specific subclass
• use QueueThatCanBeAddedWhileExecutingForEachXX and not Queue

otherwise it may be dangerous to modify the code that accesses the pendingMessagesQueue, as access patterns may change in the future

agree, QueueThatCanBeAddedWhileExecutingForEachXX are more clearness

[Anonymitaet]

Thanks for your contribution. For this PR, do we need to update docs?

(The PR template contains info about doc, which helps others know more about the changes. Can you provide doc-related info in this and future PR descriptions? Thanks)

[lhotari]

Several comments have been about using a concurrent queue implementation. I don't think that is necessary as long as the access of the queue is properly synchronized. The solution in this PR demonstrates in a unit test how the ConcurrentModificationException can happen in a single thread. When using an ordinary ArrayDeque the test would fail.

The root cause of #11783 is that while the current thread is executing the action in the forEach block, the callback code might try to add a new OpSendMsg entry in the calling thread.

Some locations:

https://github.com/apache/pulsar/blob/d86db3f4ec4fb6bd04216a123cde2fee5c43f9d9/pulsar-
client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java#L886-L891

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

Lines 1638 to 1657 in d86db3f

	pendingMessages.forEach(op -> {
	releaseCount.addAndGet(batchMessagingEnabled ? op.numMessagesInBatch: 1);
	try {
	// Need to protect ourselves from any exception being thrown in the future handler from the
	// application
	ex.setSequenceId(op.sequenceId);
	// if message is chunked then call callback only on last chunk
	if (op.totalChunks <= 1 || (op.chunkId == op.totalChunks - 1)) {
	// Need to protect ourselves from any exception being thrown in the future handler from the
	// application
	op.sendComplete(ex);
	}
	} catch (Throwable t) {
	log.warn("[{}] [{}] Got exception while completing the callback for msg {}:", topic, producerName,
	op.sequenceId, t);
	}
	client.getMemoryLimitController().releaseMemory(op.uncompressedSize);
	ReferenceCountUtil.safeRelease(op.cmd);
	op.recycle();
	});

I guess it's a valid case that after failing all pending messages the client code is allowed to access the producer in the callback to send more messages. (there are more details in #11783)

@merlimat @codelipenghui Does the approach in this PR make sense, to postpone elements added while the forEach loop is in progress?

[eolivelli]

+1

[eolivelli]

@merlimat @codelipenghui @315157973
can you please take a look ?

[315157973]

I don't know if it is a good way, waiting for others to CR. @codelipenghui @hangc0276 @merlimat

[merlimat]

The change looks good. I just left a couple of comments.

[eolivelli]

Lgtm