From 909fac3ca6e056c4d60f209e58e33dcc2db03b67 Mon Sep 17 00:00:00 2001 From: yrsh Date: Tue, 13 Oct 2020 20:23:36 +0300 Subject: [PATCH 1/7] Negative acknowledging for WS API --- .../pulsar/websocket/ConsumerHandler.java | 27 +++++++++++++++---- .../client-libraries-websocket.md | 12 +++++++++ 2 files changed, 34 insertions(+), 5 deletions(-) diff --git a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ConsumerHandler.java b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ConsumerHandler.java index 35b2b0c108dc0..9edc8f0071d78 100644 --- a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ConsumerHandler.java +++ b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ConsumerHandler.java @@ -234,6 +234,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 { handleAck(command); } @@ -247,11 +249,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) { @@ -261,6 +259,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"); @@ -370,6 +384,9 @@ private ConsumerBuilder 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()); } diff --git a/site2/website/versioned_docs/version-2.2.0/client-libraries-websocket.md b/site2/website/versioned_docs/version-2.2.0/client-libraries-websocket.md index 6c252c72df9e6..391d12489a744 100644 --- a/site2/website/versioned_docs/version-2.2.0/client-libraries-websocket.md +++ b/site2/website/versioned_docs/version-2.2.0/client-libraries-websocket.md @@ -143,6 +143,7 @@ Key | Type | Required? | Explanation `receiverQueueSize` | int | no | Size of the consumer receive queue (default: 1000) `consumerName` | string | no | Consumer name `priorityLevel` | int | no | Define a [priority](http://pulsar.apache.org/api/client/org/apache/pulsar/client/api/ConsumerConfiguration.html#setPriorityLevel-int-) for the consumer +`negativeAckRedeliveryDelay` | int | no | When a negative acked message will be redelivered to DLQ. ##### Receiving messages @@ -180,6 +181,17 @@ Key | Type | Required? | Explanation :---|:-----|:----------|:----------- `messageId`| string | yes | Message ID of the processed message +#### Negative acknowledging the message +```json +{ + "type": "negativeAcknowledge", + "messageId": "CAAQAw==" +} +``` + +Key | Type | Required? | Explanation +:---|:-----|:----------|:----------- +`messageId`| string | yes | Message ID of the processed message ### Reader endpoint From f02c48e3b019ddde8da7216c8f7a998df790be14 Mon Sep 17 00:00:00 2001 From: Anton K Date: Sat, 17 Oct 2020 10:26:24 +0300 Subject: [PATCH 2/7] Update site2/website/versioned_docs/version-2.2.0/client-libraries-websocket.md Co-authored-by: HuanliMeng <48120384+Huanli-Meng@users.noreply.github.com> --- .../versioned_docs/version-2.2.0/client-libraries-websocket.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/site2/website/versioned_docs/version-2.2.0/client-libraries-websocket.md b/site2/website/versioned_docs/version-2.2.0/client-libraries-websocket.md index 391d12489a744..739f6091364cf 100644 --- a/site2/website/versioned_docs/version-2.2.0/client-libraries-websocket.md +++ b/site2/website/versioned_docs/version-2.2.0/client-libraries-websocket.md @@ -181,7 +181,7 @@ Key | Type | Required? | Explanation :---|:-----|:----------|:----------- `messageId`| string | yes | Message ID of the processed message -#### Negative acknowledging the message +#### Negatively acknowledging messages ```json { "type": "negativeAcknowledge", From be74ad6f5c10098ccebb67fcc92d68947eb0d378 Mon Sep 17 00:00:00 2001 From: yrsh Date: Sun, 18 Oct 2020 11:26:30 +0300 Subject: [PATCH 3/7] WS docs update (only latest) --- site2/docs/client-libraries-websocket.md | 14 ++++++++++++++ .../version-2.2.0/client-libraries-websocket.md | 12 ------------ 2 files changed, 14 insertions(+), 12 deletions(-) diff --git a/site2/docs/client-libraries-websocket.md b/site2/docs/client-libraries-websocket.md index 39ddf0ce802ef..2619df6a6db9c 100644 --- a/site2/docs/client-libraries-websocket.md +++ b/site2/docs/client-libraries-websocket.md @@ -145,6 +145,8 @@ 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 negative acked message will be redelivered to DLQ. + NB: these parameter (except `pullMode`) apply to the internal consumer of the WebSocket service. So messages will be subject to the redelivery settings as soon as the get into the receive queue, @@ -186,6 +188,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 diff --git a/site2/website/versioned_docs/version-2.2.0/client-libraries-websocket.md b/site2/website/versioned_docs/version-2.2.0/client-libraries-websocket.md index 739f6091364cf..6c252c72df9e6 100644 --- a/site2/website/versioned_docs/version-2.2.0/client-libraries-websocket.md +++ b/site2/website/versioned_docs/version-2.2.0/client-libraries-websocket.md @@ -143,7 +143,6 @@ Key | Type | Required? | Explanation `receiverQueueSize` | int | no | Size of the consumer receive queue (default: 1000) `consumerName` | string | no | Consumer name `priorityLevel` | int | no | Define a [priority](http://pulsar.apache.org/api/client/org/apache/pulsar/client/api/ConsumerConfiguration.html#setPriorityLevel-int-) for the consumer -`negativeAckRedeliveryDelay` | int | no | When a negative acked message will be redelivered to DLQ. ##### Receiving messages @@ -181,17 +180,6 @@ 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 ### Reader endpoint From d204bd484e0a2f560c26bf696a158a398b02db4d Mon Sep 17 00:00:00 2001 From: Anton K Date: Sun, 18 Oct 2020 18:12:47 +0300 Subject: [PATCH 4/7] Update client-libraries-websocket.md docs update --- site2/docs/client-libraries-websocket.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/site2/docs/client-libraries-websocket.md b/site2/docs/client-libraries-websocket.md index 2619df6a6db9c..61b41724b1678 100644 --- a/site2/docs/client-libraries-websocket.md +++ b/site2/docs/client-libraries-websocket.md @@ -145,7 +145,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 negative acked message will be redelivered to DLQ. +`negativeAckRedeliveryDelay` | int | no | When a message is negatively acknowledged, it will be redelivered to the DLQ. NB: these parameter (except `pullMode`) apply to the internal consumer of the WebSocket service. From 2dd2d260473ff6a4efca7b5dd90e92e89c5e9b7a Mon Sep 17 00:00:00 2001 From: yrsh Date: Tue, 10 Nov 2020 14:20:39 +0300 Subject: [PATCH 5/7] functional test for ws nack added --- .../proxy/ProxyPublishConsumeTest.java | 71 +++++++++++++++++++ .../proxy/SimpleConsumerMessageHandler.java | 7 ++ .../websocket/proxy/SimpleConsumerSocket.java | 18 +++-- 3 files changed, 92 insertions(+), 4 deletions(-) create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/SimpleConsumerMessageHandler.java diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTest.java index 2518333c349a0..b0a20d341eba5 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTest.java @@ -70,6 +70,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; public class ProxyPublishConsumeTest extends ProducerConsumerBase { protected String methodName; @@ -564,6 +566,75 @@ 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) -> { + log.info(data.toString()); + 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 consumerFuture1 = consumeClient1.connect(consumeSocket1, URI.create(consumerUri), consumeRequest1); + Future consumerFuture2 = consumeClient2.connect(consumeSocket2, URI.create(dlqUri), consumeRequest2); + + assertTrue(consumerFuture1.get().isOpen()); + assertTrue(consumerFuture2.get().isOpen()); + + ClientUpgradeRequest produceRequest = new ClientUpgradeRequest(); + produceClient.start(); + Future producerFuture = produceClient.connect(produceSocket, URI.create(producerUri), produceRequest); + assertTrue(producerFuture.get().isOpen()); + + log.info("SEND"); + 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); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/SimpleConsumerMessageHandler.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/SimpleConsumerMessageHandler.java new file mode 100644 index 0000000000000..45380c475d146 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/SimpleConsumerMessageHandler.java @@ -0,0 +1,7 @@ +package org.apache.pulsar.websocket.proxy; + +import com.google.gson.JsonObject; + +public interface SimpleConsumerMessageHandler { + String handle(String id, JsonObject message); +} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/SimpleConsumerSocket.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/SimpleConsumerSocket.java index 897c72df8a2b1..bd8e7a9351e4f 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/SimpleConsumerSocket.java +++ b/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 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,12 +79,16 @@ 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); - 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()); + if (customMessageHandler != null) { + this.getRemote().sendString(customMessageHandler.handle(messageId, message)); + } else { + JsonObject ack = new JsonObject(); + ack.add("messageId", new JsonPrimitive(messageId)); + // Acking the proxy + this.getRemote().sendString(ack.toString()); + } } public void sendPermits(int nbPermits) throws IOException { From 241629b865c2d66d9f2d34cfdc6ffac2ea0a3da4 Mon Sep 17 00:00:00 2001 From: yrsh Date: Tue, 10 Nov 2020 14:40:45 +0300 Subject: [PATCH 6/7] functional test for ws nack added (additional assert) --- .../pulsar/websocket/proxy/ProxyPublishConsumeTest.java | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTest.java index b0a20d341eba5..18d65b5cb01ec 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTest.java @@ -594,7 +594,6 @@ public void nackMessageTest() throws Exception { SimpleProducerSocket produceSocket = new SimpleProducerSocket(); consumeSocket1.setMessageHandler((id, data) -> { - log.info(data.toString()); JsonObject nack = new JsonObject(); nack.add("messageId", new JsonPrimitive(id)); nack.add("type", new JsonPrimitive("negativeAcknowledge")); @@ -617,7 +616,9 @@ public void nackMessageTest() throws Exception { Future producerFuture = produceClient.connect(produceSocket, URI.create(producerUri), produceRequest); assertTrue(producerFuture.get().isOpen()); - log.info("SEND"); + assertEquals(consumeSocket1.getReceivedMessagesCount(), 0); + assertEquals(consumeSocket2.getReceivedMessagesCount(), 0); + produceSocket.sendMessage(1); Thread.sleep(500); From c19e1c28d731f68ffa6fc5b9f43df40408c4e2bf Mon Sep 17 00:00:00 2001 From: yrsh Date: Wed, 20 Jan 2021 17:37:09 +0300 Subject: [PATCH 7/7] license header added --- .../proxy/SimpleConsumerMessageHandler.java | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/SimpleConsumerMessageHandler.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/SimpleConsumerMessageHandler.java index 45380c475d146..cc7302a777775 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/SimpleConsumerMessageHandler.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/SimpleConsumerMessageHandler.java @@ -1,3 +1,21 @@ +/** + * 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;