[improve][broker] ServerCnx: go to Failed state when auth fails (apache#19312)

PIP: apache#12105

When authentication fails in the `ServerCnx`, the state is left in `Start` if the primary `authData` fails authentication and in `Connecting` or `Connected` if the `originalAuthData` authentication fails. To prevent any kind of unexpected behavior, we should go to `Failed` state.

Note that the tests verify the current behavior where a failed `originalAuthData` results first in a `Connected` command from the broker and then an `Error` command. I documented that I think this is sub optimal here apache#19311.

* Update `ServerCnx` state to `Failed` when there is an authentication exception during `handleConnect` and during `handleAuthResponse`.
* Update `handleAuthResponse` reply to `"Unable to authenticate"` instead of the `AuthenticationState` exception.

A new test is added. The added test covers the change made in apache#19295 where we updated `ServerCnx` so that we call `AuthState#authenticate` instead of relying on the implementation detail that the initialization calls `authenticate`. That PR should have added a test.

This is not a breaking change.

- [x] `doc-not-needed`

PR in forked repository: #18

(cherry picked from commit 8049690)

Author: michaeljmarshall

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java

@@ -966,7 +966,7 @@ protected void handleAuthResponse(CommandAuthResponse authResponse) {
             service.getPulsarStats().recordConnectionCreateFail();
             state = State.Failed;
             log.warn("[{}] Authentication failed: {} ", remoteAddress, e.getMessage());
-            ctx.writeAndFlush(Commands.newError(-1, ServerError.AuthenticationError, e.getMessage()));
+            ctx.writeAndFlush(Commands.newError(-1, ServerError.AuthenticationError, "Unable to authenticate"));
             close();
         } catch (Exception e) {
             service.getPulsarStats().recordConnectionCreateFail();

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java

@@ -728,7 +728,7 @@ public void testAuthResponseWithFailingAuthData() throws Exception {
 
         Object response3 = getResponse();
         assertTrue(response3 instanceof CommandError);
-        assertEquals(((CommandError) response3).getMessage(), "Do not pass");
+        assertEquals(((CommandError) response3).getMessage(), "Unable to authenticate");
         assertEquals(serverCnx.getState(), State.Failed);
         assertFalse(serverCnx.isActive());
         channel.finish();

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/utils/ClientChannelHelper.java

@@ -19,9 +19,8 @@
 package org.apache.pulsar.broker.service.utils;
 
 import java.util.Queue;
-
-import org.apache.pulsar.common.api.proto.CommandPartitionedTopicMetadataResponse;
 import org.apache.pulsar.common.api.proto.CommandAuthChallenge;
+import org.apache.pulsar.common.api.proto.CommandPartitionedTopicMetadataResponse;
 import org.apache.pulsar.common.protocol.PulsarDecoder;
 import org.apache.pulsar.common.api.proto.CommandAck;
 import org.apache.pulsar.common.api.proto.CommandCloseConsumer;