private NioEchoServer startServerWithoutSaslAuthenticateHeader(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 -> {
            ApiVersionsResponse defaultApiVersionResponse = TestUtils.defaultApiVersionsResponse(
                ApiMessageType.ListenerType.ZK_BROKER);
            ApiVersionCollection apiVersions = new ApiVersionCollection();
            for (ApiVersion apiVersion : defaultApiVersionResponse.data().apiKeys()) {
                if (apiVersion.apiKey() != ApiKeys.SASL_AUTHENTICATE.id) {
                    // ApiVersion can NOT be reused in second ApiVersionCollection
                    // due to the internal pointers it contains.
                    apiVersions.add(apiVersion.duplicate());
                }

            }
            ApiVersionsResponseData data = new ApiVersionsResponseData()
                .setErrorCode(Errors.NONE.code())
                .setThrottleTimeMs(0)
                .setApiKeys(apiVersions);
            return new ApiVersionsResponse(data);
        };

        SaslChannelBuilder serverChannelBuilder = new SaslChannelBuilder(ConnectionMode.SERVER, jaasContexts,
                securityProtocol, listenerName, false, saslMechanism, true,
                credentialCache, null, null, time, new LogContext(), apiVersionSupplier) {
            @Override
            protected SaslServerAuthenticator buildServerAuthenticator(Map<String, ?> configs,
                                                                       Map<String, AuthenticateCallbackHandler> callbackHandlers,
                                                                       String id,
                                                                       TransportLayer transportLayer,
                                                                       Map<String, Subject> subjects,
                                                                       Map<String, Long> connectionsMaxReauthMsByMechanism,
                                                                       ChannelMetadataRegistry metadataRegistry) {
                return new SaslServerAuthenticator(configs, callbackHandlers, id, subjects, null, listenerName,
                    securityProtocol, transportLayer, connectionsMaxReauthMsByMechanism, metadataRegistry, time, apiVersionSupplier) {
                    @Override
                    protected void enableKafkaSaslAuthenticateHeaders(boolean flag) {
                        // Don't enable Kafka SASL_AUTHENTICATE headers
                    }
                };
            }
        };
        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