[fix][client] Fixed stuck issues caused by an exception encountered in the receiveIndividualMessagesFromBatch()

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

@@ -1786,6 +1786,7 @@ void receiveIndividualMessagesFromBatch(BrokerEntryMetadata brokerEntryMetadata,
 
         SingleMessageMetadata singleMessageMetadata = new SingleMessageMetadata();
         int skippedMessages = 0;
+        int processedMessages = 0;
         try {
             for (int i = 0; i < batchSize; ++i) {
                 final MessageImpl<T> message = newSingleMessage(i, batchSize, brokerEntryMetadata, msgMetadata,
@@ -1815,13 +1816,15 @@ void receiveIndividualMessagesFromBatch(BrokerEntryMetadata brokerEntryMetadata,
                     continue;
                 }
                 executeNotifyCallback(message);
+                processedMessages++;
             }
             if (ackBitSet != null) {
                 ackBitSet.recycle();
             }
-        } catch (IllegalStateException e) {
-            log.warn("[{}] [{}] unable to obtain message in batch", subscription, consumerName, e);
-            discardCorruptedMessage(messageId, cnx, ValidationError.BatchDeSerializeError);
+        } catch (IllegalStateException | IllegalArgumentException e) {
+            // For IllegalArgumentException see PR: https://github.com/apache/pulsar/pull/24061
+            discardCorruptedBatchMessage(messageId, cnx, batchSize,
+                    skippedMessages, processedMessages, ValidationError.BatchDeSerializeError);
         }
 
         if (deadLetterPolicy != null && possibleSendToDeadLetterTopicMessages != null) {
@@ -2153,6 +2156,38 @@ private void discardCorruptedMessage(MessageIdData messageId, ClientCnx currentC
         discardMessage(messageId, currentCnx, validationError, 1);
     }
 
+    /**
+     * When batch index ack is enabled, ack the messages that failed to deserialize by their index,
+     * while keeping successfully enqueued messages unacknowledged to avoid message loss.
+     */
+    private void discardCorruptedBatchMessage(MessageIdData messageId, ClientCnx currentCnx,
+            int batchSize, int skipped, int processed, ValidationError validationError) {
+        log.error("[{}] [{}] Discarding corrupted batch messages with batch index ack at {}:{}, "
+                        + "batchSize={}, skipped={}, processed={}, exception={}",
+                subscription, consumerName, messageId.getLedgerId(), messageId.getEntryId(),
+                batchSize, skipped, processed, validationError);
+        BitSetRecyclable ackBitSet = null;
+        int corruptedStartIndex = skipped + processed;
+        if (conf.isBatchIndexAckEnabled()) {
+            // When batch index ack is enabled, only ack the messages that failed to deserialize.
+            // Messages that have been successfully enqueued remain unacknowledged,
+            // waiting for the user to consume and acknowledge them normally.
+            ackBitSet = BitSetRecyclable.create();
+            ackBitSet.set(0, batchSize);
+            for (int i = corruptedStartIndex; i < batchSize; i++) {
+                ackBitSet.clear(i);
+            }
+        }
+        ByteBuf cmd = Commands.newAck(consumerId, messageId.getLedgerId(), messageId.getEntryId(),
+                ackBitSet, AckType.Individual, validationError, Collections.emptyMap(), -1);
+        currentCnx.ctx().writeAndFlush(cmd, currentCnx.ctx().voidPromise());
+        if (ackBitSet != null) {
+            ackBitSet.recycle();
+        }
+        increaseAvailablePermits(currentCnx, batchSize - corruptedStartIndex);
+        stats.incrementNumReceiveFailed();
+    }
+
     private void discardMessage(MessageIdData messageId, ClientCnx currentCnx, ValidationError validationError,
             int batchMessages) {
         ByteBuf cmd = Commands.newAck(consumerId, messageId.getLedgerId(), messageId.getEntryId(), null,