diff --git a/clients/src/main/java/org/apache/kafka/common/security/plain/internals/PlainServerCallbackHandler.java b/clients/src/main/java/org/apache/kafka/common/security/plain/internals/PlainServerCallbackHandler.java
index 7930fed033bd3..c8994689bdd54 100644
--- a/clients/src/main/java/org/apache/kafka/common/security/plain/internals/PlainServerCallbackHandler.java
+++ b/clients/src/main/java/org/apache/kafka/common/security/plain/internals/PlainServerCallbackHandler.java
@@ -36,6 +36,7 @@
 public class PlainServerCallbackHandler implements AuthenticateCallbackHandler {
 
     private static final String JAAS_USER_PREFIX = "user_";
+    private static final char[] UNKNOWN_USER_DUMMY_PASSWORD = {'\0'};
     private List<AppConfigurationEntry> jaasConfigEntries;
 
     @Override
@@ -65,7 +66,9 @@ protected boolean authenticate(String username, char[] password) throws IOExcept
             String expectedPassword = JaasContext.configEntryOption(jaasConfigEntries,
                     JAAS_USER_PREFIX + username,
                     PlainLoginModule.class.getName());
-            return expectedPassword != null && Utils.isEqualConstantTime(password, expectedPassword.toCharArray());
+            char[] comparedPassword = expectedPassword != null ? expectedPassword.toCharArray() : UNKNOWN_USER_DUMMY_PASSWORD;
+            boolean matches = Utils.isEqualConstantTime(password, comparedPassword);
+            return matches && expectedPassword != null;
         }
     }
 
diff --git a/clients/src/test/java/org/apache/kafka/common/security/plain/internals/PlainSaslServerTest.java b/clients/src/test/java/org/apache/kafka/common/security/plain/internals/PlainSaslServerTest.java
index 9d4249ad28b8d..9b16816bc905b 100644
--- a/clients/src/test/java/org/apache/kafka/common/security/plain/internals/PlainSaslServerTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/security/plain/internals/PlainSaslServerTest.java
@@ -20,16 +20,23 @@
 import org.apache.kafka.common.security.JaasContext;
 import org.apache.kafka.common.security.authenticator.TestJaasConfig;
 import org.apache.kafka.common.security.plain.PlainLoginModule;
+import org.apache.kafka.common.utils.Utils;
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
 
 import java.nio.charset.StandardCharsets;
 import java.util.HashMap;
 import java.util.Map;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.AdditionalMatchers.aryEq;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mockStatic;
 
 public class PlainSaslServerTest {
 
@@ -38,6 +45,7 @@ public class PlainSaslServerTest {
     private static final String USER_B = "userB";
     private static final String PASSWORD_B = "passwordB";
 
+    private PlainServerCallbackHandler callbackHandler;
     private PlainSaslServer saslServer;
 
     @BeforeEach
@@ -48,7 +56,7 @@ public void setUp() {
         options.put("user_" + USER_B, PASSWORD_B);
         jaasConfig.addEntry("jaasContext", PlainLoginModule.class.getName(), options);
         JaasContext jaasContext = new JaasContext("jaasContext", JaasContext.Type.SERVER, jaasConfig, null);
-        PlainServerCallbackHandler callbackHandler = new PlainServerCallbackHandler();
+        callbackHandler = new PlainServerCallbackHandler();
         callbackHandler.configure(null, "PLAIN", jaasContext.configurationEntries());
         saslServer = new PlainSaslServer(callbackHandler);
     }
@@ -70,6 +78,28 @@ public void authorizationIdNotEqualsAuthenticationId() {
         assertThrows(SaslAuthenticationException.class, () -> saslServer.evaluateResponse(saslMessage(USER_B, USER_A, PASSWORD_A)));
     }
 
+    @Test
+    public void unknownUserOrInvalidPassword() throws Exception {
+        Exception e = assertThrows(SaslAuthenticationException.class, () ->
+            saslServer.evaluateResponse(saslMessage("", "unknownUser", PASSWORD_A)));
+        assertEquals("Authentication failed: Invalid username or password", e.getMessage());
+
+        e = assertThrows(SaslAuthenticationException.class, () ->
+            saslServer.evaluateResponse(saslMessage("", USER_A, PASSWORD_B)));
+        assertEquals("Authentication failed: Invalid username or password", e.getMessage());
+
+        assertFalse(callbackHandler.authenticate("unknownUser", new char[] {'\0'}));
+    }
+
+    @Test
+    public void passwordIsComparedForUnknownUser() {
+        try (MockedStatic<Utils> mockedUtils = mockStatic(Utils.class, Mockito.CALLS_REAL_METHODS)) {
+            assertThrows(SaslAuthenticationException.class, () ->
+                saslServer.evaluateResponse(saslMessage("", "unknownUser", PASSWORD_A)));
+            mockedUtils.verify(() -> Utils.isEqualConstantTime(aryEq(PASSWORD_A.toCharArray()), any(char[].class)));
+        }
+    }
+
     @Test
     public void emptyTokens() {
         Exception e = assertThrows(SaslAuthenticationException.class, () ->