[Bug] Client with shared subscription is blocked

[michalcukierman]

Search before asking

• I searched in the issues and found nothing similar.

Version

Client - 3.1.0
Pulsar - 3.1.0 (and later builds)

Also reported on 3.0.1

Minimal reproduce step

My reproducible steps:

1. Create persistent topic with 3 partitions
2. Publish 1 mln messages (30KB)
3. Run the client and consumer:

    PulsarClient client = PulsarClient.builder()
        .serviceUrl(this.pulsarBrokerUrl)
        .build();

    Consumer consumer = client.newConsumer()
        .topic(sourceTopic)
        .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
        .subscriptionName(subscriptionName)
        .subscriptionType(SubscriptionType.Shared)
        .receiverQueueSize(8)
        .ackTimeout(5, TimeUnit.SECONDS)
        .subscribe();

What did you expect to see?

All messages are received

What did you see instead?

Client stops to receive messages, restart client helps, but it get stuck after some time.

Anything else?

The issue was originally created described here: #21082
@MichalKoziorowski-TomTom also faces the issue.

I've created new issue, because it in #21082 the author says that broker restart helps. In case of this issue, it looks like it's client related and some race condition observed in 3.x.x. after introducing ackTimeout

Are you willing to submit a PR?

• I'm willing to submit a PR!

[codelipenghui]

@michalcukierman Could you please share the topic stats and internal-stats when the issue is reproduced?

[michalcukierman]

Two files attached:
partitioned.stats.internal.txt
partitioned.stats.txt

I see that in stats we have:

        "availablePermits" : 0,
        "unackedMessages" : 25,

        "availablePermits" : 0,
        "unackedMessages" : 32,

, but I do ack all of the messages:

    Consumer consumer = client.newConsumer()
        .topic(sourceTopic)
        .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
        .subscriptionName(subscriptionName)
        .subscriptionType(SubscriptionType.Shared)
        .receiverQueueSize(8)
        .ackTimeout(ackTimeout, TimeUnit.SECONDS)
        .subscribe();

    Producer<byte[]> producer = client.newProducer(Schema.BYTES)
        .topic(destinationTopic)
        .compressionType(CompressionType.LZ4)
        .maxPendingMessages(8)  // Support 5 MB files
        .blockIfQueueFull(true) // Support 5 MB files
        .batchingMaxBytes(5242880)
        .create();

    Multi<Messagemessages = Multi.createBy().repeating()
        .completionStage(consumer::receiveAsync)
        .until(m -> closed.get());

    messages.subscribe().with(msg -> {
      receivedDistribution.record(getKiloBytesSize(msg.getData()));
      Uni.createFrom().completionStage(producer.newMessage(Schema.BYTES)
              .key(msg.getKey())
              .value(msg.getValue().getContent().getPayload().toByteArray())
              .eventTime(msg.getEventTime()).sendAsync())
          .subscribe().with(msgId -> {
            sentCounter.increment();
            try {
              consumer.acknowledge(msg.getMessageId());
            } catch (PulsarClientException e) {
              throw new RuntimeException("Unable to send or ACK the message", e);
            }
          });
    });
  }

askTimeout is set to 120 seconds.

It can be confirmed in the log files, where sentCounter is just on message behind

[michalcukierman]

Here are the alternative stats:
#21082 (comment)

[michalcukierman]

I think it happens:

• there is ackTimeout set on a consumer
• high throughput
• the receviverQueue size is low

There may be a race condition in 3.1.0 client, as the situation was not observed with 2.10.4 (we've downgraded, also @MichalKoziorowski-TomTom reported this as a fix).

[MichalKoziorowski-TomTom]

We don't have ackTimeout and it still happens. In my case, my topic has only one partition.

[redoc123]

you said 1 million messages right? Could you post your sample payload? I want to reproduce and debug.

[michalcukierman]

It's 1 mln x 30 kb of sample HTML file. Any text file would be good.
What we are doing:

topic:source ( 1mln x 30k) -> app:router -> topic:destination

source has 3 partitions
destination has 6 partitions
Router has 2 replicas, uses shared subscription to read from source, writing to destination and ack's incoming message once the message is stored and acked on destination

[michalcukierman]

Here is the exact payload:
30k.html.zip

[lvdelu]

+1,it seems also occurred in other consumer mode(shared,keyshared)

[michalcukierman]

Happened today again on different module.

topic - not partitioned, without retention
messages - 50k, around 5k each
subscription - shared, with two clients

The solution is to use 2.10.4 client for now.

[mattisonchao]

I'll try to reproduce it and get back to you. :)