static <RequestT, ResponseT> Call<RequestT, ResponseT> of(
      Caller<RequestT, ResponseT> caller, Coder<ResponseT> responseTCoder) {
    caller = SerializableUtils.ensureSerializable(caller);
    return new Call<>(
        Configuration.<RequestT, ResponseT>builder()
            .setCaller(caller)
            .setResponseCoder(responseTCoder)
            .build());
  }
@Test
  public void givenCallerThrowsUserCodeExecutionException_emitsIntoFailurePCollection() {
    Result<Response> result =
        pipeline
            .apply(Create.of(new Request("a")))
            .apply(
                Call.of(
                    new CallerThrowsUserCodeExecutionException(),
                    NON_DETERMINISTIC_RESPONSE_CODER));

    PCollection<ApiIOError> failures = result.getFailures();
    PAssert.thatSingleton(countStackTracesOf(failures, UserCodeExecutionException.class))
        .isEqualTo(1L);
    PAssert.thatSingleton(countStackTracesOf(failures, UserCodeQuotaException.class)).isEqualTo(0L);
    PAssert.thatSingleton(countStackTracesOf(failures, UserCodeTimeoutException.class))
        .isEqualTo(0L);

    pipeline.run();
  }