[Bug] function Interrupted for unknown reasons

### Search before asking

- [X] I searched in the [issues](https://github.com/apache/pulsar/issues) and found nothing similar.

### Version

Pulsar & Function 3.1.1

### Minimal reproduce step

running with effectively once

### What did you expect to see?

function can handle InterruptedException

### What did you see instead?

When function is running, the following exception will occasionally occur, and function reading and writing are stuck.

```java
023-12-19T23:55:03,134+0800 [pulsar-external-listener-6-1] ERROR org.apache.pulsar.client.impl.ConsumerBase - [persistent://pulsar/log_Production/alltables-partition-13][sink-v2] Message listener error in processing message: 777380:54372:13
java.lang.RuntimeException: java.lang.InterruptedException
at org.apache.pulsar.functions.source.PushPulsarSource.consume(PushPulsarSource.java:64) ~[io.streamnative-pulsar-functions-instance-3.0.1.3.jar:3.0.1.3]
at org.apache.pulsar.functions.source.MultiConsumerPulsarSource.received(MultiConsumerPulsarSource.java:81) ~[io.streamnative-pulsar-functions-instance-3.0.1.3.jar:3.0.1.3]
at org.apache.pulsar.client.impl.ConsumerBase.callMessageListener(ConsumerBase.java:1153) ~[io.streamnative-pulsar-client-original-3.0.1.3.jar:3.0.1.3]
at org.apache.pulsar.client.impl.ConsumerBase.lambda$triggerListener$10(ConsumerBase.java:1119) ~[io.streamnative-pulsar-client-original-3.0.1.3.jar:3.0.1.3]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) ~[?:?]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) ~[?:?]
at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) ~[io.netty-netty-common-4.1.94.Final.jar:4.1.94.Final]
at java.lang.Thread.run(Thread.java:833) ~[?:?]
Caused by: java.lang.InterruptedException
at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:1638) ~[?:?]
at java.util.concurrent.LinkedBlockingQueue.put(LinkedBlockingQueue.java:343) ~[?:?]
at org.apache.pulsar.functions.source.PushPulsarSource.consume(PushPulsarSource.java:62) ~[io.streamnative-pulsar-functions-instance-3.0.1.3.jar:3.0.1.3]
... 7 more
```

According to the log, the close interface was called and the client was closed.
```java
2023-12-27T10:01:14,985+0800 [Thread-3] INFO function-functions-pro - start to close all the producers
2023-12-27T10:01:14,988+0800 [Thread-3] INFO function-functions-pro - finish to close all the producers
2023-12-27T10:01:14,988+0800 [Thread-3] INFO org.apache.pulsar.client.impl.PulsarClientImpl - Client closing. URL: pulsar://21.24.16.111:6650
2023-12-27T10:01:14,988+0800 [pulsar/log_Production/functions-pro-0] INFO org.apache.pulsar.functions.instance.JavaInstanceRunnable - Encountered exception in sink write:
org.apache.pulsar.client.api.PulsarClientException: java.lang.InterruptedException
org.apache.pulsar.client.api.PulsarClientException.unwrap(PulsarClientException.java:1031) ~[java-instance.jar:?]
org.apache.pulsar.client.impl.ProducerBuilderImpl.create(ProducerBuilderImpl.java:89) ~[io.streamnative-pulsar-client-original-3.0.1.3.jar:3.0.1.3]
com.tencent.DynamicTopicSink.getProducer(DynamicTopicSink.java:164) ~[8Ia5QqpY9lE3mwskma2zRA/:?]
com.tencent.DynamicTopicSink.write(DynamicTopicSink.java:142) ~[8Ia5QqpY9lE3mwskma2zRA/:?]
org.apache.pulsar.functions.instance.JavaInstanceRunnable.sendOutputMessage(JavaInstanceRunnable.java:429) ~[io.streamnative-pulsar-functions-instance-3.0.1.3.jar:?]
org.apache.pulsar.functions.instance.JavaInstanceRunnable.handleResult(JavaInstanceRunnable.java:391) ~[io.streamnative-pulsar-functions-instance-3.0.1.3.jar:?]
org.apache.pulsar.functions.instance.JavaInstanceRunnable.run(JavaInstanceRunnable.java:331) ~[io.streamnative-pulsar-functions-instance-3.0.1.3.jar:?]
java.lang.Thread.run(Thread.java:833) ~[?:?]
Caused by: java.lang.InterruptedException
java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:386) ~[?:?]
java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2073) ~[?:?]
org.apache.pulsar.client.impl.ProducerBuilderImpl.create(ProducerBuilderImpl.java:87) ~[io.streamnative-pulsar-client-original-3.0.1.3.jar:3.0.1.3]
\t... 6 more
2023-12-27T10:01:14,989+0800 [pulsar-client-io-17-5] INFO org.apache.pulsar.client.impl.ProducerImpl - [persistent://pulsar/log_Production/EquipInfo-partition-3] [sink-EquipInfo] Closed Producer
2023-12-27T10:01:14,989+0800 [Thread-3] INFO org.apache.pulsar.client.impl.ProducerImpl - [persistent://pulsar/log_Production/CombatEffectivenessFlow-partition-4] [sink-CombatEffectivenessFlow] Closed Producer (not connected)
2023-12-27T10:01:14,990+0800 [Thread-3] INFO org.apache.pulsar.client.impl.ProducerImpl - [persistent://pulsar/log_Production/CombatEffectivenessFlow-partition-1] [sink-CombatEffectivenessFlow] Closed Producer (not connected)
2023-12-27T10:01:14,990+0800 [Thread-3] INFO org.apache.pulsar.client.impl.ProducerImpl - [persistent://pulsar/log_Production/CombatEffectivenessFlow-partition-3] [sink-CombatEffectivenessFlow] Closed Producer (not connected)
2023-12-27T10:01:14,990+0800 [Thread-3] INFO org.apache.pulsar.client.impl.ProducerImpl - [persistent://pulsar/log_Production/CombatEffectivenessFlow-partition-2] [sink-CombatEffectivenessFlow] Closed Producer (not connected)
2023-12-27T10:01:14,990+0800 [pulsar-client-io-17-5] INFO org.apache.pulsar.client.impl.ProducerImpl - [persistent://pulsar/log_Production/shenqibianhua-partition-4] [sink-shenqibianhua] Closed Producer
2023-12-27T10:01:14,990+0800 [pulsar-client-io-17-5] INFO org.apache.pulsar.client.impl.ProducerImpl - [persistent://pulsar/log_Production/shenqibianhua-partition-1] [sink-shenqibianhua] Closed Producer
....
2023-12-27T10:01:14,991+0800 [pulsar-client-io-17-9] INFO org.apache.pulsar.client.impl.ProducerImpl - [persistent://pulsar/log_Production/EquipInfo-partition-0] [sink-EquipInfo] Closed Producer
2023-12-27T10:01:14,991+0800 [pulsar-client-io-17-7] INFO org.apache.pulsar.client.impl.PartitionedProducerImpl - [persistent://pulsar/log_Production/shenqibianhua] Closed Partitioned Producer
2023-12-27T10:01:14,991+0800 [pulsar-client-io-17-9] INFO org.apache.pulsar.client.impl.PartitionedProducerImpl - [persistent://pulsar/log_Production/EquipInfo] Closed Partitioned Producer
2023-12-27T10:01:14,991+0800 [pulsar/log_Production/functions-pro-0] ERROR org.apache.pulsar.functions.instance.JavaInstanceRunnable - [pulsar/log_Production/functions-pro:0] Uncaught exception in Java Instance
java.lang.RuntimeException: Failed to process message: 1306487:441847:0
org.apache.pulsar.functions.source.PulsarSource.lambda$buildRecord$6(PulsarSource.java:155) ~[io.streamnative-pulsar-functions-instance-3.0.1.3.jar:3.0.1.3]
org.apache.pulsar.functions.source.PulsarRecord.fail(PulsarRecord.java:133) ~[io.streamnative-pulsar-functions-instance-3.0.1.3.jar:3.0.1.3]
org.apache.pulsar.functions.instance.JavaInstanceRunnable.sendOutputMessage(JavaInstanceRunnable.java:434) ~[io.streamnative-pulsar-functions-instance-3.0.1.3.jar:?]
org.apache.pulsar.functions.instance.JavaInstanceRunnable.handleResult(JavaInstanceRunnable.java:391) ~[io.streamnative-pulsar-functions-instance-3.0.1.3.jar:?]
org.apache.pulsar.functions.instance.JavaInstanceRunnable.run(JavaInstanceRunnable.java:331) ~[io.streamnative-pulsar-functions-instance-3.0.1.3.jar:?]
java.lang.Thread.run(Thread.java:833) ~[?:?]
2023-12-27T10:01:14,991+0800 [pulsar-client-io-17-7] INFO org.apache.pulsar.client.impl.ProducerImpl - [persistent://pulsar/log_Production/ShiMenMission-partition-3] [sink-ShiMenMission] Closed Producer
```
I override the close interface

```java
@Override
public void close() throws Exception {
logger.info("start to close all the producers");
closeProducer();
closeClient();
}

public void closeProducer() {
List> closeFutures = new ArrayList<>(producerMap.size());
try {
for (Map.Entry> entry : producerMap.entrySet()) {
Producer producer = entry.getValue();
closeFutures.add(producer.closeAsync());
producerMap.clear();
}
org.apache.pulsar.common.util.FutureUtil.waitForAll(closeFutures);
logger.info("finish to close all the producers");
} catch (Exception e) {
logger.warn("Failed to close all the producers", e);
}
}

public void closeClient() {
List> closeFutures = new ArrayList<>(10);
try {
closeFutures.add(pulsarClient.closeAsync());
org.apache.pulsar.common.util.FutureUtil.waitForAll(closeFutures);
logger.info("finish to close all the client");
} catch (Exception e) {
logger.warn("Failed to close all the producers", e);
}
}
```

### Anything else?

_No response_

### Are you willing to submit a PR?

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

[error.log](https://github.com/apache/pulsar/files/13792035/error.log)

Start with the reported stack traces in PushPulsarSource.java, MultiConsumerPulsarSource.java, and JavaInstanceRunnable.java, then compare them with the shutdown sequence in the custom close methods. Reproduce the interruption while running an effectively-once function and determine how shutdown affects function reads and writes; done means the interruption behavior is understood and the stuck-processing failure is addressed.

Written by the indexing model from the issue text.