Consumer filtering

README_CONSUMER_FILTERS

@@ -0,0 +1,81 @@
+Tag based Consumer Filtering
+----------------------------
+
+Enables broker side "tag" based filtering of messages on selected message properties by allowing consumers to 
+specify a subset of message properties as being “tags” to filter on, using existing consumer (meta data) 
+property API (and underlying protocol). Message properties which have a key beginning with the string "tag" 
+are matched by value against the consumer subscription properties that are specified on subscription with 
+meta data string keys starting with either "anytag" or "alltag". What follows “tag”, “anytag”, “alltag” 
+textually is immaterial except to allow multiple key /  values to be enumerated (e.g. “tag1”, “tag2” etc).
+
+An "anytag" matches successfully if any of the messages "tag" properties has the same property value as the 
+consumer property value. This allows "or" matching.
+e.g. Say a consumer subscribes with an "anytag" property with "anytag1" of "urgent" and another with key 
+"anytag2" and value "compressed". Then a message with property "tag0" as key and "urgent" as value will match 
+as would a message with "tagA" / "compressed".
+
+To subscribe the consumer:
+
+Consumer<String> consumer = client.newConsumer(Schema.STRING)
+                     .topic("sometopic")
+                    .property("anytag0", "urgent") // A filtering meta data tag.
+                    .property("anytag1", "compressed") // Another filtering meta data tag.
+                    .subscribe();
+
+To send a message with a tag that matches:
+
+Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+                    .topic(topic)
+                    .create();
+producer.newMessage()
+                   .property("tag0", "urgent") // tag with a matching property value.
+                   .value("my-matching-message")
+                   .sendAsync();
+
+An "alltag" will only match if all other alltags in the consumer subscription meta data properties also 
+can be matched. This allows "and" matching of multiple values.
+e.g. Say a consumer subscribes with an "alltag" property with "alltag1" of "urgent" and another with key 
+"alltag2" and value "compressed". Then a message with property "tag0" as key and "urgent" and property 
+"tag1" of "compressed" as value will match while a message with "tag0" of "urgent" and "tag1" of 
+"uncompressed" property values will not.
+
+Consumer<String> consumer = client.newConsumer(Schema.STRING)
+                    .topic("sometopic")
+                    .property("alltag0", "urgent") // A filtering meta tag.
+                    .property("alltag1", "compressed") // Additional filtering meta tag.
+                    .subscribe();
+
+will match on the following message:
+
+producer.newMessage()
+                   .property("tag0", "urgent") // A tag property value.
+                   .property("tag1", "compressed") // A second tag property value.
+                   .value("my-matching-message")
+                   .sendAsync();
+
+But the following will not match:
+
+producer.newMessage()
+                   .property("tag0", "urgent") // A tag property value.
+                   .property("tag1", "uncompressed") // A second tag property value.
+                   .value("my-not-matching-message")
+                   .sendAsync();
+
+producer.newMessage()
+                   .property("tag0", "urgent") // A tag property value.
+                   .value("my-other-not-matching-message")
+                   .sendAsync();   
+
+If the consumer subscribes with both anytags and alltags then at least one anytag and all of the alltags 
+must match tags in the message (with overlap allowed) for the message to be passed by the filter.
+The default implementation matches only on properties that have keys beginning with "tag" to help avoid 
+unintended matching. A new Java ConsumerFilter interface and factory allows other property-based filtering 
+schemes to be implemented and deployed.
+Apart from message tag properties starting with (lowercase) "tag" and consumer tag subscription meta data 
+properties starting with "anytag" and "alltag" there are no other requirements or conventions on tag naming 
+and usage. For example, tags could use more meaningful keys such as "tag_priority" and namespace tag values 
+such as values "prority_urgent" and "priority_low". 
+Another possibility is to use values such as "priority=urgent" and match by text on such categories or classes 
+(with keys just enumerating "tag0","tag1" etc as in examples above). 
+These suggestions are a convention only and not checked or enforced in any way. In the examples above we have 
+used "0" and "1" but other unique strings (such as "A","B" etc) could have been used.

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java

@@ -674,6 +674,18 @@ public class ServiceConfiguration implements PulsarConfiguration {
         doc = "Number of worker threads to serve non-persistent topic")
     private int numWorkerThreadsForNonPersistentTopic = Runtime.getRuntime().availableProcessors();
 
+    @FieldContext(
+            category = CATEGORY_SERVER,
+            doc = "Enable broker to filter messages for consumer subscriptions"
+    )
+    private boolean enableConsumerFilters = true;
+
+    @FieldContext(
+            category = CATEGORY_SERVER,
+            doc = "Consumer filter implementation"
+    )
+    private String consumerFilterClass = "";
+
     @FieldContext(
         category = CATEGORY_SERVER,
         doc = "Enable broker to load persistent topics"

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java

@@ -25,6 +25,7 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.Optional;
 
 import lombok.extern.slf4j.Slf4j;
 import org.apache.bookkeeper.mledger.Entry;
@@ -33,6 +34,8 @@
 import org.apache.bookkeeper.mledger.impl.PositionImpl;
 import org.apache.commons.lang3.tuple.Pair;
 import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.bookkeeper.mledger.impl.EntryImpl;
+import org.apache.pulsar.client.impl.RawBatchConverter;
 import org.apache.pulsar.common.api.proto.PulsarApi;
 import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
 import org.apache.pulsar.common.api.proto.PulsarApi.MessageMetadata;
@@ -61,6 +64,9 @@ protected AbstractBaseDispatcher(Subscription subscription) {
      * <li>Message is not meant to be delivered immediately
      * </ul>
      *
+     * @param consumer
+     *            the consumer we are filtering for.
+     *
      * @param entries
      *            a list of entries as read from storage
      *
@@ -71,24 +77,75 @@ protected AbstractBaseDispatcher(Subscription subscription) {
      * @param sendMessageInfo
      *            an object where the total size in messages and bytes will be returned back to the caller
      */
-    public void filterEntriesForConsumer(List<Entry> entries, EntryBatchSizes batchSizes,
+    public void filterEntriesForConsumer(Consumer consumer, List<Entry> entries, EntryBatchSizes batchSizes,
                                          SendMessageInfo sendMessageInfo, EntryBatchIndexesAcks indexesAcks,
                                          ManagedCursor cursor, boolean isReplayRead) {
         int totalMessages = 0;
         long totalBytes = 0;
         int totalChunkedMessages = 0;
 
         boolean isAfterTxnCommitMarker = false;
+        consumer.initFiltering(); // Ensure consumer filtering is initialized on first use.
 
         for (int i = 0, entriesSize = entries.size(); i < entriesSize; i++) {
+
+            MessageMetadata msgMetadata = null;
             Entry entry = entries.get(i);
             if (entry == null) {
                 continue;
             }
 
             ByteBuf metadataAndPayload = entry.getDataBuffer();
 
-            MessageMetadata msgMetadata = Commands.peekMessageMetadata(metadataAndPayload, subscription.toString(), -1);
+            msgMetadata = Commands.peekMessageMetadata(metadataAndPayload, subscription.toString(), consumer.consumerId());
+            boolean isFiltering = consumer.getConsumerFilter().isFiltering();
+            if (isFiltering) {
+                try {
+                    if (msgMetadata.hasNumMessagesInBatch() && msgMetadata.getEncryptionKeysCount() == 0) {
+                        msgMetadata = null;
+                        int readerIdx = metadataAndPayload.readerIndex();
+                        Optional<ByteBuf> filteredOrEmpty;
+                        try {
+                            filteredOrEmpty = RawBatchConverter.filter(metadataAndPayload, consumer.getConsumerFilter());
+                        } finally {
+                            metadataAndPayload.readerIndex(readerIdx);
+                        }
+                        if (filteredOrEmpty.isPresent()) {
+                            ByteBuf filtered = filteredOrEmpty.get();
+                            Entry filteredEntry = EntryImpl.create(entry.getLedgerId(), entry.getEntryId(), filtered);
+
+                            entries.set(i, filteredEntry);
+                            entry.release();
+                            metadataAndPayload.release();
+                            metadataAndPayload = filtered;
+                            msgMetadata = Commands.peekMessageMetadata(metadataAndPayload, subscription.toString(), consumer.consumerId());
+                        } // else nothing to forwards and msgMetadata remains null which is handled below.
+                    } else {
+                        if (msgMetadata.getEncryptionKeysCount() == 0) {
+                            // Filter single message.
+                            List<PulsarApi.KeyValue> properties = msgMetadata.getPropertiesList();
+                            // Position reader to message.
+                            int readerIdx = metadataAndPayload.readerIndex();
+                            try {
+                                Commands.skipChecksumIfPresent(metadataAndPayload);
+                                int metadataSize = (int) metadataAndPayload.readUnsignedInt();
+                                metadataAndPayload.readerIndex(metadataAndPayload.readerIndex() + metadataSize);
+
+                                if (!consumer.getConsumerFilter().filter(properties, metadataAndPayload)) {
+                                    msgMetadata.recycle();
+                                    msgMetadata = null;
+                                }
+                            } finally {
+                                metadataAndPayload.readerIndex(readerIdx);
+                            }
+                        }
+                    }
+                } catch (IOException ioe) {
+                    log.error("Error in filtering for consumer - dropping messages.", ioe);
+                    entries.set(i, null);
+                    entry.release();
+                }
+            }
 
             try {
                 if (!isReplayRead && msgMetadata != null
@@ -106,7 +163,7 @@ public void filterEntriesForConsumer(List<Entry> entries, EntryBatchSizes batchS
                     continue;
                 } else if (msgMetadata == null || Markers.isServerOnlyMarker(msgMetadata)) {
                     PositionImpl pos = (PositionImpl) entry.getPosition();
-                    // Message metadata was corrupted or the messages was a server-only marker
+                    // Message metadata was filtered out or corrupted or the messages was a server-only marker
 
                     if (Markers.isReplicatedSubscriptionSnapshotMarker(msgMetadata)) {
                         processReplicatedSubscriptionSnapshot(pos, metadataAndPayload);
@@ -135,6 +192,7 @@ && trackDelayedDelivery(entry.getLedgerId(), entry.getEntryId(), msgMetadata)) {
                 totalBytes += metadataAndPayload.readableBytes();
                 totalChunkedMessages += msgMetadata.hasChunkId() ? 1: 0;
                 batchSizes.setBatchSize(i, batchSize);
+
                 if (indexesAcks != null && cursor != null) {
                     long[] ackSet = cursor.getDeletedBatchIndexesAsLongArray(PositionImpl.get(entry.getLedgerId(), entry.getEntryId()));
                     if (ackSet != null) {
@@ -143,8 +201,14 @@ && trackDelayedDelivery(entry.getLedgerId(), entry.getEntryId(), msgMetadata)) {
                         indexesAcks.setIndexesAcks(i,null);
                     }
                 }
+
             } finally {
-                msgMetadata.recycle();
+                if (msgMetadata != null) {
+                    msgMetadata.recycle();
+                }
+                if (isFiltering) {
+                    metadataAndPayload.release();
+                }
             }
         }
 

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java

@@ -92,6 +92,8 @@ public class Consumer {
     private long lastAckedTimestamp;
     private Rate chuckedMessageRate;
 
+    private final ConsumerFilter consumerFilter;
+
     // Represents how many messages we can safely send to the consumer without
     // overflowing its receiving queue. The consumer will use Flow commands to
     // increase its availability
@@ -179,6 +181,8 @@ public Consumer(Subscription subscription, SubType subType, String topicName, lo
             // We don't need to keep track of pending acks if the subscription is not shared
             this.pendingAcks = null;
         }
+
+        consumerFilter = ConsumerFilterFactory.createConsumerFilter(this.cnx.getBrokerService().getPulsar().getConfiguration());
     }
 
     public SubType subType() {
@@ -201,11 +205,11 @@ void notifyActiveConsumerChange(Consumer activeConsumer) {
 
         if (log.isDebugEnabled()) {
             log.debug("notify consumer {} - that [{}] for subscription {} has new active consumer : {}",
-                consumerId, topicName, subscription.getName(), activeConsumer);
+                    consumerId, topicName, subscription.getName(), activeConsumer);
         }
         cnx.ctx().writeAndFlush(
-            Commands.newActiveConsumerChange(consumerId, this == activeConsumer),
-            cnx.ctx().voidPromise());
+                Commands.newActiveConsumerChange(consumerId, this == activeConsumer),
+                cnx.ctx().voidPromise());
     }
 
     public boolean readCompacted() {
@@ -218,10 +222,10 @@ public boolean readCompacted() {
      *
      * @return a SendMessageInfo object that contains the detail of what was sent to consumer
      */
-
     public ChannelPromise sendMessages(final List<Entry> entries, EntryBatchSizes batchSizes, EntryBatchIndexesAcks batchIndexesAcks,
                int totalMessages, long totalBytes, long totalChunkedMessages, RedeliveryTracker redeliveryTracker) {
         this.lastConsumedTimestamp = System.currentTimeMillis();
+
         final ChannelHandlerContext ctx = cnx.ctx();
         final ChannelPromise writePromise = ctx.newPromise();
 
@@ -248,6 +252,7 @@ public ChannelPromise sendMessages(final List<Entry> entries, EntryBatchSizes ba
                     int batchSize = batchSizes.getBatchSize(i);
                     pendingAcks.put(entry.getLedgerId(), entry.getEntryId(), batchSize, 0);
                 }
+
             }
         }
 
@@ -287,10 +292,10 @@ public ChannelPromise sendMessages(final List<Entry> entries, EntryBatchSizes ba
 
                 MessageIdData.Builder messageIdBuilder = MessageIdData.newBuilder();
                 MessageIdData messageId = messageIdBuilder
-                    .setLedgerId(entry.getLedgerId())
-                    .setEntryId(entry.getEntryId())
-                    .setPartition(partitionIdx)
-                    .build();
+                        .setLedgerId(entry.getLedgerId())
+                        .setEntryId(entry.getEntryId())
+                        .setPartition(partitionIdx)
+                        .build();
 
                 ByteBuf metadataAndPayload = entry.getDataBuffer();
                 // increment ref-count of data and release at the end of process: so, we can get chance to call entry.release
@@ -390,6 +395,7 @@ void doUnsubscribe(final long requestId) {
     CompletableFuture<Void> messageAcked(CommandAck ack) {
         this.lastAckedTimestamp = System.currentTimeMillis();
         Map<String,Long> properties = Collections.emptyMap();
+
         if (ack.getPropertiesCount() > 0) {
             properties = ack.getPropertiesList().stream()
                 .collect(Collectors.toMap(PulsarApi.KeyLongValue::getKey,
@@ -489,8 +495,7 @@ void flowPermits(int additionalNumberOfMessages) {
      * Triggers dispatcher to dispatch {@code blockedPermits} number of messages and adds same number of permits to
      * {@code messagePermits} as it maintains count of actual dispatched message-permits.
      *
-     * @param consumer:
-     *            Consumer whose blockedPermits needs to be dispatched
+     * @param consumer: Consumer whose blockedPermits needs to be dispatched
      */
     void flowConsumerBlockedPermits(Consumer consumer) {
         int additionalNumberOfPermits = PERMITS_RECEIVED_WHILE_CONSUMER_BLOCKED_UPDATER.getAndSet(consumer, 0);
@@ -608,10 +613,9 @@ public int hashCode() {
     /**
      * first try to remove ack-position from the current_consumer's pendingAcks.
      * if ack-message doesn't present into current_consumer's pendingAcks
-     *  a. try to remove from other connected subscribed consumers (It happens when client
+     * a. try to remove from other connected subscribed consumers (It happens when client
      * tries to acknowledge message through different consumer under the same subscription)
      *
-     *
      * @param position
      */
     private void removePendingAcks(PositionImpl position) {
@@ -725,6 +729,18 @@ public Subscription getSubscription() {
         return subscription;
     }
 
+    public Map<String, String> getMetadata() {
+        return metadata;
+    }
+
+    public void initFiltering() {
+        consumerFilter.initFiltering(this.getMetadata());
+    }
+
+    public ConsumerFilter getConsumerFilter() {
+        return consumerFilter;
+    }
+
     private int addAndGetUnAckedMsgs(Consumer consumer, int ackedMessages) {
         subscription.addUnAckedMessages(ackedMessages);
         return UNACKED_MESSAGES_UPDATER.addAndGet(consumer, ackedMessages);