private NioEchoServer startServerApiVersionsUnsupportedByClient(final SecurityProtocol securityProtocol, String saslMechanism) throws Exception {
        final ListenerName listenerName = ListenerName.forSecurityProtocol(securityProtocol);
        final Map<String, ?> configs = Collections.emptyMap();
        final JaasContext jaasContext = JaasContext.loadServerContext(listenerName, saslMechanism, configs);
        final Map<String, JaasContext> jaasContexts = Collections.singletonMap(saslMechanism, jaasContext);

        boolean isScram = ScramMechanism.isScram(saslMechanism);
        if (isScram)
            ScramCredentialUtils.createCache(credentialCache, Collections.singletonList(saslMechanism));

        Function<Short, ApiVersionsResponse> apiVersionSupplier = version -> {
            ApiVersionCollection versionCollection = new ApiVersionCollection(2);
            versionCollection.add(new ApiVersion().setApiKey(ApiKeys.SASL_HANDSHAKE.id).setMinVersion((short) 0).setMaxVersion((short) 100));
            versionCollection.add(new ApiVersion().setApiKey(ApiKeys.SASL_AUTHENTICATE.id).setMinVersion((short) 0).setMaxVersion((short) 100));
            return new ApiVersionsResponse(new ApiVersionsResponseData().setApiKeys(versionCollection));
        };

        SaslChannelBuilder serverChannelBuilder = new SaslChannelBuilder(ConnectionMode.SERVER, jaasContexts,
                securityProtocol, listenerName, false, saslMechanism, true,
                credentialCache, null, null, time, new LogContext(), apiVersionSupplier);

        serverChannelBuilder.configure(saslServerConfigs);
        server = new NioEchoServer(listenerName, securityProtocol, new TestSecurityConfig(saslServerConfigs),
                "localhost", serverChannelBuilder, credentialCache, time);
        server.start();
        return server;
    }
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