Fix Bug of ConsumerInterceptor on method onAcknowledge()
Motivation
To fix bug for
Modifications
Add call interceptor logic in markAckForBatchMessage method.
Result
Consumer acknowledge the batch message will call onAcknowledge() for every messages in batch message.
@@ -426,6 +426,8 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
&& !batchMessageId.getAcker().isPrevBatchCumulativelyAcked()) {
sendAcknowledge(batchMessageId.prevBatchMessageId(), AckType.Cumulative, properties);
batchMessageId.getAcker().setPrevBatchCumulativelyAcked(true);
+ } else {
+ onAcknowledge(batchMessageId, null);
}
if (log.isDebugEnabled()) {
log.debug("[{}] [{}] cannot ack message to broker {}, acktype {}, pending acks - {}", subscription,
