@Override
  public KsqlSecurityContext provide(final ApiSecurityContext apiSecurityContext) {

    final Optional<KsqlPrincipal> principal = apiSecurityContext.getPrincipal();
    final Optional<String> authHeader = apiSecurityContext.getAuthHeader();
    final List<Entry<String, String>> requestHeaders = apiSecurityContext.getRequestHeaders();

    // A user context is not necessary if a user context provider is not present or the user
    // principal is missing. If a failed authentication attempt results in a missing principle,
    // then the authentication plugin will have already failed the connection before calling
    // this method. Therefore, if we've reached this method with a missing principle, then this
    // must be a valid connection that does not require authentication.
    // For these cases, we create a default service context that the missing user can use.
    final boolean requiresUserContext =
        securityExtension != null
            && securityExtension.getUserContextProvider().isPresent()
            && principal.isPresent();

    if (!requiresUserContext) {
      return new KsqlSecurityContext(
          principal,
          defaultServiceContextFactory.create(
              ksqlConfig,
              authHeader,
              schemaRegistryClientFactory,
              connectClientFactory,
              sharedClient,
              requestHeaders,
              principal)
      );
    }

    return securityExtension.getUserContextProvider()
        .map(provider -> new KsqlSecurityContext(
            principal,
            userServiceContextFactory.create(
                ksqlConfig,
                authHeader,
                provider.getKafkaClientSupplier(principal.get()),
                provider.getSchemaRegistryClientFactory(principal.get()),
                connectClientFactory,
                sharedClient,
                requestHeaders,
                principal)))
        .get();
  }
@Test
  public void shouldCreateDefaultServiceContextIfUserContextProviderIsNotEnabled() {
    // Given:
    when(securityExtension.getUserContextProvider()).thenReturn(Optional.empty());

    // When:
    final KsqlSecurityContext ksqlSecurityContext =
        ksqlSecurityContextProvider.provide(apiSecurityContext);

    // Then:
    assertThat(ksqlSecurityContext.getUserPrincipal(), is(Optional.of(user1)));
    assertThat(ksqlSecurityContext.getServiceContext(), is(defaultServiceContext));
  }