[pulsar-client] Avoid leak on publish failure on batch message
@@ -468,11 +468,14 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
log.debug("[{}] [{}] Closing out batch to accommodate large message with size {}", topic, producerName,
msg.getDataBuffer().readableBytes());
}
+ try {
batchMessageAndSend();
batchMessageContainer.add(msg, callback);
lastSendFuture = callback.getFuture();
+ } finally {
payload.release();
}
+ }
private boolean isValidProducerState(SendCallback callback) {
switch (getState()) {
