[Bug] when define Consumer in class field， and restart consumer, message received More than once

[sdvdxl]

Search before asking

• I searched in the issues and found nothing similar.

Read release policy

• I understand that unsupported versions don't get bug fixes. I will attempt to reproduce the issue on a supported version of Pulsar client and Pulsar broker.

Version

server: pulsar-3.2.2 standalone

client: java-client 3.2.3

Minimal reproduce step

java code:

1. enable testProducer(); disable: testConsumer(); and start
2. disable testProducer(); enable: testConsumer(); and restart
3. disable testProducer(); enable: testConsumer(); and restart
4. delete static Consumer<byte[]> consumer; and restart
5. delete static Consumer<byte[]> consumer; and restart

package top.todu.leaning.pulsar.sample;

import cn.hutool.core.date.DateTime;
import java.util.concurrent.TimeUnit;
import org.apache.pulsar.client.api.*;

/**
 * @author du 2024/5/20 14:48
 */
public class PulsarSample {
  private static PulsarClient pulsarClient;
  private static String host = "pulsar://10.10.81.28:6650";
  private static String topic = "data_sync_queue_test";
  static Consumer<byte[]> consumer;

  public static void main(String[] args) throws Exception {
    pulsarClient = PulsarClient.builder().serviceUrl(host).build();

    testConsumer();
    testProducer();
    System.out.println("-=-------");
    TimeUnit.SECONDS.sleep(10);
    consumer.close();
    pulsarClient.close();
  }

  private static void testProducer() throws PulsarClientException {
    Producer<byte[]> producer =
        pulsarClient
            .newProducer()
            .topic(topic)
            .messageRoutingMode(MessageRoutingMode.SinglePartition)
            .create();
    for (int i = 0; i < 5; i++) {
      String msg = i + " Hello, Pulsar! " + DateTime.now();
      System.out.println("send msg： " + msg);
      producer.newMessage().value(msg.getBytes()).key("const").send();
    }
    producer.close();
  }

  private static void testConsumer() throws PulsarClientException {
    consumer =
        pulsarClient
            .newConsumer()
            .subscriptionName("data_sync_queue_test")
            .topic(topic)
            .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
            .subscriptionType(SubscriptionType.Key_Shared)
            .messageListener(
                new MessageListener<byte[]>() {
                  @Override
                  public void received(Consumer<byte[]> consumer, Message<byte[]> msg) {
                    System.out.println("Received message: " + new String(msg.getData()));
                    try {
                      consumer.acknowledge(msg);
                    } catch (PulsarClientException e) {
                      throw new RuntimeException(e);
                    }
                  }
                })
            .subscribe();
  }

  

}

What did you expect to see?

message should receive only once

What did you see instead?

if defined static Consumer<byte[]> consumer;, restart application, message will receive more than once.

if delete the field, it is good.

Anything else?

No response

Are you willing to submit a PR?

• I'm willing to submit a PR!

[sdvdxl]

this may be a mistake. other place used consumer.unsubscribe()