PulsarSpout to use consumer.redeliverUnacknowledgedMessages() ?

[jnioche]

The PulsarSpout keeps track of the pending messages as well as the failed ones in a queue and map. If the JVM containing a PulsarSpout crashes, this information will get lost with it and the pending or failed messages won't get acked in Pulsar but they also won't have a chance to get retried either when a new instance of the spout is started.

Should we add consumer.redeliverUnacknowledgedMessages() to the initialisation of the spout? That way any message which hasn't been acked will get retried?

[merlimat]

@jnioche Yes, that makes total sense. The only reason for this to not use redeliverUnacknowledgedMessages() in the spout was that the method didn't exist when we added the spout :).

Probably it would even be better to use the method at #2211 so that we can precisely control the redelivery.

Should we add consumer.redeliverUnacknowledgedMessages() to the initialisation of the spout?

When subscribing the unacked messages are automatically re-delivered. The only downside of redeliverUnacknowledgedMessages() would be that it triggers the re-delivery of unacked messages.

The most complicated part in the PulsarSpout is the caching of client and consumer instances. This was done to shared resources across threads when users set high parallelism in the topologies.

[jnioche]

Probably it would even be better to use the method at #2211 so that we can precisely control the redelivery.

We need both redeliverUnacknowledgedMessages() and #2211. The latter could be used for failed messages but we'd still need to keep track of the number of times it failed and a delay to apply. This won't help for messages which were being processed when the JVM died, which is why we need redeliverUnacknowledgedMessages() at init time.

When subscribing the unacked messages are automatically re-delivered.

I'm confused. Do you mean that this is what it does even without using redeliverUnacknowledgedMessages?

[tisonkun]

Closed as stale and no one worked on it. Please open a new issue if it's still relevant to the maintained versions.