public RestResponse<KsqlEntityList> postKsqlRequest(
      final String ksql,
      final Map<String, ?> requestProperties,
      final Optional<Long> previousCommandSeqNum
  ) {
    return post(
        KSQL_PATH,
        createKsqlRequest(ksql, requestProperties, previousCommandSeqNum),
        r -> deserialize(r.getBody(), KsqlEntityList.class)
    );
  }
@Test
  public void shouldSendAdditionalHeadersWithKsqlRequest() {
    // Given:
    final Map<String, String> additionalHeaders = ImmutableMap.of("h1", "v1", "h2", "v2");
    ksqlTarget = new KsqlTarget(httpClient, socketAddress, localProperties, authHeader, HOST,
        additionalHeaders, RequestOptions.DEFAULT_TIMEOUT);

    // When:
    executor.submit(() -> {
      try {
        ksqlTarget.postKsqlRequest("some ksql;", Collections.emptyMap(), Optional.empty());
      } catch (Exception e) {
        // ignore response error since this test is just testing headers on the outgoing request
      }
    });
    assertThatEventually(requestStarted::get, is(true));
    handlerCaptor.getValue().handle(Buffer.buffer());

    // Then:
    verify(httpClientRequest).putHeader("h1", "v1");
    verify(httpClientRequest).putHeader("h2", "v2");
  }