[fix][broker] Avoid IllegalStateException while client_version is not…

… set (#16788)

(cherry picked from commit 4c6989c)

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

@@ -19,6 +19,7 @@
 package org.apache.pulsar.broker.service;
 
 import static com.google.common.base.Preconditions.checkArgument;
+import static org.apache.commons.lang3.StringUtils.EMPTY;
 import static org.apache.commons.lang3.StringUtils.isNotBlank;
 import static org.apache.pulsar.broker.admin.impl.PersistentTopicsBase.unsafeGetPartitionedTopicMetadataAsync;
 import static org.apache.pulsar.broker.lookup.TopicLookupBase.lookupTopicAsync;
@@ -905,7 +906,8 @@ protected void handleAuthResponse(CommandAuthResponse authResponse) {
 
         try {
             AuthData clientData = AuthData.of(authResponse.getResponse().getAuthData());
-            doAuthentication(clientData, authResponse.getProtocolVersion(), authResponse.getClientVersion());
+            doAuthentication(clientData, authResponse.getProtocolVersion(),
+                    authResponse.hasClientVersion() ? authResponse.getClientVersion() : EMPTY);
         } catch (AuthenticationException e) {
             service.getPulsarStats().recordConnectionCreateFail();
             log.warn("[{}] Authentication failed: {} ", remoteAddress, e.getMessage());

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

@@ -23,11 +23,15 @@
 import static org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest.createMockZooKeeper;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.CALLS_REAL_METHODS;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.matches;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertFalse;
 import static org.testng.Assert.assertNotNull;
@@ -89,6 +93,7 @@
 import org.apache.pulsar.common.api.proto.BaseCommand;
 import org.apache.pulsar.common.api.proto.BaseCommand.Type;
 import org.apache.pulsar.common.api.proto.CommandAck.AckType;
+import org.apache.pulsar.common.api.proto.CommandAuthResponse;
 import org.apache.pulsar.common.api.proto.CommandConnected;
 import org.apache.pulsar.common.api.proto.CommandError;
 import org.apache.pulsar.common.api.proto.CommandLookupTopicResponse;
@@ -1996,4 +2001,22 @@ public boolean isCompletedExceptionally(){
             channel.finish();
         }
     }
+
+    @Test
+    public void testHandleAuthResponseWithoutClientVersion() {
+        ServerCnx cnx = mock(ServerCnx.class, CALLS_REAL_METHODS);
+        CommandAuthResponse authResponse = mock(CommandAuthResponse.class);
+        org.apache.pulsar.common.api.proto.AuthData authData = mock(org.apache.pulsar.common.api.proto.AuthData.class);
+        when(authResponse.getResponse()).thenReturn(authData);
+        when(authResponse.hasResponse()).thenReturn(true);
+        when(authResponse.getResponse().hasAuthMethodName()).thenReturn(true);
+        when(authResponse.getResponse().hasAuthData()).thenReturn(true);
+        when(authResponse.hasClientVersion()).thenReturn(false);
+        try {
+            cnx.handleAuthResponse(authResponse);
+        } catch (Exception ignore) {
+        }
+        verify(authResponse, times(1)).hasClientVersion();
+        verify(authResponse, times(0)).getClientVersion();
+    }
 }