diff --git a/pulsar-broker-auth-sasl/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderSasl.java b/pulsar-broker-auth-sasl/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderSasl.java
index 0743d8d20f846..01d4639e14d34 100644
--- a/pulsar-broker-auth-sasl/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderSasl.java
+++ b/pulsar-broker-auth-sasl/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderSasl.java
@@ -119,9 +119,8 @@ public AuthenticationState newAuthState(AuthData authData,
                                             SocketAddress remoteAddress,
                                             SSLSession sslSession) throws AuthenticationException {
         try {
-            return new SaslAuthenticationState(
-                new SaslAuthenticationDataSource(
-                    new PulsarSaslServer(jaasCredentialsContainer.getSubject(), allowedIdsPattern)));
+            PulsarSaslServer server = new PulsarSaslServer(jaasCredentialsContainer.getSubject(), allowedIdsPattern);
+            return new SaslAuthenticationState(server);
         } catch (Throwable t) {
             log.error("Failed create sasl auth state" , t);
             throw new AuthenticationException(t.getMessage());
diff --git a/pulsar-broker-auth-sasl/src/main/java/org/apache/pulsar/broker/authentication/SaslAuthenticationDataSource.java b/pulsar-broker-auth-sasl/src/main/java/org/apache/pulsar/broker/authentication/SaslAuthenticationDataSource.java
index 59b408a67bbf3..f023556bc2bac 100644
--- a/pulsar-broker-auth-sasl/src/main/java/org/apache/pulsar/broker/authentication/SaslAuthenticationDataSource.java
+++ b/pulsar-broker-auth-sasl/src/main/java/org/apache/pulsar/broker/authentication/SaslAuthenticationDataSource.java
@@ -18,17 +18,12 @@
  */
 package org.apache.pulsar.broker.authentication;
 
-import javax.naming.AuthenticationException;
-
 import lombok.extern.slf4j.Slf4j;
-import org.apache.pulsar.common.api.AuthData;
 
 @Slf4j
 public class SaslAuthenticationDataSource implements AuthenticationDataSource {
     private static final long serialVersionUID = 1L;
 
-    // server side token data, that will passed to sasl client side.
-    protected AuthData serverSideToken;
     private PulsarSaslServer pulsarSaslServer;
 
     public SaslAuthenticationDataSource(PulsarSaslServer saslServer) {
@@ -40,16 +35,6 @@ public boolean hasDataFromCommand() {
         return true;
     }
 
-    @Override
-    public AuthData authenticate(AuthData data) throws AuthenticationException {
-        serverSideToken = pulsarSaslServer.response(data);
-        return serverSideToken;
-    }
-
-    public boolean isComplete() {
-        return this.pulsarSaslServer.isComplete();
-    }
-
     public String getAuthorizationID() {
         return pulsarSaslServer.getAuthorizationID();
     }
diff --git a/pulsar-broker-auth-sasl/src/main/java/org/apache/pulsar/broker/authentication/SaslAuthenticationState.java b/pulsar-broker-auth-sasl/src/main/java/org/apache/pulsar/broker/authentication/SaslAuthenticationState.java
index f2dbb6ff2b659..6e852a450cb05 100644
--- a/pulsar-broker-auth-sasl/src/main/java/org/apache/pulsar/broker/authentication/SaslAuthenticationState.java
+++ b/pulsar-broker-auth-sasl/src/main/java/org/apache/pulsar/broker/authentication/SaslAuthenticationState.java
@@ -39,16 +39,17 @@ public class SaslAuthenticationState implements AuthenticationState {
     private final long stateId;
     private static final AtomicLong stateIdGenerator = new AtomicLong(0L);
     private final SaslAuthenticationDataSource authenticationDataSource;
+    private PulsarSaslServer pulsarSaslServer;
 
-    public SaslAuthenticationState(AuthenticationDataSource authenticationDataSource) {
+    public SaslAuthenticationState(PulsarSaslServer server) {
         stateId = stateIdGenerator.incrementAndGet();
-        checkArgument(authenticationDataSource instanceof SaslAuthenticationDataSource);
-        this.authenticationDataSource = (SaslAuthenticationDataSource)authenticationDataSource;
+        this.authenticationDataSource = new SaslAuthenticationDataSource(server);
+        this.pulsarSaslServer = server;
     }
 
     @Override
     public String getAuthRole() {
-        return authenticationDataSource.getAuthorizationID();
+        return pulsarSaslServer.getAuthorizationID();
     }
 
     @Override
@@ -58,7 +59,7 @@ public AuthenticationDataSource getAuthDataSource() {
 
     @Override
     public boolean isComplete() {
-        return authenticationDataSource.isComplete();
+        return pulsarSaslServer.isComplete();
     }
 
     /**
@@ -67,7 +68,7 @@ public boolean isComplete() {
      */
     @Override
     public AuthData authenticate(AuthData authData) throws AuthenticationException {
-        return authenticationDataSource.authenticate(authData);
+        return pulsarSaslServer.response(authData);
     }
 
     @Override