pulsar: [Bug] Pulsar shutdowns Sink connector on exception

Search before asking

• I searched in the issues and found nothing similar.

Version

• Pulsar 2.7.3 onwards

Minimal reproduce step

• Change any sink implementation for test and call record.fail() from the write method. The processing guarantee for the sink should be set to EFFECTIVELY_ONCE and auto ack should be false.

What did you expect to see?

For the sinks having EFFECTIVE_ONCE guarantee, prior to pulsar v2.7.3 , if there was any exception in record processing, the CompletableFuture callback was swallowing the RuntimeException. From v2.7.3, it is not processed in the callback hence caught by the run method of JavaInstanceRunnable which shutdowns the sink. Is this expected behavior? In this case, on failure, no further processing will happen as the sink is shutdown. Shouldn’t this exception be handled and the record be retried and allow other future message processing as well? At least, there should be a config option that would decide to shutdown the sink or keep it running.

What did you see instead?

Pulsar shutdowns the sink service hence that particular record and any further records are not processed by the sink.

Anything else?

No response

Are you willing to submit a PR?

• I’m willing to submit a PR!