[Bug] Producer produced a message that consumer unable to receive (throw from ClientCnx.handleMessage)

### Search before asking

- [X] I searched in the [issues](https://github.com/apache/pulsar/issues) and found nothing similar.

### Version

- producer 2.8.4 (java 11)
- consumer 2.8.4, 2.8.1, 3.0.1 all failed (java 11)
- broker 2.8.1.x (java 8)

### Minimal reproduce step

Producer application is under memory pressure (OOM occurs), I've write a simple program to simulate `ClientCnx.handleMessage` to process broken message.

### What did you expect to see?

- Producers or brokers should never generate or dispatch a corrupted message to consumers.
- If the above is unavoidable, consumers should be robust enough to handle receiving a corrupted message (perhaps by simply discarding it?)

### What did you see instead?

After a problematic application sent a large volume of messages to the broker, several brokers crashed due to Out-Of-Memory (OOM) errors related to direct memory.

Subsequently, we observed that one of our consumer applications consistently encountered errors while acknowledging messages:
```
org.apache.pulsar.client.api.PulsarClientException$ConnectException: Consumer connect fail! consumer state:Connecting
at org.apache.pulsar.client.api.PulsarClientException.unwrap(PulsarClientException.java:973)
at org.apache.pulsar.client.impl.ConsumerBase.acknowledge(ConsumerBase.java:317)
at org.apache.pulsar.client.impl.ConsumerBase.acknowledge(ConsumerBase.java:305)
```
There are some other error messages on the same consumer instance:
```
2024-01-22T13:22:02.240Z [pulsar-client-io-6-1] WARN org.apache.pulsar.client.impl.ClientCnx - \
[xx.xx.92.10/xx.xx.92.10:6650] Got exception java.lang.IndexOutOfBoundsException: \
readerIndex(92) + length(4) exceeds writerIndex(92): PooledSlicedByteBuf(ridx: 92, widx: 92, cap: 92/92, unwrapped: PooledUnsafeDirectByteBuf(ridx: 96, widx: 2048, cap: 2048))
at org.apache.pulsar.shade.io.netty.buffer.AbstractByteBuf.checkReadableBytes0(AbstractByteBuf.java:1442)
at org.apache.pulsar.shade.io.netty.buffer.AbstractByteBuf.readInt(AbstractByteBuf.java:809)
at org.apache.pulsar.shade.io.netty.buffer.AbstractByteBuf.readUnsignedInt(AbstractByteBuf.java:825)
at org.apache.pulsar.common.protocol.Commands.deSerializeSingleMessageInBatch(Commands.java:1634)
at org.apache.pulsar.client.impl.ConsumerImpl.receiveIndividualMessagesFromBatch(ConsumerImpl.java:1261)
at org.apache.pulsar.client.impl.ConsumerImpl.messageReceived(ConsumerImpl.java:1100)
at org.apache.pulsar.client.impl.ClientCnx.handleMessage(ClientCnx.java:451)
at org.apache.pulsar.common.protocol.PulsarDecoder.channelRead(PulsarDecoder.java:186)
at org.apache.pulsar.shade.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.pulsar.shade.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.pulsar.shade.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at org.apache.pulsar.shade.io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:327)
at org.apache.pulsar.shade.io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:314)
at org.apache.pulsar.shade.io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:435)
at org.apache.pulsar.shade.io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:279)
at org.apache.pulsar.shade.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.pulsar.shade.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.pulsar.shade.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at org.apache.pulsar.shade.io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
at org.apache.pulsar.shade.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.pulsar.shade.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.pulsar.shade.io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
at org.apache.pulsar.shade.io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:800)
at org.apache.pulsar.shade.io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:487)
at org.apache.pulsar.shade.io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:385)
at org.apache.pulsar.shade.io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:995)
at org.apache.pulsar.shade.io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
at org.apache.pulsar.shade.io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
at java.lang.Thread.run(Thread.java:748)
```

I've seen that consumer client keep sent `RST` to close underling tcp connection through packet capture:


I believe this is the expected behavior when `ClientCnx` encounters unknown exceptions:
https://github.com/apache/pulsar/blob/480a229d8fb637232bd37b7a72dfe8a298c72000/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java#L343-L357

I've extracted the specific broken message from the capture, and write a simple program to read this protocol `MESSAGE` to simulate `ClientCnx.handleMessage`:
```java
import com.google.common.io.Files;
import org.apache.pulsar.common.api.proto.BaseCommand;
import org.apache.pulsar.common.api.proto.CommandMessage;
import org.apache.pulsar.common.api.proto.MessageIdData;
import org.apache.pulsar.common.api.proto.MessageMetadata;
import org.apache.pulsar.common.protocol.Commands;
import org.apache.pulsar.shade.io.netty.buffer.ByteBuf;
import org.apache.pulsar.shade.io.netty.buffer.ByteBufAllocator;

import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;

class Scratch {
public static void main(String[] args) throws IOException {
File file = new File("broken-message-extracted.bin");
byte[] data = Files.asByteSource(file).read();
ByteBuffer msg = ByteBuffer.wrap(data);
int length = msg.getInt();
byte[] dst = new byte[length];
msg.get(dst);
ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer().writeBytes(dst);
BaseCommand cmd = new BaseCommand();
cmd.parseFrom(buffer, (int) buffer.readUnsignedInt());
// print "MESSAGE"
System.out.println(cmd.getType());
CommandMessage message = cmd.getMessage();
MessageIdData messageId = message.getMessageId();
// print "ledgerId: 115091324, entryId: 1644, partition: 1"
System.out.printf("ledgerId: %d, entryId: %d, partition: %d\n", messageId.getLedgerId(), messageId.getEntryId(), messageId.getPartition());
MessageMetadata messageMetadata = Commands.parseMessageMetadata(buffer);
// print "sequenceId: -1, highestSequenceId: -1"
System.out.printf("sequenceId: %d, highestSequenceId: %d\n", messageMetadata.getSequenceId(), messageMetadata.getHighestSequenceId());
// print "compression: NONE, uncompressedSize: 0"
System.out.printf("compression: %s, uncompressedSize: %d\n", messageMetadata.getCompression(), messageMetadata.getUncompressedSize());
// print "publishTime: 1705902570650, eventTime: 0"
System.out.printf("publishTime: %d, eventTime: %d\n", messageMetadata.getPublishTime(), messageMetadata.getEventTime());
// print "producerName: pulsar-idc-a-464-428776, numMessagesInBatch: 1"
System.out.printf("producerName: %s, numMessagesInBatch: %d\n", messageMetadata.getProducerName(), messageMetadata.getNumMessagesInBatch());
// throw java.lang.IndexOutOfBoundsException: readerIndex(92) + length(4) exceeds writerIndex(92): PooledUnsafeDirectByteBuf(ridx: 92, widx: 92, cap: 256)
buffer.readUnsignedInt();
}
}
```
Here is `broken-message-extracted.bin` in base64:
`AAAAXAAAABIICUoOCDwSCgj8zvA2EOwMGAEOAW361skAAAA8ChdwdWxzYXItaWRjLWEtNDY0LTQyODc3NhD///////////8BGJr53f3SMVgBggEAwAH///////////8B`

Looks like it is a batch message with batchSize=1, but the `SingleMessageMetadata` and payload is missing. Therefore, the message decoding process will throw an `IndexOutOfBoundsException` in `Commands.deSerializeSingleMessageInBatch`:
https://github.com/apache/pulsar/blob/480a229d8fb637232bd37b7a72dfe8a298c72000/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java#L1876-L1885

### Anything else?

More context:
- Unloading the topic and restarting the broker does not resolve this issue. The only workaround seems to be skipping the broken message.
- The topic have 3 partitions, there are only one broken message found on partition-1
- Producer application have memory pressure, found some OOM error in the log around the `publishTime` of broken message
```
Caused by: java.lang.OutOfMemoryError: Java heap space
at java.base/java.util.Arrays.copyOf(Arrays.java:3689)
at java.base/java.util.ArrayList.grow(ArrayList.java:237)
at java.base/java.util.ArrayList.grow(ArrayList.java:242)
at java.base/java.util.ArrayList.add(ArrayList.java:485)
at java.base/java.util.ArrayList.add(ArrayList.java:498)
at org.apache.pulsar.client.impl.BatchMessageContainerImpl.add(BatchMessageContainerImpl.java:108)
at org.apache.pulsar.client.impl.BatchMessageKeyBasedContainer.add(BatchMessageKeyBasedContainer.java:52)
at org.apache.pulsar.client.impl.ProducerImpl.serializeAndSendMessage(ProducerImpl.java:548)
at org.apache.pulsar.client.impl.ProducerImpl.sendAsync(ProducerImpl.java:478)
at org.apache.pulsar.client.impl.ProducerImpl.internalSendAsync(ProducerImpl.java:310)
at org.apache.pulsar.client.impl.ProducerImpl.internalSendWithTxnAsync(ProducerImpl.java:381)
at org.apache.pulsar.client.impl.PartitionedProducerImpl.internalSendWithTxnAsync(PartitionedProducerImpl.java:191)
at org.apache.pulsar.client.impl.PartitionedProducerImpl.internalSendAsync(PartitionedProducerImpl.java:167)
at org.apache.pulsar.client.impl.TypedMessageBuilderImpl.sendAsync(TypedMessageBuilderImpl.java:103)
```
- There is a related issue #11930 closed as stale

### Are you willing to submit a PR?

- [ ] I'm willing to submit a PR!

Start with ClientCnx.handleMessage and Commands.deSerializeSingleMessageInBatch at the linked revisions, then run the supplied Scratch reproducer against the provided broken message. Trace how ConsumerImpl.receiveIndividualMessagesFromBatch handles the incomplete batch and determine whether the intended change is producer/broker prevention or consumer-side handling. Done means the corrupted-message scenario has defined behavior and regression coverage.

Written by the indexing model from the issue text.