[GitHub] [pulsar] hrsakai commented on a change in pull request #11216: [Websocket] Support cumulative acknowledge for Pulsar Websocket consumer.

[GitBox <gi...@apache.org>]

hrsakai commented on a change in pull request #11216:
URL: https://github.com/apache/pulsar/pull/11216#discussion_r713505146



##########
File path: pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ProducerHandler.java
##########
@@ -187,20 +195,32 @@ public void onWebSocketText(String message) {
 
         final long now = System.nanoTime();
 
-        builder.sendAsync().thenAccept(msgId -> {
-            updateSentMsgStats(msgSize, TimeUnit.NANOSECONDS.toMicros(System.nanoTime() - now));
-            if (isConnected()) {
-                String messageId = Base64.getEncoder().encodeToString(msgId.toByteArray());
-                sendAckResponse(new ProducerAck(messageId, sendRequest.context));
+        if (!consumerCumulativeAck || consumerPullMode) {
+            builder.sendAsync().thenAccept(msgId -> sendMessageSucceed(msgSize, now, sendRequest, msgId))
+                .exceptionally(exception -> {
+                    log.warn("[{}] Error occurred while producer handler was sending msg from {}: {}", producer.getTopic(),
+                            getRemote().getInetSocketAddress().toString(), exception.getMessage());
+                    numMsgsFailed.increment();
+                    sendAckResponse(
+                            new ProducerAck(UnknownError, exception.getMessage(), null, sendRequest.context));
+                    return null;
+                }
+            );
+        } else {
+            // if consumer not using pull mode but still want to use cumulative ack, then produce and cumulative ack
+            // can't happen at the same time as produce will also affect backlog msg count which will affect
+            // consumer permit calculation
+            try {

Review comment:
       Do we still need to make changes to producerHandler?




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org