public KsqlTarget target(final URI server) {
    return target(server, Collections.emptyMap());
  }
@Test
  public void shouldPostQueryRequestStreamed() throws Exception {

    // Given:

    int numRows = 10;
    List<StreamedRow> expectedResponse = setQueryStreamResponse(numRows, false);
    String sql = "some sql";

    // When:
    KsqlTarget target = ksqlClient.target(serverUri);
    RestResponse<StreamPublisher<StreamedRow>> response = target
        .postQueryRequestStreamed(sql, Collections.emptyMap(), Optional.of(321L));

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

    assertThat(server.getPath(), is("/query"));
    assertThat(server.getHeaders().get("Accept"), is("application/json"));
    assertThat(getKsqlRequest(), is(new KsqlRequest(sql, properties, Collections.emptyMap(), 321L)));

    List<StreamedRow> rows = getElementsFromPublisher(numRows, response.getResponse());
    assertThat(rows, is(expectedResponse));
  }