@Override
  public RestResponse<List<StreamedRow>> makeQueryRequest(
      final URI serverEndPoint,
      final String sql,
      final Map<String, ?> configOverrides,
      final Map<String, ?> requestProperties
  ) {

    final KsqlTarget target = sharedClient
        .target(serverEndPoint)
        .properties(configOverrides)
        .timeout(getQueryTimeout(configOverrides));

    final RestResponse<List<StreamedRow>> resp = getTarget(target)
        .postQueryRequest(sql, requestProperties, Optional.empty());

    if (resp.isErroneous()) {
      return RestResponse.erroneous(resp.getStatusCode(), resp.getErrorMessage());
    }

    return RestResponse.successful(resp.getStatusCode(), resp.getResponse());
  }
@Test
  public void shouldSetQueryTimeout() {
    // Given:
    when(ksqlConfig.getLong(KsqlConfig.KSQL_QUERY_PULL_FORWARDING_TIMEOUT_MS_CONFIG))
        .thenReturn(300L);

    // When:
    final RestResponse<List<StreamedRow>> result = client.makeQueryRequest(SERVER_ENDPOINT, "Sql",
        ImmutableMap.of(), ImmutableMap.of());

    // Then:
    verify(target).postQueryRequest("Sql", ImmutableMap.of(), Optional.empty());
    verify(target).timeout(300L);
    assertThat(result.getStatusCode(), is(queryResponse.getStatusCode()));
  }