Duplicate Message Consumption in Spring-Pulsar Consumer (There is nothing wrong with the official Pulsar client.)

[Programmer-yyds]

Description

During testing, it was found that Spring-Pulsar’s ack handling behaves inconsistently compared to the native Pulsar Client, Spring-Pulsar There is a problem of repeated consumption..

Comparison logic is as follows:

1. Consumer ack timeout is set to 8 seconds (ackTimeout = 8s / ackTimeoutMillis = 8000).
2. Business logic simulates a 5-second block, theoretically completing ack within the timeout period.
3. Expected behavior: Business execution time < timeout → message should be properly acked, no redelivery should occur.
4. Actual result:
   • With native Pulsar Client → works as expected, message is acked properly, no duplicate consumption.
   • With Spring-Pulsar (@PulsarListener, ackMode = RECORD) → even when business processing time < timeout, duplicate consumption still occurs, indicating that Spring-Pulsar’s ack logic might not be taking effect in time.

---

Environment Information

• Pulsar Version: apachepulsar/pulsar:3.3.0 (standalone mode)
• Spring Boot Version: 3.4.7
• Spring-Pulsar Version: 3.4.7
• pulsar-client-all版本： Directly using the Spring-Pulsar memory-dependent package (3.3.7).
• JDK Version: 21
• Deployment Environment: Local (pulsar://localhost:6650)

Attachments：

pulsar-consume4.zip

A complete demo has been uploaded (including pom.xml, Producer, Native Consumer, Spring-Pulsar Consumer, and log files) that can be run directly to reproduce the issue.

Project Structure

├── log
│   ├── PulsarConsume.log             # Native Pulsar Client consumer log
│   └── SpringPulsarConsume.log       # Spring-Pulsar consumer log
├── pom.xml
├── src
│   ├── main
│   │   ├── java
│   │   │   └── org
│   │   │       └── pulsar
│   │   │           └── consume
│   │   │               ├── PulsarConsumeApplication.java
│   │   │               ├── original
│   │   │               │   └── PulsarConsume.java          # Native Pulsar Client consumer
│   │   │               └── spring
│   │   │                   └── SpringPulsarConsume.java    # Spring-Pulsar consumer
│   │   └── resources
│   │       └── application.yml
│   └── test
│       └── java
│           └── org
│               └── pulsar
│                   └── consume
│                       └── PulsarConsumeApplicationTest.java # Producer

Reproduction Code

Native Pulsar Client (Works Fine, No Duplicate Consumption)

Consumer<String> consumer = client.newConsumer(Schema.STRING)
    .topic("persistent://public/pulsar/mys-test")
    .subscriptionName("my-shared-sub")
    .subscriptionType(SubscriptionType.Shared)
    .ackTimeout(8, TimeUnit.SECONDS)
    .receiverQueueSize(10)
    .subscribe();

while (true) {
    Message<String> msg = consumer.receive();
    try {
        log.info("Message received: {}", msg.getValue());
        Thread.sleep(5000); // Simulate business processing
        consumer.acknowledge(msg); // Ack prevents duplicate consumption
    } catch (Exception e) {
        consumer.negativeAcknowledge(msg);
    }
}

Spring-Pulsar Consumer (Duplicate Consumption Occurs)

@PulsarListener(
    topics = "mys-test",
    ackMode = AckMode.RECORD,
    subscriptionType = SubscriptionType.Shared,
    schemaType = SchemaType.STRING,
    properties = {"ackTimeoutMillis=8000", "receiverQueueSize=10"}
)
public void consumeString2(Message<String> message) throws InterruptedException {
    log.info("{} message: {} Thread:{}", message.getTopicName(), message.getValue(), Thread.currentThread().getName());
    if (messageSet.contains(message.getValue())) {
        log.info("Repeated consumption, TopicName: {} message: {}", message.getTopicName(), message.getValue());
    }
    messageSet.add(message.getValue());
    TimeUnit.SECONDS.sleep(5); // Simulate business processing
}

[oneby-wang]

The reason is in Spring DefaultPulsarMessageListenerContainer consumption code:

checkIfPausedAndHandleAccordingly();
// Always receive messages in batch mode.
try {
    if (!inRetryMode.get() && !messagesPendingInBatch.get()) {
       DefaultPulsarMessageListenerContainer.this.receiveInProgress.set(true);
       if (!isPaused()) {
          messages = this.consumer.batchReceive();
       }
    }
}

In pulsar official client, you should write code like this:

while (true) {
    Messages<String> messages = consumer.batchReceive();
    for (Message<String> msg : messages) {
        if (msg == null) {
            continue;
        }
        try {
            log.info("Message received: {} Thread:{}", msg.getValue(), Thread.currentThread().getName());
            if (messageSet.contains(msg.getValue())) {
                log.info("Repeated consumption message: {} Thread:{}", msg.getValue(),
                    Thread.currentThread().getName());
            }
            messageSet.add(msg.getValue());
            // Simulation processing time
            Thread.sleep(5000);
            consumer.acknowledge(msg);
        } catch (Exception e) {
            consumer.negativeAcknowledge(msg);
        }
    }
}

And then message re-delivery will be re-produced.

15:51:56.192 [main] INFO org.pulsar.consume.PulsarConsumeApplication -- Message received: hello2 Thread:main
15:52:01.195 [main] INFO org.pulsar.consume.PulsarConsumeApplication -- Message received: hello3 Thread:main
15:52:04.694 [pulsar-timer-22-1] INFO org.apache.pulsar.client.impl.UnAckedMessageTracker -- [ConsumerBase{subscription='my-shared-sub', consumerName='VPsp9', topic='persistent://public/default/mys-test'}] 9 messages will be re-delivered
15:52:06.196 [main] INFO org.pulsar.consume.PulsarConsumeApplication -- Message received: hello4 Thread:main

Spring Pulsar doesn't use single Consumer.receive() method, which I think should be supported, because Pulsar client already supports pre-fetch feature.
@onobc Hi, Chris, what do you think about adding an field in @PulsarListener annotation to support single receive operation? Using batchReceive method indeed affects re-delivery policies, because pulsar client calculate re-delivery time elapse after the messages are polled out from pulsar pre-fetch queue.

[Programmer-yyds]

Thanks a lot @oneby-wang for your helpful reply 🙏

While using @PulsarListener, I noticed that the batch attribute does not behave as expected.

According to the annotation definition and documentation:

@Target({ ElementType.METHOD, ElementType.ANNOTATION_TYPE })
@Retention(RetentionPolicy.RUNTIME)
@MessageMapping
public @interface PulsarListener {

    /**
     * Whether the listener is a batch listener.
     * <p>Default is {@code false}.
     */
    boolean batch() default false;
}

• Default (batch=false) → should represent single-message listener mode, where the container pulls messages with consumer.receive() and delivers them one by one.
• Explicit (batch=true) → should enable batch mode, where the container uses consumer.batchReceive() and passes a Messages<?> object to the listener.

However, in the implementation (DefaultPulsarMessageListenerContainer), the code always calls:

// Always receive messages in batch mode.
messages = this.consumer.batchReceive();

This means that batch=false is effectively ignored, and single-message mode is not respected.

---

Impact

1. Users who configure @PulsarListener(batch=false) still get batch-based pulling.
2. Because of consumer.batchReceive(), the redelivery backoff strategy (e.g., RedeliveryBackoff) does not work as intended:
   • Messages are grouped in a batch and re-fetched together.
   • This may lead to duplicate message consumption when retries happen, even if only one message in the batch fails.
3. This behavior differs from the intended semantics of Pulsar client and may confuse users relying on timeout-based redelivery.

---

Expected Behavior

• When batch=false, the container should call consumer.receive().
• When batch=true, the container should call consumer.batchReceive().
• This ensures that single-message mode works correctly, and the redelivery backoff strategy behaves consistently without causing duplicate consumption.

---

Suggestion

Please consider aligning the implementation with the annotation’s contract and the Pulsar client’s behavior.
This would avoid unexpected redelivery issues and make @PulsarListener(batch=false) meaningful.

@onobc could you please take a look and consider fixing this? 🙏
Thanks again!

[oneby-wang]

Thanks a lot @oneby-wang for your helpful reply 🙏

While using @PulsarListener, I noticed that the batch attribute does not behave as expected.

According to the annotation definition and documentation:

@Target({ ElementType.METHOD, ElementType.ANNOTATION_TYPE })
@Retention(RetentionPolicy.RUNTIME)
@MessageMapping
public @interface PulsarListener {

    /**
     * Whether the listener is a batch listener.
     * <p>Default is {@code false}.
     */
    boolean batch() default false;
}

• Default (batch=false) → should represent single-message listener mode, where the container pulls messages with consumer.receive() and delivers them one by one.
• Explicit (batch=true) → should enable batch mode, where the container uses consumer.batchReceive() and passes a Messages<?> object to the listener.

However, in the implementation (DefaultPulsarMessageListenerContainer), the code always calls:

// Always receive messages in batch mode.
messages = this.consumer.batchReceive();

This means that batch=false is effectively ignored, and single-message mode is not respected.

Impact

1. Users who configure @PulsarListener(batch=false) still get batch-based pulling.

2. Because of consumer.batchReceive(), the redelivery backoff strategy (e.g., RedeliveryBackoff) does not work as intended:

   • Messages are grouped in a batch and re-fetched together.
   • This may lead to duplicate message consumption when retries happen, even if only one message in the batch fails.

3. This behavior differs from the intended semantics of Pulsar client and may confuse users relying on timeout-based redelivery.

Expected Behavior

• When batch=false, the container should call consumer.receive().
• When batch=true, the container should call consumer.batchReceive().
• This ensures that single-message mode works correctly, and the redelivery backoff strategy behaves consistently without causing duplicate consumption.

Suggestion

Please consider aligning the implementation with the annotation’s contract and the Pulsar client’s behavior. This would avoid unexpected redelivery issues and make @PulsarListener(batch=false) meaningful.

@onobc could you please take a look and consider fixing this? 🙏 Thanks again!

Hi, @onobc if the following behavior is expected(I think it's reasonable), I can help fix this.
When batch=false, the container should call consumer.receive().
When batch=true, the container should call consumer.batchReceive().
This ensures that single-message mode works correctly, and the redelivery backoff strategy behaves consistently without causing duplicate consumption.

[oneby-wang]

Seems reactive consumer doesn't have this problem.

spring-pulsar/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarConsumerFactory.java

Lines 76 to 89 in 4253a92

	public ReactiveMessageConsumer<T> createConsumer(Schema<T> schema,
	List<ReactiveMessageConsumerBuilderCustomizer<T>> customizers) {
	ReactiveMessageConsumerBuilder<T> consumerBuilder = this.reactivePulsarClient.messageConsumer(schema);
	// Apply the default customizers
	if (!CollectionUtils.isEmpty(this.defaultConfigCustomizers)) {
	this.defaultConfigCustomizers.forEach((customizer -> customizer.customize(consumerBuilder)));
	}
	// Apply the user specified customizers
	if (!CollectionUtils.isEmpty(customizers)) {
	customizers.forEach((c) -> c.customize(consumerBuilder));
	}
	this.ensureTopicNamesFullyQualified(consumerBuilder);
	return consumerBuilder.build();
	}

Pulsar official reactive consumer: org.apache.pulsar.reactive.client.internal.adapter.AdaptedReactiveMessageConsumer
class AdaptedReactiveMessageConsumer implements ReactiveMessageConsumer {

static <T> Mono<Message<T>> readNextMessage(Consumer<T> consumer) {
   return PulsarFutureAdapter.adaptPulsarFuture(consumer::receiveAsync);
}