public static Optional<KsqlAuthorizationValidator> create(
      final KsqlConfig ksqlConfig,
      final ServiceContext serviceContext,
      final Optional<KsqlAuthorizationProvider> externalAuthorizationProvider
  ) {
    final Optional<KsqlAccessValidator> accessValidator = getAccessValidator(
        ksqlConfig,
        serviceContext,
        externalAuthorizationProvider
    );

    return accessValidator.map(v ->
        new KsqlAuthorizationValidatorImpl(cacheIfEnabled(ksqlConfig, v)));
  }
@Test
  public void shouldNotReturnProvidedAuthorizationValidatorIfAccessValidatorFlagIsOff() {
    // Given:
    when(ksqlConfig.getString(KsqlConfig.KSQL_ENABLE_ACCESS_VALIDATOR))
        .thenReturn(KsqlConfig.KSQL_ACCESS_VALIDATOR_OFF);

    // When:
    final Optional<KsqlAuthorizationValidator> validator = KsqlAuthorizationValidatorFactory.create(
        ksqlConfig,
        serviceContext,
        Optional.of(authorizationProvider)
    );

    // Then:
    assertThat(validator, is(Optional.empty()));
  }