[fix][fn] Fix orphan exclusive producer on creation timeout in WorkerUtils.createExclusiveProducerWithRetry

[lhotari]

Fixes #25936

Motivation

WorkerUtils.createExclusiveProducerWithRetry bounded the exclusive producer creation with createAsync().get(10, TimeUnit.SECONDS) inside a retry loop. When get() threw TimeoutException, the catch block logged and retried, abandoning the in-flight CompletableFuture<Producer>. The underlying producer state machine kept running on the client IO thread, and when the creation eventually succeeded, the producer registered with the broker as an orphan: no caller held a reference to it, so it was never closed.

The orphan held the topic's exclusive producer access, so every subsequent retry was rejected — locking up leader election in the function worker, since the elected leader could not acquire the exclusive producer it needs on the internal topics and remained stuck retrying. The orphan was never reaped: the client never sends CloseProducer for it (no reference exists to invoke close()), the TCP connection stays alive because it is shared with other producers/consumers on the same PulsarClient, and the functions internal topics are created with deduplication disabled, so the broker's inactivity-based producer reaper does not apply.

In addition, the generic catch (Exception e) swallowed InterruptedException, so the retry loop kept running with the interrupt status cleared and abandoned the in-flight creation the same way. This is the WorkerUtils counterpart of the client-side interrupt fix in #24539.

Modifications

• Remove the 10-second timeout on the .get call: .get(10, TimeUnit.SECONDS) is effectively replaced with .get(), since the Pulsar client already bounds producer creation with the client's operationTimeout (applied via the lookup timeout, which defaults to operationTimeout) and completes the future exceptionally instead of abandoning it. This eliminates the TimeoutException path that orphaned in-flight producer creations.
• Wait for the creation with FutureUtil.getAndCleanupOnInterrupt(producerFuture, Producer::closeAsync) (the helper introduced in [fix][client] Close orphan producer or consumer when the creation is interrupted #24539) so that an interrupt registers a cleanup that closes the producer if the pending creation completes later.
• Handle InterruptedException explicitly: restore the interrupt status and propagate it, aborting the retry loop instead of retrying.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• Added WorkerUtilsTest.testCreateExclusiveProducerWithRetryClosesProducerOnInterrupt verifying that an interrupt aborts the retry loop instead of retrying, restores the interrupt status, and closes the producer once the pending creation completes after the interrupt.

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment