Number of connection issue

The maximum number of connections using JmsIO.Read seems to grow in an uncontrolled way, causing new connections to be rejected by the broker if the number exceeds the configured quota on broker side.

The number of connections is related to the number of workers allocated by the runner, and the number of threads configured per worker, at least at the beginning.

Our assumption is:

JmsCheckPointMark is mutable (private Instant oldestMessageTimestamp – this field is updated for each message received). What we think it’s happening, with direct runner at least, the runner is unable to retrieve existing JMSIO readers (JmsCheckPointMark is used as a key - as part of  splitable pardo restriction object - for retrieving active JMSIO Readers, and because it’s mutated for each received message,  the runner will create a new reader for each new received message – that’s why the connections number spins out of control at least in DirectRunner). This is the reason that target parallelism parameter helps on controlling the ‘initial’ number of active connections, but later on we have issues.

Imported from Jira [BEAM-12523](https://issues.apache.org/jira/browse/BEAM-12523). Original Jira may contain additional context.
Reported by: rvballada.