Fix parsing metadata

datastax · May 7, 2024 · f73df6c · f73df6c
1 parent 17bf86c
commit f73df6c
Showing 1 changed file with 13 additions and 3 deletions.
diff --git a/...ar-jms-filters/src/main/java/com/datastax/oss/pulsar/jms/selectors/JMSPublishFilters.java b/...ar-jms-filters/src/main/java/com/datastax/oss/pulsar/jms/selectors/JMSPublishFilters.java
@@ -53,7 +53,7 @@
 import org.apache.pulsar.common.api.proto.CommandAck;
 import org.apache.pulsar.common.api.proto.MessageMetadata;
 import org.apache.pulsar.common.intercept.InterceptException;
-import org.apache.pulsar.common.protocol.Commands;
+import org.jetbrains.annotations.NotNull;
 
 @Slf4j
 public class JMSPublishFilters implements BrokerInterceptor {
@@ -284,8 +284,7 @@ private void filterAndAckMessage(
       // this operation has been enqueued before the broker shutdown
       return;
     }
-    MessageMetadata messageMetadata = new MessageMetadata();
-    Commands.parseMessageMetadata(messageMetadataUnparsed, messageMetadata);
+    MessageMetadata messageMetadata = getMessageMetadata(messageMetadataUnparsed);
     long now = System.nanoTime();
     try {
       FilterContext filterContext = new FilterContext();
@@ -317,6 +316,17 @@ private void filterAndAckMessage(
     }
   }
 
+  @NotNull
+  private static MessageMetadata getMessageMetadata(ByteBuf messageMetadataUnparsed) {
+    MessageMetadata messageMetadata = new MessageMetadata();
+    synchronized (messageMetadataUnparsed) {
+      int index = messageMetadataUnparsed.readerIndex();
+      messageMetadata.parseFrom(messageMetadataUnparsed, messageMetadataUnparsed.readableBytes());
+      messageMetadataUnparsed.readerIndex(index);
+    }
+    return messageMetadata;
+  }
+
   private static void scheduleOnDispatchThread(Subscription subscription, Runnable runnable) {
     try {
       Dispatcher dispatcher = subscription.getDispatcher();