Spring/spring-pulsar
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Remove duplicate handleBatchAcks method

Browse Source
This commit is contained in:
Chris Bono
2024-04-19 11:22:02 -05:00
parent 7fc1fa0479
commit 01e96ddfa3
1 changed files with 25 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

59
spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java
View File

@@ -529,7 +529,7 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
}
// All the records are processed at this point - handle acks
if (this.ackMode.equals(AckMode.BATCH)) {
handleBatchAcks(messages, null);
handleBatchAcksForRecordListener(messages, null);
}
}
@@ -670,21 +670,7 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
this.batchMessageListener.received(this.consumer, messageList);
}
if (this.ackMode.equals(AckMode.BATCH)) {
try {
if (isSharedSubscriptionType()) {
AckUtils.handleAck(this.consumer, messages, txn);
}
else {
Stream<Message<T>> stream = StreamSupport.stream(messages.spliterator(), true);
Message<T> last = stream.reduce((a, b) -> b).orElse(null);
AckUtils.handleAckCumulative(this.consumer, last, txn);
}
}
catch (PulsarException pe) {
DefaultPulsarMessageListenerContainer.this.logger.warn(pe,
() -> "Batch acknowledgment failed: " + pe.getMessage());
this.consumer.negativeAcknowledge(messages);
}
handleBatchAcks(messages, txn);
}
if (this.pulsarConsumerErrorHandler != null) {
pendingMessagesHandledSuccessfully(inRetryMode, messagesPendingInBatch);
@@ -793,25 +779,9 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
|| this.subscriptionType.equals(SubscriptionType.Key_Shared));
}
private void handleBatchAcks(Messages<T> messages, @Nullable Transaction txn) {
private void handleBatchAcksForRecordListener(Messages<T> messages, @Nullable Transaction txn) {
if (this.nackableMessages.isEmpty()) {
try {
if (messages.size() > 0) {
if (isSharedSubscriptionType()) {
AckUtils.handleAck(this.consumer, messages, txn);
}
else {
Stream<Message<T>> stream = StreamSupport.stream(messages.spliterator(), true);
Message<T> last = stream.reduce((a, b) -> b).orElse(null);
AckUtils.handleAckCumulative(this.consumer, last, txn);
}
}
}
catch (PulsarException pe) {
DefaultPulsarMessageListenerContainer.this.logger.warn(pe,
() -> "Batch acks failed: " + pe.getMessage());
this.consumer.negativeAcknowledge(messages);
}
handleBatchAcks(messages, txn);
}
else {
for (Message<T> message : messages) {
@@ -826,6 +796,27 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
}
}
private void handleBatchAcks(Messages<T> messages, @Nullable Transaction txn) {
if (messages.size() <= 0) {
return;
}
try {
if (isSharedSubscriptionType()) {
AckUtils.handleAck(this.consumer, messages, txn);
}
else {
Stream<Message<T>> stream = StreamSupport.stream(messages.spliterator(), true);
Message<T> last = stream.reduce((a, b) -> b).orElse(null);
AckUtils.handleAckCumulative(this.consumer, last, txn);
}
}
catch (PulsarException pe) {
DefaultPulsarMessageListenerContainer.this.logger.warn(pe,
() -> "Batch acknowledgment failed: " + pe.getMessage());
this.consumer.negativeAcknowledge(messages);
}
}
private void handleAck(Message<T> message, @Nullable Transaction txn) {
AckUtils.handleAckWithNackOnFailure(this.consumer, message.getMessageId(), txn);
}