@Override
  public void checkAuthorization(
      final KsqlSecurityContext securityContext,
      final MetaStore metaStore,
      final Statement statement
  ) {
    if (statement instanceof Query) {
      validateQuery(securityContext, metaStore, (Query)statement);
    } else if (statement instanceof InsertInto) {
      validateInsertInto(securityContext, metaStore, (InsertInto)statement);
    } else if (statement instanceof CreateAsSelect) {
      validateCreateAsSelect(securityContext, metaStore, (CreateAsSelect)statement);
    } else if (statement instanceof PrintTopic) {
      validatePrintTopic(securityContext, (PrintTopic)statement);
    } else if (statement instanceof CreateSource) {
      validateCreateSource(securityContext, (CreateSource)statement);
    }
  }
@Test
  public void shouldThrowWhenJoinSelectWithoutReadPermissionsDenied() {
    // Given:
    givenTopicAccessDenied(KAFKA_TOPIC, AclOperation.READ);
    final Statement statement = givenStatement(String.format(
        "SELECT * FROM %s A JOIN %s B ON A.F1 = B.F1;", KAFKA_STREAM_TOPIC, AVRO_STREAM_TOPIC)
    );

    // When:
    final Exception e = assertThrows(
        KsqlTopicAuthorizationException.class,
        () -> authorizationValidator.checkAuthorization(securityContext, metaStore, statement)
    );

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