@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 testSecurityContextSet() throws IOException, URISyntaxException {
        File credentialFile = setupPropertyLoginFile(true);
        JaasBasicAuthFilter jaasBasicAuthFilter = setupJaasFilter("KafkaConnect", credentialFile.getPath());
        ContainerRequestContext requestContext = setMock("Basic", "user1", "password1");

        when(requestContext.getUriInfo()).thenReturn(mock(UriInfo.class));
        when(requestContext.getUriInfo().getRequestUri()).thenReturn(new URI("https://foo.bar"));

        jaasBasicAuthFilter.filter(requestContext);

        ArgumentCaptor<SecurityContext> capturedContext = ArgumentCaptor.forClass(SecurityContext.class);
        verify(requestContext).setSecurityContext(capturedContext.capture());
        assertEquals("user1", capturedContext.getValue().getUserPrincipal().getName());
        assertTrue(capturedContext.getValue().isSecure());
    }