[GitHub] [pulsar] rdhabalia commented on a change in pull request #4329: Moved entries filtering from consumer to dispatcher

[GitBox <gi...@apache.org>]

rdhabalia commented on a change in pull request #4329: Moved entries filtering from consumer to dispatcher
URL: https://github.com/apache/pulsar/pull/4329#discussion_r286160135
 
 

 ##########
 File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java
 ##########
 @@ -284,70 +249,6 @@ private void incrementUnackedMessages(int ackedMessages) {
         }
     }
 
-    public static int getBatchSizeforEntry(ByteBuf metadataAndPayload, Subscription subscription, long consumerId) {
-        try {
-            // save the reader index and restore after parsing
-            metadataAndPayload.markReaderIndex();
-            PulsarApi.MessageMetadata metadata = Commands.parseMessageMetadata(metadataAndPayload);
-            metadataAndPayload.resetReaderIndex();
-            int batchSize = metadata.getNumMessagesInBatch();
-            metadata.recycle();
-            if (log.isDebugEnabled()) {
-                log.debug("[{}] [{}] num messages in batch are {} ", subscription, consumerId, batchSize);
-            }
-            return batchSize;
-        } catch (Throwable t) {
-            log.error("[{}] [{}] Failed to parse message metadata", subscription, consumerId, t);
-        }
-        return -1;
-    }
-
-    void updatePermitsAndPendingAcks(final List<Entry> entries, SendMessageInfo sentMessages) throws PulsarServerException {
-        int permitsToReduce = 0;
-        Iterator<Entry> iter = entries.iterator();
-        boolean unsupportedVersion = false;
-        long totalReadableBytes = 0;
-        boolean clientSupportBatchMessages = cnx.isBatchMessageCompatibleVersion();
-        while (iter.hasNext()) {
-            Entry entry = iter.next();
-            ByteBuf metadataAndPayload = entry.getDataBuffer();
-            int batchSize = getBatchSizeforEntry(metadataAndPayload, subscription, consumerId);
-            if (batchSize == -1) {
-                // this would suggest that the message might have been corrupted
-                iter.remove();
-                PositionImpl pos = (PositionImpl) entry.getPosition();
-                entry.release();
-                subscription.acknowledgeMessage(Collections.singletonList(pos), AckType.Individual, Collections.emptyMap());
-                continue;
-            }
-            if (pendingAcks != null) {
-                pendingAcks.put(entry.getLedgerId(), entry.getEntryId(), batchSize, 0);
-            }
-            // check if consumer supports batch message
-            if (batchSize > 1 && !clientSupportBatchMessages) {
 
 Review comment:
   doesn't look good by removing this logic.. we still want to keep the check if consumer doesn't support batch message.

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services