Bookkeeper triggered exception IndexOutOfBoundsException when consumer use Key_Shared and producer enableBatching/set compressionType

[tabalt]

Describe the bug

Bookkeeper triggered exception IndexOutOfBoundsException when consumer use Key_Shared and producer enable Batching/set compressionType

17:05:50.237 [bookkeeper-ml-workers-OrderedExecutor-3-0] ERROR org.apache.bookkeeper.common.util.SafeRunnable - Unexpected throwable caught
java.lang.IndexOutOfBoundsException: readerIndex: 89, writerIndex: 2023521513 (expected: 0 <= readerIndex <= writerIndex <= capacity(398))
	at io.netty.buffer.AbstractByteBuf.checkIndexBounds(AbstractByteBuf.java:112) ~[netty-buffer-4.1.48.Final.jar:4.1.48.Final]
	at io.netty.buffer.AbstractByteBuf.writerIndex(AbstractByteBuf.java:135) ~[netty-buffer-4.1.48.Final.jar:4.1.48.Final]
	at org.apache.pulsar.common.protocol.Commands.deSerializeSingleMessageInBatch(Commands.java:1707) ~[pulsar-common.jar:2.6.0]
	at org.apache.pulsar.broker.service.AbstractBaseDispatcher.peekStickyKey(AbstractBaseDispatcher.java:161) ~[pulsar-broker.jar:2.6.0]
	at org.apache.pulsar.broker.service.persistent.PersistentStickyKeyDispatcherMultipleConsumers.sendMessagesToConsumers(PersistentStickyKeyDispatcherMultipleConsumers.java:145) ~[pulsar-broker.jar:2.6.0]
	at org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers.readEntriesComplete(PersistentDispatcherMultipleConsumers.java:476) ~[pulsar-broker.jar:2.6.0]
	at org.apache.bookkeeper.mledger.impl.OpReadEntry.lambda$checkReadCompletion$2(OpReadEntry.java:152) ~[managed-ledger.jar:2.6.0]
	at org.apache.bookkeeper.mledger.util.SafeRun$1.safeRun(SafeRun.java:32) ~[managed-ledger.jar:2.6.0]
	at org.apache.bookkeeper.common.util.SafeRunnable.run(SafeRunnable.java:36) [bookkeeper-common-4.10.0.jar:4.10.0]
	at org.apache.bookkeeper.common.util.OrderedExecutor$TimedRunnable.run(OrderedExecutor.java:203) [bookkeeper-common-4.10.0.jar:4.10.0]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [?:1.8.0_151]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [?:1.8.0_151]
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) [netty-common-4.1.48.Final.jar:4.1.48.Final]
	at java.lang.Thread.run(Thread.java:748) [?:1.8.0_151]

To Reproduce

Producer<byte[]> producer = client.newProducer()
                    .blockIfQueueFull(true)
                    .enableBatching(true)
                    .batchingMaxPublishDelay(50, TimeUnit.MILLISECONDS)
                    .batchingMaxMessages(1024)
                    .compressionType(CompressionType.ZLIB)
                    .messageRoutingMode(MessageRoutingMode.RoundRobinPartition)
                    .maxPendingMessages(1024)
                    .sendTimeout(0, TimeUnit.SECONDS)
                    .topic("my-topic")
                    .create();

Consumer consumer = client.newConsumer()
                .topic("my-topic")
                .subscriptionName("my-subscription")
                .consumerName("my-consumer-name")
                .subscriptionType(SubscriptionType.Key_Shared)
                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
                .messageListener(new PulsarMessageListener())
                .receiverQueueSize(5000)
                .ackTimeout(3000, TimeUnit.MILLISECONDS)
                .maxTotalReceiverQueueSizeAcrossPartitions(50000)
                .subscribe();

Desktop (please complete the following information):

• OS: MacOS

[sijie]

@tabalt This issue has been fixed by #7416 and released as part of 2.6.1 release. Which version of Pulsar broker are you using?

[alexku7]

@tabalt @sijie
I am prety sure that this is the same issue like this #8024 (our issue:) )
it has been fixed but not still released.

@sijie I think this bug so much severe that it worth to release 2.6.2 version and not to wait for 2.7 :) for you consideration

[sijie]

@alexku7 the issue has been labeled for 2.6.2 release. We will kick off the 2.6.2 release in this week or so.

[tabalt]

@sijie I tried the 2.6.1 release, the exception not reproduce, thanks very much!