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 shouldReturn400OnBadStatement() {
    // Given:
    when(mockStatementParser.parseSingleStatement(any()))
        .thenThrow(new IllegalArgumentException("some error message"));

    // When:
    final KsqlRestException e = assertThrows(
        KsqlRestException.class,
        () -> testResource.streamQuery(
            securityContext,
            new KsqlRequest("query", Collections.emptyMap(), Collections.emptyMap(), null),
            new CompletableFuture<>(),
            Optional.empty(),
            new MetricsCallbackHolder(),
            context
        )
    );

    // Then:
    assertThat(e, exceptionStatusCode(is(BAD_REQUEST.code())));
    assertThat(e, exceptionErrorMessage(errorMessage(is("some error message"))));
    assertThat(e, exceptionErrorMessage(errorCode(is(ERROR_CODE_BAD_STATEMENT))));
  }