[Java Client] Improve state changes to Connecting

This commit is a partial cherry pick from apache@6162ccf.
That commit relied on several other transaction commits not present in this branch.
This commit makes it easier to cherry-pick a fix to the Java Client Producer.

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java

@@ -103,12 +103,16 @@ protected void reconnectLater(Throwable exception) {
         long delayMs = backoff.next();
         log.warn("[{}] [{}] Could not get connection to broker: {} -- Will try again in {} s", state.topic, state.getHandlerName(),
                 exception.getMessage(), delayMs / 1000.0);
-        state.setState(State.Connecting);
-        state.client.timer().newTimeout(timeout -> {
-            log.info("[{}] [{}] Reconnecting after connection was closed", state.topic, state.getHandlerName());
-            incrementEpoch();
-            grabCnx();
-        }, delayMs, TimeUnit.MILLISECONDS);
+        if (state.changeToConnecting()) {
+            state.client.timer().newTimeout(timeout -> {
+                log.info("[{}] [{}] Reconnecting after connection was closed", state.topic, state.getHandlerName());
+                incrementEpoch();
+                grabCnx();
+            }, delayMs, TimeUnit.MILLISECONDS);
+        } else {
+            log.info("[{}] [{}] Ignoring reconnection request (state: {})",
+                    state.topic, state.getHandlerName(), state.getState());
+        }
     }
 
     protected long incrementEpoch() {

pulsar-client/src/main/java/org/apache/pulsar/client/impl/HandlerState.java

@@ -64,6 +64,13 @@ protected State getState() {
         return STATE_UPDATER.get(this);
     }
 
+    protected boolean changeToConnecting() {
+        return (STATE_UPDATER.compareAndSet(this, State.Uninitialized, State.Connecting)
+                || STATE_UPDATER.compareAndSet(this, State.Ready, State.Connecting)
+                || STATE_UPDATER.compareAndSet(this, State.RegisteringSchema, State.Connecting)
+                || STATE_UPDATER.compareAndSet(this, State.Connecting, State.Connecting));
+    }
+
     protected void setState(State s) {
         STATE_UPDATER.set(this, s);
     }