Add debug logging when consumer fails to send permits to Broker
@@ -866,10 +866,21 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
if (log.isDebugEnabled()) {
log.debug("[{}] [{}] Adding {} additional permits", topic, subscription, numMessages);
}
-
+ if (log.isDebugEnabled()) {
+ cnx.ctx().writeAndFlush(Commands.newFlow(consumerId, numMessages))
+ .addListener(writeFuture -> {
+ if (!writeFuture.isSuccess()) {
+ log.debug("Consumer {} failed to send {} permits to broker: {}", consumerId, numMessages,
+ writeFuture.cause().getMessage());
+ } else {
+ log.debug("Consumer {} sent {} permits to broker", consumerId, numMessages);
+ }
+ });
+ } else {
cnx.ctx().writeAndFlush(Commands.newFlow(consumerId, numMessages), cnx.ctx().voidPromise());
}
}
+ }
@Override
public void connectionFailed(PulsarClientException exception) {
