[fix][broker]Fix NonPersistentDispatcherMultipleConsumers ArrayIndexOutOfBoundsException

[Technoboy-]

Fixes #20924

Motivation

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherMultipleConsumers.java

Line 195 in 67eb3c4

Consumer consumer = TOTAL_AVAILABLE_PERMITS_UPDATER.get(this) > 0 ? getNextConsumer() : null;

The root cause is that : when call getNextConsumer (), we don't have any lock, the consumerList has changed after line-120.

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherMultipleConsumers.java

Lines 114 to 127 in 67eb3c4

	public Consumer getNextConsumer() {
	if (consumerList.isEmpty() || IS_CLOSED_UPDATER.get(this) == TRUE) {
	// abort read if no consumers are connected or if disconnect is initiated
	return null;
	}
	if (currentConsumerRoundRobinIndex >= consumerList.size()) {
	currentConsumerRoundRobinIndex = 0;
	}
	int currentRoundRobinConsumerPriority = consumerList.get(currentConsumerRoundRobinIndex).getPriorityLevel();
	// first find available-consumer on higher level unless currentIndex is not on highest level which is 0
	if (currentRoundRobinConsumerPriority != 0) {

2023-07-28T09:13:34,703+0800 [pulsar-io-12-13] WARN  org.apache.pulsar.broker.service.ServerCnx - [/10.65.17.39:65308] Got exception java.lang.ArrayIndexOutOfBoundsException: Index 2 out of bounds for length 2
	at java.base/java.util.concurrent.CopyOnWriteArrayList.elementAt(CopyOnWriteArrayList.java:385)
	at java.base/java.util.concurrent.CopyOnWriteArrayList.get(CopyOnWriteArrayList.java:398)
	at org.apache.pulsar.broker.service.AbstractDispatcherMultipleConsumers.getNextConsumerFromSameOrLowerLevel(AbstractDispatcherMultipleConsumers.java:192)
	at org.apache.pulsar.broker.service.AbstractDispatcherMultipleConsumers.getNextConsumer(AbstractDispatcherMultipleConsumers.java:137)
	at org.apache.pulsar.broker.service.nonpersistent.NonPersistentDispatcherMultipleConsumers.sendMessages(NonPersistentDispatcherMultipleConsumers.java:188)
	at org.apache.pulsar.broker.service.nonpersistent.NonPersistentTopic.lambda$publishMessage$2(NonPersistentTopic.java:202)
	at org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap$Section.forEach(ConcurrentOpenHashMap.java:554)
	at org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap.forEach(ConcurrentOpenHashMap.java:277)
	at org.apache.pulsar.broker.service.nonpersistent.NonPersistentTopic.publishMessage(NonPersistentTopic.java:196)
	at org.apache.pulsar.broker.service.Producer.publishMessageToTopic(Producer.java:281)
	at org.apache.pulsar.broker.service.Producer.publishMessage(Producer.java:194)
	at org.apache.pulsar.broker.service.ServerCnx.handleSend(ServerCnx.java:1733)
	at org.apache.pulsar.common.protocol.PulsarDecoder.channelRead(PulsarDecoder.java:222)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
	at io.netty.handler.flow.FlowControlHandler.dequeue(FlowControlHandler.java:202)

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

[codecov-commenter]

Codecov Report

All modified and coverable lines are covered by tests ✅

Comparison is base (1a4a9d9) 73.66% compared to head (f33ab8c) 73.59%.
Report is 6 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #21856      +/-   ##
============================================
- Coverage     73.66%   73.59%   -0.08%     
+ Complexity    32343    32302      -41     
============================================
  Files          1858     1858              
  Lines        138146   138141       -5     
  Branches      15141    15141              
============================================
- Hits         101763   101660     -103     
- Misses        28544    28608      +64     
- Partials       7839     7873      +34     

Flag	Coverage Δ
inttests	24.15% <86.66%> (-0.04%)	⬇️
systests	23.69% <73.33%> (-0.18%)	⬇️
unittests	72.89% <100.00%> (-0.06%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files	Coverage Δ
...er/loadbalance/impl/ModularLoadManagerWrapper.java	91.30% <100.00%> (ø)
...rg/apache/pulsar/broker/service/AbstractTopic.java	87.77% <ø> (-0.15%)	⬇️
...rg/apache/pulsar/broker/service/BrokerService.java	81.24% <100.00%> (+0.10%)	⬆️
...tent/NonPersistentDispatcherMultipleConsumers.java	68.67% <ø> (ø)
...oker/service/nonpersistent/NonPersistentTopic.java	69.91% <100.00%> (ø)
...sar/broker/service/persistent/PersistentTopic.java	78.51% <100.00%> (-0.27%)	⬇️
...java/org/apache/pulsar/common/util/FutureUtil.java	74.54% <100.00%> (+0.47%)	⬆️

... and 83 files with indirect coverage changes