static <T> void handleOldApiRequest(
      final Server server,
      final RoutingContext routingContext,
      final Class<T> requestClass,
      final Optional<MetricsCallbackHolder> metricsCallbackHolder,
      final BiFunction<T, ApiSecurityContext, CompletableFuture<EndpointResponse>> requestor) {
    final long startTimeNanos = Time.SYSTEM.nanoseconds();
    final T requestObject;
    if (requestClass != null) {
      final Optional<T> optRequestObject = ServerUtils
          .deserialiseObject(routingContext.getBody(), routingContext, requestClass);
      if (!optRequestObject.isPresent()) {
        return;
      }
      requestObject = optRequestObject.get();
      if (requestObject instanceof KsqlRequest) {
        final KsqlRequest request = (KsqlRequest) requestObject;
        ApiServerUtils.setMaskedSql(request);
      }
    } else {
      requestObject = null;
    }
    final CompletableFuture<EndpointResponse> completableFuture = requestor
        .apply(requestObject, DefaultApiSecurityContext.create(routingContext, server));
    completableFuture.thenAccept(endpointResponse -> {
      handleOldApiResponse(
          server, routingContext, endpointResponse, metricsCallbackHolder, startTimeNanos);
    }).exceptionally(t -> {
      if (t instanceof CompletionException) {
        t = t.getCause();
      }
      handleOldApiResponse(
          server, routingContext, mapException(t), metricsCallbackHolder, startTimeNanos);
      return null;
    });
  }
@Test
  public void shouldMaskKsqlRequestQuery() throws ExecutionException, InterruptedException {
    // Given
    final String query = "--this is a comment. \n"
        + "CREATE SOURCE CONNECTOR `test-connector` WITH ("
        + "    \"connector.class\" = 'PostgresSource', \n"
        + "    'connection.url' = 'jdbc:postgresql://localhost:5432/my.db',\n"
        + "    \"mode\"='bulk',\n"
        + "    \"topic.prefix\"='jdbc-',\n"
        + "    \"table.whitelist\"='users',\n"
        + "    \"key\"='username');";
    final KsqlRequest req = new KsqlRequest(query, ImmutableMap.of(), ImmutableMap.of(), 1L);
    when(routingContext.getBody()).thenReturn(ServerUtils.serializeObject(req));

    // when, Then
    final String expected = "CREATE SOURCE CONNECTOR `test-connector` WITH "
        + "(\"connector.class\"='PostgresSource', "
        + "'connection.url'='[string]', "
        + "\"mode\"='[string]', "
        + "\"topic.prefix\"='[string]', "
        + "\"table.whitelist\"='[string]', "
        + "\"key\"='[string]');";

    OldApiUtils.handleOldApiRequest(server, routingContext, KsqlRequest.class, Optional.empty(),
        (ksqlRequest, securityContext) -> {
      assertThat(ksqlRequest.getMaskedKsql(), is(expected));
      final VertxCompletableFuture<EndpointResponse> vcf = new VertxCompletableFuture<>();
      vcf.complete(EndpointResponse.failed(1));
      return vcf;
    });
  }