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 812116af56c60..a13a14e25e730 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 @@ -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 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()); + + 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); 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..cc7302a777775 --- /dev/null +++ b/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); +} 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 bf3a779dc4d99..f6917ccc8e971 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,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()); } } 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 53b61184d0b6c..c7987390e80bf 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 @@ -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 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/docs/client-libraries-websocket.md b/site2/docs/client-libraries-websocket.md index efb3e15ce2f88..d887e452ba030 100644 --- a/site2/docs/client-libraries-websocket.md +++ b/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