[feat] Add a new ack timeout mode where a callback is called instead of nacking the message

[lhotari]

Search before asking

• [X] I searched in the issues and found nothing similar.

Motivation

When a Pulsar application has ordered processing requirements, it's necessary to use either Failover, Exclusive or Key_Shared subscriptions. Ack timeouts shouldn't be used at all since this could cause messages to be processed in the wrong order. The application should take responsibility of handling possible error cases.

Since ack timeouts aren't used, there's a chance that the application logic contains a bug and the application doesn't acknowledge a message. Detecting this is very hard currently. It's hard to tell whether lost acks are caused by a Pulsar bug or feature or it's a problem caused by the application. This should be made easier.

Solution

Since Pulsar already contains the ack timeout feature, it would be natural to use it as the basis for detecting when the application is not acknowledging a message in time.

example of configuring the ack timeout handler:

consumerBuilder.ackTimeout(10, TimeUnit.SECOND)
               .ackTimeoutHandler((consumer, messageIds) -> messageIds.forEach(messageId -> log.warn("message with id {} wasn't acknowledged!", messageId))

interface:

import java.util.Set;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.MessageId;

public interface AckHandler {
    void handleAckTimeout(Consumer<?> consumer, Set<MessageId> messageIds);
}

Alternatives

Anything else?

Are you willing to submit a PR?

• [X] I'm willing to submit a PR!

[crossoverJie]

https://github.com/apache/pulsar-client-go/issues/403#issuecomment-735995132 https://github.com/apache/pulsar-client-go/issues/217#issuecomment-619384515

I have a doubt. Based on the discussion here, ackTimout is a legacy feature; do we still need to add new functionalities for this feature?

[lhotari]

apache/pulsar-client-go#403 (comment) apache/pulsar-client-go#217 (comment)

I have a doubt. Based on the discussion here, ackTimout is a legacy feature; do we still need to add new functionalities for this feature?

@crossoverJie Thanks for sharing. Perhaps we rename this completely. :) However, it doesn't take away the usefulness of having a way to detect when the application is possibly misbehaving and missing to acknowledge a message.