public static WriteStreams writeStreams() {
    return new AutoValue_RedisIO_WriteStreams.Builder()
        .setConnectionConfiguration(RedisConnectionConfiguration.create())
        .setMaxLen(0L)
        .setApproximateTrim(true)
        .build();
  }
@Test
  public void testWriteStreams() {

    /* test data is 10 keys (stream IDs), each with two entries, each entry having one k/v pair of data */
    List<String> redisKeys =
        IntStream.range(0, 10).boxed().map(idx -> UUID.randomUUID().toString()).collect(toList());

    Map<String, String> fooValues = ImmutableMap.of("sensor-id", "1234", "temperature", "19.8");
    Map<String, String> barValues = ImmutableMap.of("sensor-id", "9999", "temperature", "18.2");

    List<KV<String, Map<String, String>>> allData =
        redisKeys.stream()
            .flatMap(id -> Stream.of(KV.of(id, fooValues), KV.of(id, barValues)))
            .collect(toList());

    PCollection<KV<String, Map<String, String>>> write =
        p.apply(
            Create.of(allData)
                .withCoder(
                    KvCoder.of(
                        StringUtf8Coder.of(),
                        MapCoder.of(StringUtf8Coder.of(), StringUtf8Coder.of()))));
    write.apply(RedisIO.writeStreams().withEndpoint(REDIS_HOST, port));
    p.run();

    for (String key : redisKeys) {
      List<StreamEntry> streamEntries =
          client.xrange(key, (StreamEntryID) null, (StreamEntryID) null, Integer.MAX_VALUE);
      assertEquals(2, streamEntries.size());
      assertThat(transform(streamEntries, StreamEntry::getFields), hasItems(fooValues, barValues));
    }
  }