[issue] In effectively_once mode, when a single pod fails, the entire function will fail due to failure to create the producer.

Start by reproducing effectively_once mode with five Kubernetes pods and fail one pod, then inspect producer-creation and failover logs. Also reproduce the PersistenceError case; done means subscription failover no longer causes all function pods to restart and the function does not remain stuck.

1. When I enable effectively_once and deploy the function in k8s, for example, after deploying 5 pods, when one of my pods crashes, his subscription will be transferred to other pods due to failover mode. At this time, other pods will fail to create the producer because the producer on the server side is not closed.

Failed to create producer: {"errorMsg":"org.apache.pulsar.broker.service.BrokerServiceException$NamingException: Producer with name 'persistent://pulsar/default/input_test-partition-5-0' is already connected to topic","reqId":1766584259806202457, "remote":"21.21.47.12/21.21.47.12:6650", "local":"/9.165.174.197:46786"}

After this, the function restarts due to an exception, and due to failover, the function once again transfers the subscription and fails due to failure to create the producer. Causes the all function pods to constantly restart
Therefore, when I need to enable effectively_once, I have to deploy multiple functions to consume partitioned topics separately. But this is not an easy way to maintain

2. In addition, the function can easily fall into the following error and be stuck because of this error until the broker is restarted.

WARN  org.apache.pulsar.client.impl.ClientCnx - [id: 0x24fe09d6, L:/9.165.182.50:36944 ! R:21.21.134.241/21.21.134.241:6650] Received send error from server: PersistenceError : Cannot determine whether the message is a duplicate at this time

If we have any optimization suggestions, I hope can provide them, thank very much