Close the dispatcher when a KV or Object Store watch fails to start

[scottf]

Fixes #1642.

NatsWatchSubscription.finishInit creates a dispatcher and then subscribes. When the subscribe throws, for example because the bucket's stream is gone, the exception leaves the NatsKeyValueWatchSubscription or NatsObjectStoreWatchSubscription constructor, so the caller never gets an object to unsubscribe, and the dispatcher and its thread stay on the connection until the connection is closed. Every failed KeyValue.watch or ObjectStore.watch leaked one dispatcher.

The subscribe and the consumer-info read that follows it now run inside a try/catch. On any exception the dispatcher created for the watch is closed through NatsConnection.closeDispatcher, which also unsubscribes anything that was subscribed on it, an exception from the close is added as suppressed, and the original exception is rethrown. The same approach as the V3 JetStream subscribe cleanup.

Tests: KeyValueTests.testWatchThatFailsToStartReleasesItsDispatcher and ObjectStoreTests.testWatchThatFailsToStartReleasesItsDispatcher create a bucket, take the bucket context, delete the bucket, then call watch five times. Each call throws JetStreamApiException 10059 and the connection's dispatcher count stays where it was. Both fail on main (the count goes from 1 to 2 on the first call) and pass with the change.

[synadia-claude-reviewer]

Claude finished @scottf's task in 1m 0s —— View job

---

Reviewing PR #1643

• Gather context (PR diff, related code)
• Review NatsWatchSubscription.java changes
• Review test changes
• Post review summary

Review

The fix is correct and well-scoped. finishInit now wraps the subscribe + getConsumerInfo() call in a try/catch; on any IOException, JetStreamApiException, or RuntimeException it closes the dispatcher it just created via NatsConnection.closeDispatcher, which unsubscribes anything attached to it and stops its delivery thread. A failure during closeDispatcher (e.g. IllegalStateException if the connection is already closed) is added as a suppressed exception rather than swallowing the original error, and e is rethrown. Precise-rethrow (JLS 11.2.2) makes throw e; legal here without widening the method's declared throws IOException, JetStreamApiException, since RuntimeException is unchecked.

A few things worth noting (none blocking):

• dispatcher = null after the close mirrors the existing pattern in unsubscribe() (NatsWatchSubscription.java:106-114), so the field is left in a consistent state if anyone reused the instance — though since the exception escapes the constructor, the partially-built object is normally unreachable anyway.
• If createDispatcher() itself throws (e.g. connection closed/draining), it happens before the try, so there's nothing to clean up — correctly out of scope for this catch.
• Tests (KeyValueTests.testWatchThatFailsToStartReleasesItsDispatcher, ObjectStoreTests.testWatchThatFailsToStartReleasesItsDispatcher) directly assert the dispatcher count via the existing package-private NatsConnection.getDispatchers() accessor across 5 repeated failures, which is a solid regression guard for the leak.

No correctness, security, or thread-safety issues found.

[mtmk]

LGTM