@Override
  public RestResponse<KsqlEntityList> makeKsqlRequest(
      final URI serverEndpoint,
      final String sql,
      final Map<String, ?> requestProperties) {
    final KsqlRequest request = new KsqlRequest(
        sql, Collections.emptyMap(), requestProperties, null);

    final EndpointResponse response = ksqlResource.handleKsqlStatements(securityContext, request);

    final int status = response.getStatus();

    if (status == OK.code()) {
      return RestResponse.successful(status, (KsqlEntityList) response.getEntity());
    } else {
      return RestResponse.erroneous(status, (KsqlErrorMessage) response.getEntity());
    }
  }
@Test
  public void shouldMakeKsqlRequest() {
    // Given:
    when(ksqlResource.handleKsqlStatements(securityContext, EXPECTED_REQUEST)).thenReturn(response);

    // When:
    final RestResponse<KsqlEntityList> restResponse =
        ksqlClient.makeKsqlRequest(unused, KSQL_STATEMENT, ImmutableMap.of());

    // Then:
    assertThat("is successful", restResponse.isSuccessful());
    assertThat(restResponse.getResponse(), sameInstance(entities));
  }