recycled already - exception

[filthz]

We started getting the following exception in one of the consumers. It is occurring when the consumer.acknowledge() method is called. What is causing this and how can we fix this issue?

Exception:

java.lang.IllegalStateException: recycled already
	at io.netty.util.Recycler$WeakOrderQueue.transfer(Recycler.java:442)
	at io.netty.util.Recycler$Stack.scavengeSome(Recycler.java:587)
	at io.netty.util.Recycler$Stack.scavenge(Recycler.java:562)
	at io.netty.util.Recycler$Stack.pop(Recycler.java:535)
	at io.netty.util.Recycler.get(Recycler.java:162)
	at org.apache.pulsar.common.util.collections.ConcurrentBitSetRecyclable.create(ConcurrentBitSetRecyclable.java:51)
	at org.apache.pulsar.client.impl.PersistentAcknowledgmentsGroupingTracker.lambda$doIndividualBatchAckAsync$8(PersistentAcknowledgmentsGroupingTracker.java:328)
	at java.base/java.util.concurrent.ConcurrentHashMap.computeIfAbsent(ConcurrentHashMap.java:1705)
	at org.apache.pulsar.client.impl.PersistentAcknowledgmentsGroupingTracker.doIndividualBatchAckAsync(PersistentAcknowledgmentsGroupingTracker.java:323)
	at org.apache.pulsar.client.impl.PersistentAcknowledgmentsGroupingTracker.doIndividualBatchAck(PersistentAcknowledgmentsGroupingTracker.java:294)
	at org.apache.pulsar.client.impl.PersistentAcknowledgmentsGroupingTracker.doIndividualBatchAck(PersistentAcknowledgmentsGroupingTracker.java:287)
	at org.apache.pulsar.client.impl.PersistentAcknowledgmentsGroupingTracker.lambda$addAcknowledgment$4(PersistentAcknowledgmentsGroupingTracker.java:235)
	at org.apache.pulsar.client.impl.PersistentAcknowledgmentsGroupingTracker.addIndividualAcknowledgment(PersistentAcknowledgmentsGroupingTracker.java:220)
	at org.apache.pulsar.client.impl.PersistentAcknowledgmentsGroupingTracker.addAcknowledgment(PersistentAcknowledgmentsGroupingTracker.java:232)
	at org.apache.pulsar.client.impl.PersistentAcknowledgmentsGroupingTracker.addAcknowledgment(PersistentAcknowledgmentsGroupingTracker.java:196)
	at org.apache.pulsar.client.impl.ConsumerImpl.doAcknowledge(ConsumerImpl.java:560)
	at org.apache.pulsar.client.impl.ConsumerBase.doAcknowledgeWithTxn(ConsumerBase.java:688)
	at org.apache.pulsar.client.impl.ConsumerBase.acknowledgeAsync(ConsumerBase.java:634)
	at org.apache.pulsar.client.impl.ConsumerBase.acknowledgeAsync(ConsumerBase.java:619)
	at org.apache.pulsar.client.impl.ConsumerBase.acknowledge(ConsumerBase.java:416)
	at org.apache.pulsar.client.impl.ConsumerBase.acknowledge(ConsumerBase.java:409)
	at de.seepex.service.lastactive.PulsarLastActiveConsumer.messageListener(PulsarLastActiveConsumer.java:80)
	at org.apache.pulsar.client.impl.ConsumerBase.callMessageListener(ConsumerBase.java:1157)
	at org.apache.pulsar.client.impl.ConsumerBase.lambda$triggerListener$9(ConsumerBase.java:1120)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	at java.base/java.lang.Thread.run(Thread.java:834)

The consumer:

private void messageListener(Consumer<byte[]> consumer, Message<byte[]> msg) {
        try {
            byte[] payload = msg.getData();

            if (payload == null || payload.length == 0) {
                consumer.acknowledge(msg);
                return;
            }

            final LastActiveUpdate lastActiveUpdate = objectMapper.readValue(payload, LastActiveUpdate.class);
            lastActiveProcessor.process(lastActiveUpdate);
            consumer.acknowledge(msg);
        } catch (Exception e) {
            LOG.error("failed to process", e);
            consumer.acknowledgeAsync(msg);
        }
    }

[merlimat]

This looks to be a bug. Which version of the client are you using?

[filthz]

We are using the latest server / client combo.

Could this be caused by following scenario?

• Message consumed
• Consumer very slow (slower than the ackTimeout)
• Ack sent after ackTimeout

[merlimat]

I think we already have tests for that scenario, though a simple one works fine:

@Cleanup
        Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
                .topic(topic)
                .create();
        @Cleanup
        Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
                .topic(topic)
                .subscriptionName("sub")
                .subscriptionType(SubscriptionType.Shared)
                .ackTimeout(1, TimeUnit.SECONDS)
                .subscribe();

        producer.send("hello-1");
        producer.send("hello-2");

        Message<String> msg1 = consumer.receive();

        // Wait for ack-timeout to trigger
        Thread.sleep(10 * 1000);

        consumer.acknowledge(msg1);

[merlimat]

Can you share the exact client version you are using and the consumer configuration?

If you have an easy way to reproduce, it would be greatly appreciated.

[filthz]

Just checked the versions:

pulsar-client-admin-api:3.1.1
pulsar-client-api:3.1.1
pulsar-client-original:3.1.1
pulsar-common:3.1.1

Netty:

Consumer config:

@EventListener(ApplicationReadyEvent.class)
    public void init() {

        for(int i = 0; i < parallelConsumer; i++) {
            pulsarClient.newConsumer()
                    .topic(activityTopic)
                    .subscriptionName(lastActiveSub)
                    .consumerName(consumerName + "_" + randomString(8))
                    .subscriptionType(SubscriptionType.Key_Shared)
                    .enableBatchIndexAcknowledgment(true)
                    .subscriptionInitialPosition(SubscriptionInitialPosition.Latest)
                    .keySharedPolicy(KeySharedPolicy.autoSplitHashRange())
                    .messageListener(this::messageListener)
                    .startPaused(false)
                    .negativeAckRedeliveryBackoff(MultiplierRedeliveryBackoff.builder()
                            .minDelayMs(1000)
                            .maxDelayMs(60 * 1000)
                            .multiplier(2)
                            .build())
                    .ackTimeout(5, TimeUnit.MINUTES)
                    .subscribeAsync();
        }
    }

I am not able to reproduce as it is occurring randomly on production where we have quite a lot of messages

[dao-jun]

What's your pulsar client version?
This is a netty bug from a long time ago, it should be fixed in 2021 and released in 4.1.66.Final(netty version).
And I found the origin PR: netty/netty#11402

If you are using the latest pulsar client version, I think it might be a JAR package conflict, you need to check your JAR version(Pulsar client version, netty version) in your local env.

[filthz]

see above, im quite sure that my versions are fine

[lhotari]

see above, im quite sure that my versions are fine

You are using an old Netty version which contains the bug that was mentioned. pulsar-client-original doesn't come with a shaded Netty version. Could you switch to use pulsar-client dependency instead of pulsar-client-original? The other option is to use a recent Netty version that pulsar-client-original 3.1.1 supports.

[dao-jun]

JAR package conflict causes the bug.

Same as @lhotari mentioned , your Pulsar client version is fine, but netty version is incorrect.

Pulsar client 3.1.1 is using netty 4.1.94.Final, your netty version is 4.1.49.Final, It causes the recycled already exception.

I think upgrade netty to latest version or follow @lhotari 's advice will solve the problem.

[filthz]

Oh...thanks a lot for the advise!
Does this old version also cause the "entries since first not acknowleged message" to be stuck on a old value which then prevents offloading?

[dao-jun]

You mean tired-storage(https://pulsar.apache.org/docs/3.1.x/tiered-storage-overview/)?

I don't think there is any kind of relationship between them, message offloading is depends on managedLedgerOffloadThresholdInSeconds and managedLedgerOffloadAutoTriggerSizeThresholdBytes, they are not related to whether the message is Acked or not

[filthz]

Maybe I should not capsize this thread for different questions - but just in case it is related. Here is an example what I am mentioning (this is the consumer with the faulty netty version). It seems there is an unacked message in a old ledger and this (plus all following ledgers) wont be deleted even if they are offloaded. I guess because of the stuck message

[dao-jun]

Whether to delete the Ledger or not is not related to whether this Ledger has been offloaded, but related to RetentionPolicy.

For a topic, Pulsar will select the slowest subscription, and delete the ledgers which in bookkeeper and already consumed.

If a message is unAcked, it will indeed affect the deletion of the Ledgers, but will not affect the ledger offloading.

Ledgers' offloading and deleting are different mechanisms.