public KsqlTarget targetHttp2(final URI server) {
    final boolean isUriTls = server.getScheme().equalsIgnoreCase("https");
    final HttpClient client = (isUriTls ? httpTlsClientHttp2 : httpNonTlsClientHttp2).orElseThrow(
        () -> new IllegalStateException("Must provide http2 options to use targetHttp2"));
    return new KsqlTarget(client,
        socketAddressFactory.apply(server.getPort(), server.getHost()), localProperties,
        basicAuthHeader, server.getHost(), Collections.emptyMap(), RequestOptions.DEFAULT_TIMEOUT);
  }
@Test
  public void shouldPostQueryRequestStreamedHttp2() throws Exception {

    // Given:

    int numRows = 10;
    List<List<?>> expectedResponse = setQueryStreamDelimitedResponse(numRows);
    String sql = "whateva";

    // When:
    KsqlTarget target = ksqlClient.targetHttp2(serverUri);
    CompletableFuture<RestResponse<StreamPublisher<StreamedRow>>> future = target
        .postQueryRequestStreamedAsync(sql, ImmutableMap.of());
    RestResponse<StreamPublisher<StreamedRow>> response = future.get();

    // Then:
    assertThat(server.getHttpMethod(), is(HttpMethod.POST));

    assertThat(server.getPath(), is("/query-stream"));
    assertThat(server.getHeaders().get("Accept"), is("application/vnd.ksqlapi.delimited.v1"));
    assertThat(getQueryStreamArgs(), is(new QueryStreamArgs(sql, properties,
        Collections.emptyMap(), Collections.emptyMap())));

    List<StreamedRow> rows = getElementsFromPublisher(numRows + 1, response.getResponse());
    int i = 0;
    for (StreamedRow row : rows) {
      assertThat(row.getRow().isPresent(), is(true));
      assertThat(row.getRow().get().getColumns(), is(expectedResponse.get(i++)));
    }
  }