Nack support in WS

pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTest.java

@@ -80,6 +80,8 @@
 import org.testng.annotations.AfterMethod;
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonPrimitive;
 
 @Test(groups = "websocket")
 public class ProxyPublishConsumeTest extends ProducerConsumerBase {
@@ -761,6 +763,76 @@ public void socketPullModeTest() throws Exception {
         }
     }
 
+    @Test(timeOut = 10000)
+    public void nackMessageTest() throws Exception {
+        final String subscription = "my-sub";
+        final String dlqTopic = "my-property/my-ns/my-topic10";
+        final String consumerTopic = "my-property/my-ns/my-topic9";
+
+        final String dlqUri = "ws://localhost:" + proxyServer.getListenPortHTTP().get() +
+          "/ws/v2/consumer/persistent/" +
+          dlqTopic + "/" + subscription +
+          "?subscriptionType=Shared";
+
+        final String consumerUri = "ws://localhost:" + proxyServer.getListenPortHTTP().get() +
+          "/ws/v2/consumer/persistent/" +
+          consumerTopic + "/" + subscription +
+          "?deadLetterTopic=" + dlqTopic +
+          "&maxRedeliverCount=0&subscriptionType=Shared&ackTimeoutMillis=1000&negativeAckRedeliveryDelay=1000";
+
+        final String producerUri = "ws://localhost:" + proxyServer.getListenPortHTTP().get() +
+          "/ws/v2/producer/persistent/" + consumerTopic;
+
+        WebSocketClient consumeClient1 = new WebSocketClient();
+        SimpleConsumerSocket consumeSocket1 = new SimpleConsumerSocket();
+        WebSocketClient consumeClient2 = new WebSocketClient();
+        SimpleConsumerSocket consumeSocket2 = new SimpleConsumerSocket();
+        WebSocketClient produceClient = new WebSocketClient();
+        SimpleProducerSocket produceSocket = new SimpleProducerSocket();
+
+        consumeSocket1.setMessageHandler((id, data) -> {
+            JsonObject nack = new JsonObject();
+            nack.add("messageId", new JsonPrimitive(id));
+            nack.add("type", new JsonPrimitive("negativeAcknowledge"));
+            return nack.toString();
+        });
+
+        try {
+            consumeClient1.start();
+            consumeClient2.start();
+            ClientUpgradeRequest consumeRequest1 = new ClientUpgradeRequest();
+            ClientUpgradeRequest consumeRequest2 = new ClientUpgradeRequest();
+            Future<Session> consumerFuture1 = consumeClient1.connect(consumeSocket1, URI.create(consumerUri), consumeRequest1);
+            Future<Session> consumerFuture2 = consumeClient2.connect(consumeSocket2, URI.create(dlqUri), consumeRequest2);
+
+            assertTrue(consumerFuture1.get().isOpen());
+            assertTrue(consumerFuture2.get().isOpen());
+
+            ClientUpgradeRequest produceRequest = new ClientUpgradeRequest();
+            produceClient.start();
+            Future<Session> producerFuture = produceClient.connect(produceSocket, URI.create(producerUri), produceRequest);
+            assertTrue(producerFuture.get().isOpen());
+
+            assertEquals(consumeSocket1.getReceivedMessagesCount(), 0);
+            assertEquals(consumeSocket2.getReceivedMessagesCount(), 0);
+
+            produceSocket.sendMessage(1);
+
+            Thread.sleep(500);
+
+            //assertEquals(consumeSocket1.getReceivedMessagesCount(), 1);
+            assertTrue(consumeSocket1.getReceivedMessagesCount() > 0);
+
+            Thread.sleep(500);
+
+            //assertEquals(consumeSocket2.getReceivedMessagesCount(), 1);
+            assertTrue(consumeSocket1.getReceivedMessagesCount() > 0);
+
+        } finally {
+            stopWebSocketClient(consumeClient1, consumeClient2, produceClient);
+        }
+    }
+
     private void verifyTopicStat(Client client, String baseUrl, String topic) {
         String statUrl = baseUrl + topic + "/stats";
         WebTarget webTarget = client.target(statUrl);

pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/SimpleConsumerMessageHandler.java

@@ -0,0 +1,25 @@
+/**
+ * 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.websocket.proxy;
+
+import com.google.gson.JsonObject;
+
+public interface SimpleConsumerMessageHandler {
+  String handle(String id, JsonObject message);
+}

pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/SimpleConsumerSocket.java

@@ -45,6 +45,8 @@ public class SimpleConsumerSocket {
     private Session session;
     private final ArrayList<String> consumerBuffer;
     private final AtomicInteger receivedMessages = new AtomicInteger();
+    // Custom message handler to override standard message processing, if it's needed
+    private SimpleConsumerMessageHandler customMessageHandler;
 
     public SimpleConsumerSocket() {
         this.closeLatch = new CountDownLatch(1);
@@ -55,6 +57,10 @@ public boolean awaitClose(int duration, TimeUnit unit) throws InterruptedExcepti
         return this.closeLatch.await(duration, unit);
     }
 
+    public void setMessageHandler(SimpleConsumerMessageHandler handler) {
+        customMessageHandler = handler;
+    }
+
     @OnWebSocketClose
     public void onClose(int statusCode, String reason) {
         log.info("Connection closed: {} - {}", statusCode, reason);
@@ -73,15 +79,15 @@ public void onConnect(Session session) throws InterruptedException {
     public synchronized void onMessage(String msg) throws JsonParseException, IOException {
         receivedMessages.incrementAndGet();
         JsonObject message = new Gson().fromJson(msg, JsonObject.class);
-        if (message.get(X_PULSAR_MESSAGE_ID) != null) {
+        String messageId = message.get(X_PULSAR_MESSAGE_ID).getAsString();
+        consumerBuffer.add(messageId);
+        if (customMessageHandler != null) {
+            this.getRemote().sendString(customMessageHandler.handle(messageId, message));
+        } else {
             JsonObject ack = new JsonObject();
-            String messageId = message.get(X_PULSAR_MESSAGE_ID).getAsString();
-            consumerBuffer.add(messageId);
             ack.add("messageId", new JsonPrimitive(messageId));
             // Acking the proxy
             this.getRemote().sendString(ack.toString());
-        } else {
-            consumerBuffer.add(message.toString());
         }
     }
 

pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ConsumerHandler.java

@@ -217,6 +217,8 @@ public void onWebSocketText(String message) {
                 handlePermit(command);
             } else if ("unsubscribe".equals(command.type)) {
                 handleUnsubscribe(command);
+            } else if ("negativeAcknowledge".equals(command.type)) {
+                handleNack(command);
             } else if ("isEndOfTopic".equals(command.type)) {
                 handleEndOfTopic();
             } else {
@@ -260,11 +262,7 @@ private void handleUnsubscribe(ConsumerCommand command) throws PulsarClientExcep
         consumer.unsubscribe();
     }
 
-    private void handleAck(ConsumerCommand command) throws IOException {
-        // We should have received an ack
-        MessageId msgId = MessageId.fromByteArrayWithTopic(Base64.getDecoder().decode(command.messageId),
-                topic.toString());
-        consumer.acknowledgeAsync(msgId).thenAccept(consumer -> numMsgsAcked.increment());
+    private void checkResumeReceive() {
         if (!this.pullMode) {
             int pending = pendingMessages.getAndDecrement();
             if (pending >= maxPendingMessages) {
@@ -274,6 +272,22 @@ private void handleAck(ConsumerCommand command) throws IOException {
         }
     }
 
+    private void handleAck(ConsumerCommand command) throws IOException {
+        // We should have received an ack
+        MessageId msgId = MessageId.fromByteArrayWithTopic(Base64.getDecoder().decode(command.messageId),
+                topic.toString());
+        consumer.acknowledgeAsync(msgId).thenAccept(consumer -> numMsgsAcked.increment());
+        checkResumeReceive();
+    }
+
+    private void handleNack(ConsumerCommand command) throws IOException {
+        MessageId msgId = MessageId.fromByteArrayWithTopic(Base64.getDecoder().decode(command.messageId),
+            topic.toString());
+        System.out.println(msgId);
+        consumer.negativeAcknowledge(msgId);
+        checkResumeReceive();
+    }
+
     private void handlePermit(ConsumerCommand command) throws IOException {
         if (command.permitMessages == null) {
             throw new IOException("Missing required permitMessages field for 'permit' command");
@@ -383,6 +397,9 @@ protected ConsumerBuilder<byte[]> getConsumerConfiguration(PulsarClient client)
             if (queryParams.containsKey("deadLetterTopic")) {
                 dlpBuilder.deadLetterTopic(queryParams.get("deadLetterTopic"));
             }
+            if (queryParams.containsKey("negativeAckRedeliveryDelay")) {
+                builder.negativeAckRedeliveryDelay(Integer.parseInt(queryParams.get("negativeAckRedeliveryDelay")), TimeUnit.MILLISECONDS);
+            }
             builder.deadLetterPolicy(dlpBuilder.build());
         }
 

site2/docs/client-libraries-websocket.md

@@ -167,6 +167,7 @@ Key | Type | Required? | Explanation
 `maxRedeliverCount` | int | no | Define a [maxRedeliverCount](http://pulsar.apache.org/api/client/org/apache/pulsar/client/api/ConsumerBuilder.html#deadLetterPolicy-org.apache.pulsar.client.api.DeadLetterPolicy-) for the consumer (default: 0). Activates [Dead Letter Topic](https://github.com/apache/pulsar/wiki/PIP-22%3A-Pulsar-Dead-Letter-Topic) feature.
 `deadLetterTopic` | string | no | Define a [deadLetterTopic](http://pulsar.apache.org/api/client/org/apache/pulsar/client/api/ConsumerBuilder.html#deadLetterPolicy-org.apache.pulsar.client.api.DeadLetterPolicy-) for the consumer (default: {topic}-{subscription}-DLQ). Activates [Dead Letter Topic](https://github.com/apache/pulsar/wiki/PIP-22%3A-Pulsar-Dead-Letter-Topic) feature.
 `pullMode` | boolean | no | Enable pull mode (default: false). See "Flow Control" below.
+`negativeAckRedeliveryDelay` | int | no | When a message is negatively acknowledged, it will be redelivered to the DLQ.
 `token` | string | no | Authentication token, this is used for the browser javascript client
 
 NB: these parameter (except `pullMode`) apply to the internal consumer of the WebSocket service.
@@ -211,6 +212,18 @@ Key | Type | Required? | Explanation
 :---|:-----|:----------|:-----------
 `messageId`| string | yes | Message ID of the processed message
 
+#### Negatively acknowledging messages
+```json
+{
+  "type": "negativeAcknowledge",
+  "messageId": "CAAQAw=="
+}
+```
+
+Key | Type | Required? | Explanation
+:---|:-----|:----------|:-----------
+`messageId`| string | yes | Message ID of the processed message
+
 #### Flow control
 
 ##### Push Mode