private void verifySslClientAuthForSaslSslListener(boolean useListenerPrefix,
                                                       SslClientAuth configuredClientAuth) throws Exception {

        SecurityProtocol securityProtocol = SecurityProtocol.SASL_SSL;
        configureMechanisms("PLAIN", Collections.singletonList("PLAIN"));
        String listenerPrefix = useListenerPrefix ? ListenerName.forSecurityProtocol(securityProtocol).configPrefix() : "";
        saslServerConfigs.put(listenerPrefix + BrokerSecurityConfigs.SSL_CLIENT_AUTH_CONFIG, configuredClientAuth.name());
        saslServerConfigs.put(BrokerSecurityConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG, SaslSslPrincipalBuilder.class.getName());
        server = createEchoServer(securityProtocol);

        SslClientAuth expectedClientAuth = useListenerPrefix ? configuredClientAuth : SslClientAuth.NONE;
        String certDn = "O=A client,CN=localhost";
        KafkaPrincipal principalWithMutualTls = SaslSslPrincipalBuilder.saslSslPrincipal(TestJaasConfig.USERNAME, certDn);
        KafkaPrincipal principalWithOneWayTls = SaslSslPrincipalBuilder.saslSslPrincipal(TestJaasConfig.USERNAME, "ANONYMOUS");

        // Client configured with valid key store
        createAndCheckClientConnectionAndPrincipal(securityProtocol, "0",
                expectedClientAuth == SslClientAuth.NONE ? principalWithOneWayTls : principalWithMutualTls);

        // Client does not configure key store
        removeClientSslKeystore();
        if (expectedClientAuth != SslClientAuth.REQUIRED) {
            createAndCheckClientConnectionAndPrincipal(securityProtocol, "1", principalWithOneWayTls);
        } else {
            createAndCheckSslAuthenticationFailure(securityProtocol, "1");
        }

        // Client configures untrusted key store
        CertStores newStore = new CertStores(false, "localhost");
        saslClientConfigs.putAll(newStore.keyStoreProps());
        if (expectedClientAuth == SslClientAuth.NONE) {
            createAndCheckClientConnectionAndPrincipal(securityProtocol, "2", principalWithOneWayTls);
        } else {
            createAndCheckSslAuthenticationFailure(securityProtocol, "2");
        }
    }
Tests OAUTHBEARER fails the connection when the client presents a token with
insufficient scope .
clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java