diff --git a/pulsar-jms/src/main/java/com/datastax/oss/pulsar/jms/ConsumerConfiguration.java b/pulsar-jms/src/main/java/com/datastax/oss/pulsar/jms/ConsumerConfiguration.java index baf3be95..f478d6f5 100644 --- a/pulsar-jms/src/main/java/com/datastax/oss/pulsar/jms/ConsumerConfiguration.java +++ b/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. + * + *

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 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 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. + * + *

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. + * + *

This method filters out the following categories of properties: + * + *

+ * + *

Additionally, deadLetterPolicy and redelivery backoff configurations are set to null as + * browsers don't acknowledge or redeliver messages. + * + *

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 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 consumerConfigurationM) { if (consumerConfigurationM == null || consumerConfigurationM.isEmpty()) { diff --git a/pulsar-jms/src/main/java/com/datastax/oss/pulsar/jms/PulsarConnectionFactory.java b/pulsar-jms/src/main/java/com/datastax/oss/pulsar/jms/PulsarConnectionFactory.java index 7d73b2d6..8f08c4dc 100644 --- a/pulsar-jms/src/main/java/com/datastax/oss/pulsar/jms/PulsarConnectionFactory.java +++ b/pulsar-jms/src/main/java/com/datastax/oss/pulsar/jms/PulsarConnectionFactory.java @@ -1520,6 +1520,14 @@ public List> 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> 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> createReadersForBrowser( List> res = new ArrayList<>(); List 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> createReadersForBrowser( createReaderForBrowserForNonPartitionedTopic( queueSubscriptionName, fullQualifiedTopicName, - overrideConsumerConfiguration, + browserConfiguration, destination); readers.add(readerForBrowserForNonPartitionedTopic); } else { @@ -1566,7 +1574,7 @@ public List> 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 readerConfiguration = Utils.deepCopyMap(consumerConfiguration.getConsumerConfiguration()); - readerConfiguration.remove("batchIndexAckEnabled"); + ReaderBuilder builder = pulsarClient .newReader(schema)