@Override
    public void filter(ContainerRequestContext requestContext) {
        if (isInternalRequest(requestContext)) {
            log.trace("Skipping authentication for internal request");
            return;
        }

        try {
            log.debug("Authenticating request");
            BasicAuthCredentials credentials = new BasicAuthCredentials(requestContext.getHeaderString(AUTHORIZATION));
            LoginContext loginContext = new LoginContext(
                CONNECT_LOGIN_MODULE,
                null,
                new BasicAuthCallBackHandler(credentials),
                configuration);
            loginContext.login();
            setSecurityContextForRequest(requestContext, credentials);
        } catch (LoginException | ConfigException e) {
            // Log at debug here in order to avoid polluting log files whenever someone mistypes their credentials
            log.debug("Request failed authentication", e);
            requestContext.abortWith(
                Response.status(Response.Status.UNAUTHORIZED)
                    .entity("User cannot access the resource.")
                    .build());
        }
    }
@Test
    public void testEmptyCredentialsFile() throws IOException {
        File credentialFile = setupPropertyLoginFile(false);
        JaasBasicAuthFilter jaasBasicAuthFilter = setupJaasFilter("KafkaConnect", credentialFile.getPath());
        ContainerRequestContext requestContext = setMock("Basic", "user", "password");
        jaasBasicAuthFilter.filter(requestContext);

        verify(requestContext, atLeastOnce()).getMethod();
        verify(requestContext).getHeaderString(JaasBasicAuthFilter.AUTHORIZATION);
    }