@Override
  public void checkTopicAccess(
      final KsqlSecurityContext securityContext,
      final String topicName,
      final AclOperation operation
  ) {
    final Set<AclOperation> authorizedOperations = securityContext.getServiceContext()
        .getTopicClient().describeTopic(topicName).authorizedOperations();

    // Kakfa 2.2 or lower do not support authorizedOperations(). In case of running on a
    // unsupported broker version, then the authorizeOperation will be null.
    if (authorizedOperations != null && !authorizedOperations.contains(operation)) {
      // This error message is similar to what Kafka throws when it cannot access the topic
      // due to an authorization error. I used this message to keep a consistent message.
      throw new KsqlTopicAuthorizationException(operation, Collections.singleton(topicName));
    }
  }
@Test
  public void shouldDenyIfAuthorizedOperationsDoesNotContainREAD() {
    // Given:
    givenTopicPermissions(TOPIC_1, Collections.singleton(AclOperation.WRITE));

    // When:
    final Exception e = assertThrows(
        KsqlTopicAuthorizationException.class,
        () -> accessValidator.checkTopicAccess(securityContext, TOPIC_NAME_1, AclOperation.READ)
    );

    // Then:
    assertThat(e.getMessage(), containsString(String.format(
        "Authorization denied to Read on topic(s): [%s]", TOPIC_1.name()
    )));
  }