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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
