java: the listener retry leaks a started JMS connection on every reconciliation tick when the refusal comes after createConnection (#7220)

[delchev]

Summary

#7220 (fixes #7217) retries a client-Java listener subscription the broker refused at startup, every DIRIGIBLE_JAVA_RECONCILE_INTERVAL_SECONDS (30 s). The attach it retries leaks the JMS connection it has already STARTED whenever the refusal comes after createConnection - and now that the attempt repeats forever, so does the leak.

Evidence (origin/master, engine-java/.../listener/ListenerClassConsumer.java:385-425)

Connection connection = connectionFactory.createConnection(ex -> ..., subscriptionId);   // start()ed inside the factory
Session session = connectionFactory.createSession(connection);
Destination destination = topic ? session.createTopic(destinationName) : session.createQueue(destinationName);
connectionFactory.configureRedeliveryPolicy(connection, destination);
MessageConsumer consumer = topic ? session.createDurableSubscriber((Topic) destination, subscriptionId) : session.createConsumer(destination);
consumer.setMessageListener(...);
...
} catch (JMSException | RuntimeException e) {
    reportFailure(label, scope, e);
    return null;                 // `connection` is never closed
}

ActiveMQConnectionArtifactsFactory.createConnection (engine-listeners/.../config/ActiveMQConnectionArtifactsFactory.java:89-104) calls connection.start() before returning, so by the time createSession / createTopic / configureRedeliveryPolicy / createDurableSubscriber / setMessageListener can throw, a transport socket and its ActiveMQ transport thread exist. The catch reports the failure and returns null without closing them.

Before #7220 this ran once per rebuild, so the cost was one orphaned connection per refused subscription. Now JavaConsumersReconciler re-runs subscribe every 30 s for as long as the refusal lasts - and the realistic refusals are PERMANENT ones that happen exactly at this point: an external broker (DIRIGIBLE_MESSAGING_BROKER_URL) whose destination authorization refuses createDurableSubscriber / createConsumer with a JMSSecurityException, or a destination name the broker's policy rejects. One started connection per subscription per tenant per tick, at DEBUG after the first WARN, until the file-descriptor or thread limit is hit - an outage the retry itself manufactures.

None of ListenerClassConsumerRetryTest's cases makes anything AFTER createConnection throw (every refusal there is createConnection itself), so the path is untested.

Fix

Close the connection on any failure after it was created: hold it in a local declared before the try, and in the catch (or a finally guarded by "not returned") call connection.close(), logging a close failure at DEBUG with the throwable. Add a retry-test case where createSession throws and assert the connection's close() was called and nothing is registered.

Found reviewing #7220.