static Void handleEndpointException(
      final Throwable t, final RoutingContext routingContext, final String logMsg) {
    if (t instanceof CompletionException) {
      final Throwable actual = t.getCause();
      log.error(logMsg, actual);
      if (actual instanceof KsqlStatementException) {
        routingContext.fail(
            BAD_REQUEST.code(),
            new KsqlApiException(
                ((KsqlStatementException) actual).getUnloggedMessage(),
                ERROR_CODE_BAD_STATEMENT,
                ((KsqlStatementException) actual).getSqlStatement()
            )
        );
        return null;
      } else if (actual instanceof KsqlRateLimitException) {
        routingContext.fail(TOO_MANY_REQUESTS.code(),
            new KsqlApiException(actual.getMessage(), ERROR_CODE_TOO_MANY_REQUESTS));
        return null;
      } else if (actual instanceof KsqlApiException) {
        routingContext.fail(BAD_REQUEST.code(), actual);
        return null;
      }
    } else {
      log.error(logMsg, t);
    }
    // We don't expose internal error message via public API
    routingContext.fail(INTERNAL_SERVER_ERROR.code(), new KsqlApiException(
        "The server encountered an internal error when processing the query."
            + " Please consult the server logs for more information.",
        ERROR_CODE_SERVER_ERROR));
    return null;
  }
@Test
  public void testHandleServerRateLimitException() {
    String ratelimitErrorMsg = "Host is at rate limit for pull queries. Currently set to 1 QPS";
    CompletionException ratelimitException = new CompletionException(
        new KsqlRateLimitException(ratelimitErrorMsg));
    ServerUtils.handleEndpointException(ratelimitException, routingContext, "test msg");
    verify(routingContext).fail(eq(TOO_MANY_REQUESTS.code()), argThat(x -> {
      assertThat(x.getMessage(), equalTo(ratelimitErrorMsg));
      assertThat(((KsqlApiException) x).getErrorCode(), equalTo(ERROR_CODE_TOO_MANY_REQUESTS));
      return true;
    }));
  }