PIP-22: Dead Letter Topic

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursor.java

@@ -23,13 +23,16 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ExecutionException;
+
 import org.apache.bookkeeper.mledger.AsyncCallbacks.ClearBacklogCallback;
 import org.apache.bookkeeper.mledger.AsyncCallbacks.DeleteCallback;
 import org.apache.bookkeeper.mledger.AsyncCallbacks.FindEntryCallback;
 import org.apache.bookkeeper.mledger.AsyncCallbacks.MarkDeleteCallback;
 import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntriesCallback;
 import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntryCallback;
 import org.apache.bookkeeper.mledger.AsyncCallbacks.SkipEntriesCallback;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
 
 /**
  * A ManangedCursor is a persisted cursor inside a ManagedLedger.
@@ -75,6 +78,16 @@ enum IndividualDeletedEntries {
      */
     Map<String, Long> getProperties();
 
+    /**
+     * Return entry at the position.
+     */
+    Entry readEntry(PositionImpl position) throws InterruptedException, ExecutionException;
+
+    /**
+     * Return entry at the position async.
+     */
+    void asyncReadEntry(PositionImpl position, ReadEntryCallback callback, Object ctx);
+
     /**
      * Read entries from the ManagedLedger, up to the specified number. The returned list can be smaller.
      *

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java

@@ -44,7 +44,9 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
@@ -385,6 +387,27 @@ public void operationFailed(ManagedLedgerException exception) {
         });
     }
 
+    @Override
+    public Entry readEntry(PositionImpl position) throws InterruptedException, ExecutionException {
+        final CompletableFuture<Entry> readFuture = new CompletableFuture<>();
+        ledger.asyncReadEntry(position, new ReadEntryCallback() {
+            @Override
+            public void readEntryComplete(Entry entry, Object ctx) {
+                readFuture.complete(entry);
+            }
+            @Override
+            public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+                readFuture.completeExceptionally(exception);
+            }
+        }, null);
+        return readFuture.get();
+    }
+
+    @Override
+    public void asyncReadEntry(PositionImpl position, ReadEntryCallback callback, Object ctx) {
+        ledger.asyncReadEntry(position, callback, ctx);
+    }
+
     @Override
     public List<Entry> readEntries(int numberOfEntriesToRead) throws InterruptedException, ManagedLedgerException {
         checkArgument(numberOfEntriesToRead > 0);

managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorContainerTest.java

@@ -27,6 +27,8 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ExecutionException;
+
 import org.apache.bookkeeper.mledger.AsyncCallbacks;
 import org.apache.bookkeeper.mledger.AsyncCallbacks.ClearBacklogCallback;
 import org.apache.bookkeeper.mledger.AsyncCallbacks.DeleteCallback;
@@ -60,6 +62,17 @@ public Map<String, Long> getProperties() {
             return Collections.emptyMap();
         }
 
+        @Override
+        public Entry readEntry(PositionImpl position) throws InterruptedException, ExecutionException {
+            return null;
+        }
+
+
+        @Override
+        public void asyncReadEntry(PositionImpl position, ReadEntryCallback callback, Object ctx) {
+
+        }
+
         @Override
         public boolean isDurable() {
             return true;

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java

@@ -867,9 +867,8 @@ protected void internalCreateSubscription(String subscriptionName, MessageIdImpl
                 if (topic.getSubscriptions().containsKey(subscriptionName)) {
                     throw new RestException(Status.CONFLICT, "Subscription already exists for topic");
                 }
-
                 PersistentSubscription subscription = (PersistentSubscription) topic
-                        .createSubscription(subscriptionName, InitialPosition.Latest).get();
+                        .createSubscription(subscriptionName, InitialPosition.Latest, 0, null).get();
                 subscription.resetCursor(PositionImpl.get(messageId.getLedgerId(), messageId.getEntryId())).get();
                 log.info("[{}][{}] Successfully created subscription {} at message id {}", clientAppId(), topicName,
                         subscriptionName, messageId);

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

@@ -0,0 +1,37 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service;
+
+import org.apache.bookkeeper.mledger.Position;
+
+import java.util.List;
+
+public interface RedeliveryTracker {
+
+    int incrementAndGetRedeliveryCount(Position position);
+
+    int getRedeliveryCount(Position position);
+
+    void remove(Position position);
+
+    void removeBatch(List<Position> positions);
+
+    void clear();
+
+}

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

@@ -523,6 +523,9 @@ protected void handleSubscribe(final CommandSubscribe subscribe) {
         final boolean readCompacted = subscribe.getReadCompacted();
         final Map<String, String> metadata = CommandUtils.metadataFromCommand(subscribe);
         final InitialPosition initialPosition = subscribe.getInitialPosition();
+        final int maxRedeliveryCount = subscribe.getMaxRedeliveryCount();
+        final String deadLetterTopic = subscribe.getDeadLetterTopic();
+        final int maxUnackedMessagesPerConsumer = subscribe.getMaxUnackedMessagePerConsumer();
         final SchemaData schema = subscribe.hasSchema() ? getSchema(subscribe.getSchema()) : null;
 
         CompletableFuture<Boolean> isProxyAuthorizedFuture;
@@ -591,7 +594,8 @@ protected void handleSubscribe(final CommandSubscribe subscribe) {
                                             if (isCompatible) {
                                                 return topic.subscribe(ServerCnx.this, subscriptionName, consumerId,
                                                     subType, priorityLevel, consumerName, isDurable,
-                                                    startMessageId, metadata, readCompacted, initialPosition);
+                                                    startMessageId, metadata, readCompacted, initialPosition,
+                                                    0, null, maxUnackedMessagesPerConsumer);
                                             } else {
                                                 return FutureUtil.failedFuture(new BrokerServiceException(
                                                     "Trying to subscribe with incompatible schema"
@@ -601,7 +605,8 @@ protected void handleSubscribe(final CommandSubscribe subscribe) {
                                     } else {
                                         return topic.subscribe(ServerCnx.this, subscriptionName, consumerId,
                                             subType, priorityLevel, consumerName, isDurable,
-                                            startMessageId, metadata, readCompacted, initialPosition);
+                                            startMessageId, metadata, readCompacted, initialPosition,
+                                            maxRedeliveryCount, deadLetterTopic, maxUnackedMessagesPerConsumer);
                                     }
                                 })
                                 .thenAccept(consumer -> {

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

@@ -81,9 +81,11 @@ default long getOriginalSequenceId() {
 
     CompletableFuture<Consumer> subscribe(ServerCnx cnx, String subscriptionName, long consumerId, SubType subType,
             int priorityLevel, String consumerName, boolean isDurable, MessageId startMessageId,
-            Map<String, String> metadata, boolean readCompacted, InitialPosition initialPosition);
+            Map<String, String> metadata, boolean readCompacted, InitialPosition initialPosition,
+            int maxRedeliveryCount, String deadLetterTopic, int maxUnackedMessagesPerConsumer);
 
-    CompletableFuture<Subscription> createSubscription(String subscriptionName, InitialPosition initialPosition);
+    CompletableFuture<Subscription> createSubscription(String subscriptionName, InitialPosition initialPosition,
+           int maxRedeliveryCount, String deadLetterTopic);
 
     CompletableFuture<Void> unsubscribe(String subName);
 

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentRedeliveryTracker.java

@@ -0,0 +1,60 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service.nonpersistent;
+
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.pulsar.broker.service.RedeliveryTracker;
+
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class NonPersistentRedeliveryTracker implements RedeliveryTracker {
+
+    private ConcurrentHashMap<Position, AtomicInteger> trackerCache = new ConcurrentHashMap<>(16);
+
+    @Override
+    public int incrementAndGetRedeliveryCount(Position position) {
+        trackerCache.putIfAbsent(position, new AtomicInteger(0));
+        return trackerCache.get(position).incrementAndGet();
+    }
+
+    @Override
+    public int getRedeliveryCount(Position position) {
+        trackerCache.putIfAbsent(position, new AtomicInteger(0));
+        return trackerCache.get(position).get();
+    }
+
+    @Override
+    public void remove(Position position) {
+        trackerCache.remove(position);
+    }
+
+    @Override
+    public void removeBatch(List<Position> positions) {
+        if (positions != null) {
+            positions.forEach(this::remove);
+        }
+    }
+
+    @Override
+    public void clear() {
+        trackerCache.clear();
+    }
+}

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java

@@ -314,7 +314,8 @@ public void removeProducer(Producer producer) {
     @Override
     public CompletableFuture<Consumer> subscribe(final ServerCnx cnx, String subscriptionName, long consumerId,
             SubType subType, int priorityLevel, String consumerName, boolean isDurable, MessageId startMessageId,
-            Map<String, String> metadata, boolean readCompacted, InitialPosition initialPosition) {
+            Map<String, String> metadata, boolean readCompacted, InitialPosition initialPosition, int maxRedeliveryCount,
+            String deadLetterTopic, int maxUnackedMessagesPerConsumer) {
 
         final CompletableFuture<Consumer> future = new CompletableFuture<>();
 
@@ -388,7 +389,8 @@ public CompletableFuture<Consumer> subscribe(final ServerCnx cnx, String subscri
     }
 
     @Override
-    public CompletableFuture<Subscription> createSubscription(String subscriptionName, InitialPosition initialPosition) {
+    public CompletableFuture<Subscription> createSubscription(String subscriptionName, InitialPosition initialPosition,
+              int maxRedeliveryCount, String deadLetterTopic) {
         return CompletableFuture.completedFuture(new NonPersistentSubscription(this, subscriptionName));
     }
 

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/CompactorSubscription.java

@@ -41,7 +41,7 @@ public class CompactorSubscription extends PersistentSubscription {
 
     public CompactorSubscription(PersistentTopic topic, CompactedTopic compactedTopic,
                                  String subscriptionName, ManagedCursor cursor) {
-        super(topic, subscriptionName, cursor);
+        super(topic, subscriptionName, cursor, 0, null);
         checkArgument(subscriptionName.equals(Compactor.COMPACTION_SUBSCRIPTION));
         this.compactedTopic = compactedTopic;