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 shouldThrowOnDenyListedStreamProperty() {
    // Given:
    when(mockStatementParser.<Query>parseSingleStatement(PULL_QUERY_STRING)).thenReturn(query);
    testResource = new StreamedQueryResource(
        mockKsqlEngine,
        ksqlRestConfig,
        mockStatementParser,
        commandQueue,
        DISCONNECT_CHECK_INTERVAL,
        COMMAND_QUEUE_CATCHUP_TIMOEUT,
        activenessRegistrar,
        Optional.of(authorizationValidator),
        errorsHandler,
        denyListPropertyValidator,
        queryExecutor
      );
    final Map<String, Object> props = new HashMap<>(ImmutableMap.of(
        StreamsConfig.APPLICATION_SERVER_CONFIG, "something:1"
    ));
    props.put(KsqlConfig.KSQL_PROPERTIES_OVERRIDES_DENYLIST,
        StreamsConfig.NUM_STREAM_THREADS_CONFIG);
    when(mockKsqlEngine.getKsqlConfig()).thenReturn(new KsqlConfig(props));
    final Map<String, Object> overrides =
        ImmutableMap.of(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 1);
    doThrow(new KsqlException("deny override")).when(denyListPropertyValidator)
        .validateAll(overrides);
    when(errorsHandler.generateResponse(any(), any()))
        .thenReturn(badRequest("A property override was set locally for a property that the "
            + "server prohibits overrides for: 'num.stream.threads'"));

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

    // Then:
    verify(denyListPropertyValidator).validateAll(overrides);
    assertThat(response.getStatus(), CoreMatchers.is(BAD_REQUEST.code()));
    assertThat(((KsqlErrorMessage) response.getEntity()).getMessage(),
        is("A property override was set locally for a property that the server prohibits "
            + "overrides for: '" + StreamsConfig.NUM_STREAM_THREADS_CONFIG + "'"));
  }