[Bug] Pulsar shutdowns Sink connector on exception

### Search before asking

- [X] I searched in the [issues](https://github.com/apache/pulsar/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](https://github.com/apache/pulsar/blob/v2.7.2/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java#L345) was swallowing the [RuntimeException](https://github.com/apache/pulsar/blob/v2.7.2/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java#L376). From v2.7.3, it is [not processed in the callback](https://github.com/apache/pulsar/blob/v2.7.3/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java#L349) hence [caught](https://github.com/apache/pulsar/blob/v2.7.3/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java#L277) 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?

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

The linked JavaInstanceRunnable locations in v2.7.2 and v2.7.3 are the entry points; compare callback handling with the run method. Reproduce using a sink that calls record.fail() with EFFECTIVELY_ONCE and auto ack false, then use the expected retry-versus-shutdown behavior to define the regression test and completion criteria.

Written by the indexing model from the issue text.