[Bug] Delayed message isn't marked as acked when delayed time > messageTTL

[visxu]

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

OS: MAC 13.6.7
Java: 17
Pulsar: 3.0.6

Minimal reproduce step

1. Create a topic and set messageTTL 60s
2. Start a consumer on the topic
3. Send a delayed 120s message

What did you expect to see?

The consumer will not receive the delayed message.

What did you see instead?

Consumer consumed the delayed message.

Anything else?

visxu➜~/pulsar-cluster-3.0.6» ./pulsar-1/bin/pulsar-admin topics create public/default/topic1216                                                                                                                      [19:19:13]
visxu➜~/pulsar-cluster-3.0.6»                                                                                                                                                                                         [20:43:18]
visxu➜~/pulsar-cluster-3.0.6»                                                                                                                                                                                         [20:43:18]
visxu➜~/pulsar-cluster-3.0.6» ./pulsar-1/bin/pulsar-admin topics get-message-ttl public/default/topic1216                                                                                                             [20:43:19]
null
visxu➜~/pulsar-cluster-3.0.6»                                                                                                                                                                                         [20:43:53]
visxu➜~/pulsar-cluster-3.0.6»                                                                                                                                                                                         [20:43:53]
visxu➜~/pulsar-cluster-3.0.6»                                                                                                                                                                                         [20:43:53]
visxu➜~/pulsar-cluster-3.0.6»                                                                                                                                                                                         [20:43:53]
visxu➜~/pulsar-cluster-3.0.6» ./pulsar-1/bin/pulsar-admin topics set-message-ttl --ttl 60s public/default/topic1216                                                                                                   [20:45:21]
visxu➜~/pulsar-cluster-3.0.6» ./pulsar-1/bin/pulsar-admin topics get-message-ttl public/default/topic1216                                                                                                             [20:45:53]
60
visxu➜~/pulsar-cluster-3.0.6» ./pulsar-1/bin/pulsar-admin topics set-retention --size 10m --time 300s public/default/topic1216                                                                                        [20:47:09]
visxu➜~/pulsar-cluster-3.0.6» ./pulsar-1/bin/pulsar-admin topics get-retention public/default/topic1216                                                                                                               [20:47:33]
{
  "retentionTimeInMinutes" : 5,
  "retentionSizeInMB" : 10
}
visxu➜~/pulsar-cluster-3.0.6» ./pulsar-1/bin/pulsar-client consume -t Shared -s sub1 public/default/topic1216                                                                                                         [21:00:30]
2024-12-16T21:00:32,490+0800 [pulsar-client-io-1-3] INFO  org.apache.pulsar.client.impl.ConnectionPool - [[id: 0xebf0e25b, L:/127.0.0.1:54037 - R:localhost/127.0.0.1:16650]] Connected to server
2024-12-16T21:00:32,597+0800 [pulsar-client-io-1-3] INFO  org.apache.pulsar.client.impl.ConsumerStatsRecorderImpl - Starting Pulsar consumer status recorder with config: {"topicNames":["public/default/topic1216"],"topicsPattern":null,"subscriptionName":"sub1","subscriptionType":"Shared","subscriptionProperties":null,"subscriptionMode":"Durable","receiverQueueSize":1000,"acknowledgementsGroupTimeMicros":100000,"maxAcknowledgmentGroupSize":1000,"negativeAckRedeliveryDelayMicros":60000000,"maxTotalReceiverQueueSizeAcrossPartitions":50000,"consumerName":null,"ackTimeoutMillis":0,"tickDurationMillis":1000,"priorityLevel":0,"maxPendingChunkedMessage":10,"autoAckOldestChunkedMessageOnQueueFull":false,"expireTimeOfIncompleteChunkedMessageMillis":60000,"cryptoFailureAction":"FAIL","properties":{},"readCompacted":false,"subscriptionInitialPosition":"Latest","patternAutoDiscoveryPeriod":60,"regexSubscriptionMode":"PersistentOnly","deadLetterPolicy":null,"retryEnable":false,"autoUpdatePartitions":true,"autoUpdatePartitionsIntervalSeconds":60,"replicateSubscriptionState":false,"resetIncludeHead":false,"batchIndexAckEnabled":false,"ackReceiptEnabled":false,"poolMessages":true,"startPaused":false,"autoScaledReceiverQueueSizeEnabled":false,"topicConfigurations":[],"maxPendingChuckedMessage":10}
2024-12-16T21:00:32,604+0800 [pulsar-client-io-1-3] INFO  org.apache.pulsar.client.impl.ConsumerStatsRecorderImpl - Pulsar client config: {"serviceUrl":"pulsar://localhost:16650/","authPluginClassName":null,"authParams":null,"authParamMap":null,"operationTimeoutMs":30000,"lookupTimeoutMs":30000,"statsIntervalSeconds":60,"numIoThreads":8,"numListenerThreads":8,"connectionsPerBroker":1,"connectionMaxIdleSeconds":180,"useTcpNoDelay":true,"useTls":false,"tlsKeyFilePath":"","tlsCertificateFilePath":"","tlsTrustCertsFilePath":"","tlsAllowInsecureConnection":false,"tlsHostnameVerificationEnable":false,"concurrentLookupRequest":5000,"maxLookupRequest":50000,"maxLookupRedirects":20,"maxNumberOfRejectedRequestPerConnection":50,"keepAliveIntervalSeconds":30,"connectionTimeoutMs":10000,"requestTimeoutMs":60000,"readTimeoutMs":60000,"autoCertRefreshSeconds":300,"initialBackoffIntervalNanos":100000000,"maxBackoffIntervalNanos":60000000000,"enableBusyWait":false,"listenerName":null,"useKeyStoreTls":false,"sslProvider":null,"tlsKeyStoreType":"JKS","tlsKeyStorePath":"","tlsKeyStorePassword":"*****","tlsTrustStoreType":"JKS","tlsTrustStorePath":"","tlsTrustStorePassword":"*****","tlsCiphers":[],"tlsProtocols":[],"memoryLimitBytes":0,"proxyServiceUrl":null,"proxyProtocol":null,"enableTransaction":false,"dnsLookupBindAddress":null,"dnsLookupBindPort":0,"socks5ProxyAddress":null,"socks5ProxyUsername":null,"socks5ProxyPassword":null,"description":null}
2024-12-16T21:00:32,616+0800 [pulsar-client-io-1-6] INFO  org.apache.pulsar.client.impl.ConnectionPool - [[id: 0x9254fae9, L:/127.0.0.1:54038 - R:/127.0.0.1:36650]] Connected to server
2024-12-16T21:00:32,620+0800 [pulsar-client-io-1-6] INFO  org.apache.pulsar.client.impl.ConsumerImpl - [public/default/topic1216][sub1] Subscribing to topic on cnx [id: 0x9254fae9, L:/127.0.0.1:54038 - R:/127.0.0.1:36650], consumerId 0
2024-12-16T21:00:32,627+0800 [pulsar-client-io-1-6] INFO  org.apache.pulsar.client.impl.ConsumerImpl - [public/default/topic1216][sub1] Subscribed to topic on /127.0.0.1:36650 -- consumer: 0






----- got message -----
key:[null], properties:[], content:This is a test message, 12-16-2024 21:00:41
2024-12-16T21:02:41,645+0800 [pulsar-client-io-1-6] INFO  org.apache.pulsar.client.impl.ConsumerImpl - [public/default/topic1216] [sub1] Closed consumer
2024-12-16T21:02:41,646+0800 [main] INFO  org.apache.pulsar.client.impl.PulsarClientImpl - Client closing. URL: pulsar://localhost:16650/
2024-12-16T21:02:41,649+0800 [pulsar-client-io-1-6] INFO  org.apache.pulsar.client.impl.ClientCnx - [id: 0x9254fae9, L:/127.0.0.1:54038 ! R:/127.0.0.1:36650] Disconnected
2024-12-16T21:02:41,649+0800 [pulsar-client-io-1-3] INFO  org.apache.pulsar.client.impl.ClientCnx - [id: 0xebf0e25b, L:/127.0.0.1:54037 ! R:localhost/127.0.0.1:16650] Disconnected
2024-12-16T21:02:43,766+0800 [main] INFO  org.apache.pulsar.client.cli.PulsarClientTool - 1 messages successfully consumed

I sent a delayed 120s message with timestamp, but got it 2 mins later.

Are you willing to submit a PR?

• I'm willing to submit a PR!

[visxu]

BTW, according the doc, seems the delayed message should be auto acked if TTL expires.

https://pulsar.apache.org/docs/3.0.x/concepts-messaging/#delayed-message-delivery

[visxu]

Additionally, I didn't set TTL on namespace level.

visxu➜~/pulsar-cluster-3.0.6» ./pulsar-1/bin/pulsar-admin namespaces get-message-ttl public/default                                                                                                                   [21:27:52]
null
visxu➜~/pulsar-cluster-3.0.6»

[visxu]

@crossoverJie Could you help take a look if you're available ?

[Joforde]

I also found this problem in pulsar v4.0.0. Here's how I did it ：

bin/pulsar-admin --admin-url http://localhost:8080 namespaces set-message-ttl --messageTTL 30s public/default
bin/pulsar-admin --admin-url http://localhost:8080 namespaces get-message-ttl public/default

public class DelayMessageProducer {
private static final String serviceUrl = "pulsar://localhost:6650";
private static final String topicName = "persistent://public/default/test_delay_message";

    public static void main(String[] args) throws PulsarClientException {
        PulsarClient client = PulsarClient.builder().serviceUrl(serviceUrl).build();
        try (Producer<byte[]> producer = client.newProducer().topic(topicName).create()) {
            for (int i = 0; i < 100; i++) {
                MessageId messageId = producer.newMessage().deliverAfter(i, TimeUnit.SECONDS).value(("Hello Pulsar! " + i).getBytes(StandardCharsets.UTF_8)).send();
                System.out.println(messageId);
            }
        }
        client.close();
    }
}
public class DelayMessageConsumer {
    private static final String serviceUrl = "pulsar://localhost:6650";
    private static final String topicName = "persistent://public/default/test_delay_message";
    private static final String subscriptionName = "s0";

    public static void main(String[] args) throws PulsarClientException {
        PulsarClient client = PulsarClient.builder().serviceUrl(serviceUrl).build();
        Consumer<byte[]> consumer = client.newConsumer().topic(topicName).subscriptionName(subscriptionName).subscriptionType(SubscriptionType.Shared).subscriptionInitialPosition(SubscriptionInitialPosition.Earliest).subscribe();
        while (true) {
            Message<byte[]> message = consumer.receive();
            consumer.acknowledge(message);
            System.out.printf("%s,%s,%s\n", message.getPublishTime(), message.getMessageId(), new String(message.getData()));
        }
    }
}

[dao-jun]

TTL check interval's default value is 5 minutes. see messageExpiryCheckIntervalInMinutes