@Override
  public void handle(final RoutingContext routingContext) {
    // We must set it to allow chunked encoding if we're using http1.1
    if (routingContext.request().version() == HttpVersion.HTTP_1_1) {
      routingContext.response().putHeader(TRANSFER_ENCODING, CHUNKED_ENCODING);
    } else if (routingContext.request().version() == HttpVersion.HTTP_2) {
      // Nothing required
    } else {
      routingContext.fail(BAD_REQUEST.code(),
          new KsqlApiException("This endpoint is only available when using HTTP1.1 or HTTP2",
              ERROR_CODE_BAD_REQUEST));
    }

    final CommonRequest request = getRequest(routingContext);
    if (request == null) {
      return;
    }

    final Optional<Boolean> internalRequest = ServerVerticle.isInternalRequest(routingContext);
    final MetricsCallbackHolder metricsCallbackHolder = new MetricsCallbackHolder();
    final long startTimeNanos = Time.SYSTEM.nanoseconds();

    endpoints.createQueryPublisher(
            request.sql, request.configOverrides, request.sessionProperties,
            request.requestProperties,
            context, server.getWorkerExecutor(),
            DefaultApiSecurityContext.create(routingContext, server), metricsCallbackHolder,
            internalRequest)
        .thenAccept(publisher -> {
          if (publisher instanceof BlockingPrintPublisher) {
            handlePrintPublisher(
                routingContext,
                (BlockingPrintPublisher) publisher);
          } else {
            handleQueryPublisher(
                routingContext,
                (QueryPublisher) publisher,
                metricsCallbackHolder,
                startTimeNanos);
          }
        })
        .exceptionally(t ->
            ServerUtils.handleEndpointException(t, routingContext, "Failed to execute query"));

  }
@Test
  public void shouldSucceed_scalablePushQuery() {
    // Given:
    when(publisher.isPullQuery()).thenReturn(false);
    when(publisher.isScalablePushQuery()).thenReturn(true);
    final QueryStreamArgs req = new QueryStreamArgs("select * from foo emit changes;",
        Collections.emptyMap(), Collections.emptyMap(), Collections.emptyMap());
    givenRequest(req);

    // When:
    handler.handle(routingContext);
    endHandler.getValue().handle(null);

    // Then:
    assertThat(subscriber.getValue(), notNullValue());
    verify(publisher).close();
  }