static <RequestT, ResponseT>
      PTransform<PCollection<KV<RequestT, ResponseT>>, Result<KV<RequestT, ResponseT>>>
          writeUsingRedis(
              Duration expiry,
              RedisClient client,
              Coder<RequestT> requestTCoder,
              Coder<@Nullable ResponseT> responseTCoder)
              throws NonDeterministicException {
    return write(
        new UsingRedis<>(requestTCoder, responseTCoder, client).write(expiry),
        KvCoder.of(requestTCoder, responseTCoder));
  }
@Test
  public void givenNonDeterministicCoder_writeUsingRedis_throwsError()
      throws Coder.NonDeterministicException {
    URI uri = URI.create("redis://localhost:6379");
    Duration expiry = Duration.standardSeconds(1L);
    assertThrows(
        NonDeterministicException.class,
        () ->
            Cache.writeUsingRedis(
                expiry,
                new RedisClient(uri),
                CallTest.NON_DETERMINISTIC_REQUEST_CODER,
                CallTest.DETERMINISTIC_RESPONSE_CODER));

    assertThrows(
        NonDeterministicException.class,
        () ->
            Cache.writeUsingRedis(
                expiry,
                new RedisClient(uri),
                CallTest.DETERMINISTIC_REQUEST_CODER,
                CallTest.NON_DETERMINISTIC_RESPONSE_CODER));

    Cache.writeUsingRedis(
        expiry,
        new RedisClient(uri),
        CallTest.DETERMINISTIC_REQUEST_CODER,
        CallTest.DETERMINISTIC_RESPONSE_CODER);
  }