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 shouldReturnEmptyValidatorIfKafkaBrokerVersionTooLowAndExceptionWrapped()
      throws InterruptedException, ExecutionException {
    // Given:
    givenSingleNode();
    givenAuthorizerClass("a-class");
    final KafkaFuture<Set<AclOperation>> authorized = mockAuthorizedOperationsFuture();
    final DescribeClusterResult result = mock(DescribeClusterResult.class);
    when(adminClient.describeCluster(any())).thenReturn(result);
    when(result.authorizedOperations()).thenReturn(authorized);
    when(authorized.get())
        .thenThrow(new ExecutionException(new UnsupportedVersionException("too old")));

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

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