[feature] AS-5202: Added filteredconfig for browser

pulsar-jms/src/main/java/com/datastax/oss/pulsar/jms/ConsumerConfiguration.java

@@ -21,6 +21,7 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
 import org.apache.pulsar.client.api.DeadLetterPolicy;
 import org.apache.pulsar.client.api.RedeliveryBackoff;
 import org.apache.pulsar.client.api.Schema;
@@ -31,6 +32,36 @@ final class ConsumerConfiguration {
   static ConsumerConfiguration DEFAULT =
       new ConsumerConfiguration(Collections.emptyMap(), null, null, null, null);
 
+  /**
+   * Set of properties that are specific to Pulsar Consumer API and not applicable to Pulsar Reader
+   * API (used by QueueBrowser). These properties will be filtered out when creating a
+   * browser-compatible configuration.
+   *
+   * <p>QueueBrowsers use Pulsar Reader API for read-only message browsing, while Consumers use
+   * Pulsar Consumer API for message consumption with acknowledgment. Properties related to
+   * acknowledgment, subscriptions, and consumer-specific behavior are not applicable to readers.
+   */
+  private static final Set<String> BROWSER_INCOMPATIBLE_PROPERTIES =
+      Set.of(
+          // Acknowledgment-related properties (browsers don't acknowledge messages)
+          "batchIndexAckEnabled",
+          "ackTimeout",
+          "ackTimeoutTickTime",
+          "acknowledgmentGroupTime",
+          "negativeAckRedeliveryDelay",
+          "retryEnable",
+
+          // Subscription-related properties (set programmatically for consumers, not for readers)
+          "subscriptionType",
+          "subscriptionMode",
+          "subscriptionInitialPosition",
+
+          // Consumer-specific properties (not applicable to readers)
+          "priorityLevel",
+          "maxTotalReceiverQueueSizeAcrossPartitions",
+          "autoUpdatePartitions",
+          "autoUpdatePartitionsInterval");
+
   private final Map<String, Object> consumerConfiguration;
   private Schema<?> consumerSchema;
   private DeadLetterPolicy deadLetterPolicy;
@@ -100,6 +131,51 @@ ConsumerConfiguration applyDefaults(ConsumerConfiguration defaultConsumerConfigu
         mergedAckTimeoutRedeliveryBackoff);
   }
 
+  /**
+   * Creates a browser-compatible configuration by filtering out consumer-specific properties that
+   * are not applicable to the Pulsar Reader API.
+   *
+   * <p>QueueBrowsers use Pulsar Reader API for read-only message browsing, while Consumers use
+   * Pulsar Consumer API for message consumption with acknowledgment. Many consumer properties
+   * related to acknowledgment, subscriptions, and consumer behavior are not applicable to readers
+   * and should be filtered out to avoid potential compatibility issues.
+   *
+   * <p>This method filters out the following categories of properties:
+   *
+   * <ul>
+   *   <li><b>Acknowledgment-related:</b> batchIndexAckEnabled, ackTimeout, ackTimeoutTickTime,
+   *       acknowledgmentGroupTime, negativeAckRedeliveryDelay, retryEnable
+   *   <li><b>Subscription-related:</b> subscriptionType, subscriptionMode,
+   *       subscriptionInitialPosition
+   *   <li><b>Consumer-specific:</b> priorityLevel, maxTotalReceiverQueueSizeAcrossPartitions,
+   *       autoUpdatePartitions, autoUpdatePartitionsInterval
+   * </ul>
+   *
+   * <p>Additionally, deadLetterPolicy and redelivery backoff configurations are set to null as
+   * browsers don't acknowledge or redeliver messages.
+   *
+   * <p>Common properties that work with both Consumer and Reader APIs (such as receiverQueueSize,
+   * cryptoKeyReader, readCompacted, etc.) are preserved in the filtered configuration.
+   *
+   * @return A new ConsumerConfiguration with browser-incompatible properties filtered out, suitable
+   *     for use with Pulsar Reader API
+   */
+  public ConsumerConfiguration forBrowser() {
+    Map<String, Object> filteredConfig = new HashMap<>(consumerConfiguration);
+
+    // Remove all browser-incompatible properties
+    BROWSER_INCOMPATIBLE_PROPERTIES.forEach(filteredConfig::remove);
+
+    // Browsers don't use dead letter policy or redelivery backoff
+    // as they don't acknowledge or redeliver messages
+    return new ConsumerConfiguration(
+        filteredConfig,
+        consumerSchema,
+        null, // deadLetterPolicy = null for browsers
+        null, // negativeAckRedeliveryBackoff = null for browsers
+        null); // ackTimeoutRedeliveryBackoff = null for browsers
+  }
+
   static ConsumerConfiguration buildConsumerConfiguration(
       Map<String, Object> consumerConfigurationM) {
     if (consumerConfigurationM == null || consumerConfigurationM.isEmpty()) {

pulsar-jms/src/main/java/com/datastax/oss/pulsar/jms/PulsarConnectionFactory.java

@@ -1520,6 +1520,14 @@ public List<Reader<?>> createReadersForBrowser(
       PulsarQueue destination, ConsumerConfiguration overrideConsumerConfiguration)
       throws JMSException {
 
+    // Filter consumer-specific properties at the entry point for browser creation.
+    // This ensures all reader creation paths use browser-compatible configuration.
+    // The filtering is done once here rather than in each recursive call for efficiency.
+    ConsumerConfiguration browserConfiguration =
+        overrideConsumerConfiguration != null
+            ? overrideConsumerConfiguration.forBrowser()
+            : null;
+
     if (destination.isRegExp()) {
       try {
         String topicName = getPulsarTopicName(destination);
@@ -1530,7 +1538,7 @@ public List<Reader<?>> createReadersForBrowser(
         for (String sub : topicNames) {
           String queueName = sub + ":" + getQueueSubscriptionName(destination);
           PulsarQueue queue = new PulsarQueue(queueName);
-          res.addAll(createReadersForBrowser(queue, overrideConsumerConfiguration));
+          res.addAll(createReadersForBrowser(queue, browserConfiguration));
         }
         return res;
       } catch (Exception err) {
@@ -1540,7 +1548,7 @@ public List<Reader<?>> createReadersForBrowser(
       List<Reader<?>> res = new ArrayList<>();
       List<PulsarDestination> destinations = destination.getDestinations();
       for (PulsarDestination sub : destinations) {
-        res.addAll(createReadersForBrowser((PulsarQueue) sub, overrideConsumerConfiguration));
+        res.addAll(createReadersForBrowser((PulsarQueue) sub, browserConfiguration));
       }
       return res;
     } else {
@@ -1556,7 +1564,7 @@ public List<Reader<?>> createReadersForBrowser(
               createReaderForBrowserForNonPartitionedTopic(
                   queueSubscriptionName,
                   fullQualifiedTopicName,
-                  overrideConsumerConfiguration,
+                  browserConfiguration,
                   destination);
           readers.add(readerForBrowserForNonPartitionedTopic);
         } else {
@@ -1566,7 +1574,7 @@ public List<Reader<?>> createReadersForBrowser(
                 createReaderForBrowserForNonPartitionedTopic(
                     queueSubscriptionName,
                     partitionName,
-                    overrideConsumerConfiguration,
+                    browserConfiguration,
                     destination);
             readers.add(readerForBrowserForNonPartitionedTopic);
           }
@@ -1604,15 +1612,21 @@ private Reader<?> createReaderForBrowserForNonPartitionedTopic(
       }
       log.info("createBrowser {} at {}", fullQualifiedTopicName, seekMessageId);
 
+      // The overrideConsumerConfiguration passed here is already filtered by
+      // createReadersForBrowser() to remove consumer-specific properties that are
+      // not applicable to Reader API. This ensures compatibility and prevents errors.
       ConsumerConfiguration consumerConfiguration =
           getConsumerConfiguration(overrideConsumerConfiguration, destination);
+
       Schema<?> schema = consumerConfiguration.getConsumerSchema();
       if (schema == null) {
         schema = Schema.BYTES;
       }
+
+      // Use the browser-compatible configuration for the reader
       Map<String, Object> readerConfiguration =
           Utils.deepCopyMap(consumerConfiguration.getConsumerConfiguration());
-      readerConfiguration.remove("batchIndexAckEnabled");
+
       ReaderBuilder<?> builder =
           pulsarClient
               .newReader(schema)