public EndpointResponse streamQuery(
      final KsqlSecurityContext securityContext,
      final KsqlRequest request,
      final CompletableFuture<Void> connectionClosedFuture,
      final Optional<Boolean> isInternalRequest,
      final MetricsCallbackHolder metricsCallbackHolder,
      final Context context
  ) {
    throwIfNotConfigured();
    activenessRegistrar.updateLastRequestTime();

    final PreparedStatement<?> statement = parseStatement(request);

    CommandStoreUtil.httpWaitForCommandSequenceNumber(
        commandQueue, request, commandQueueCatchupTimeout);

    return handleStatement(securityContext, request, statement, connectionClosedFuture,
        isInternalRequest, metricsCallbackHolder, context);
  }
@Test
  public void shouldReturnForbiddenKafkaAccessForPullQueryAuthorizationDenied() {
    // Given:
    when(mockStatementParser.<Query>parseSingleStatement(PULL_QUERY_STRING))
        .thenReturn(query);
    doThrow(
        new KsqlTopicAuthorizationException(AclOperation.READ, Collections.singleton(TOPIC_NAME)))
        .when(authorizationValidator).checkAuthorization(any(), any(), any());

    // When:
    final EndpointResponse response = testResource.streamQuery(
        securityContext,
        new KsqlRequest(PULL_QUERY_STRING, Collections.emptyMap(), Collections.emptyMap(), null),
        new CompletableFuture<>(),
        Optional.empty(),
        new MetricsCallbackHolder(),
        context
    );

    final KsqlErrorMessage responseEntity = (KsqlErrorMessage) response.getEntity();
    final KsqlErrorMessage expectedEntity = (KsqlErrorMessage) AUTHORIZATION_ERROR_RESPONSE.getEntity();
    assertEquals(response.getStatus(), AUTHORIZATION_ERROR_RESPONSE.getStatus());
    assertEquals(responseEntity.getMessage(), expectedEntity.getMessage());
  }