class KsqlAuthorizationValidatorFactory {

  public static Optional<KsqlAuthorizationValidator> create(
      final KsqlConfig ksqlConfig,
      final ServiceContext serviceContext
  ) {
    final String enabled = ksqlConfig.getString(KsqlConfig.KSQL_ENABLE_TOPIC_ACCESS_VALIDATOR);
    if (enabled.equals(KsqlConfig.KSQL_ACCESS_VALIDATOR_ON)) {
      LOG.info("Forcing topic access validator");
      return Optional.of(createAuthorizationValidator(ksqlConfig));
    } else if (enabled.equals(KsqlConfig.KSQL_ACCESS_VALIDATOR_OFF)) {
      return Optional.empty();
    }

    final Admin adminClient = serviceContext.getAdminClient();

    if (isKafkaAuthorizerEnabled(adminClient)) {
      if (KafkaClusterUtil.isAuthorizedOperationsSupported(adminClient)) {
        LOG.info("KSQL topic authorization checks enabled.");
        return Optional.of(createAuthorizationValidator(ksqlConfig));
      }

      LOG.warn("The Kafka broker has an authorization service enabled, but the Kafka "
          + "version does not support authorizedOperations(). "
          + "KSQL topic authorization checks will not be enabled.");
    }
    return Optional.empty();
  }

  private  KsqlAuthorizationValidatorFactory();

}