public static Read read() {
    return new AutoValue_RedisIO_Read.Builder()
        .setConnectionConfiguration(RedisConnectionConfiguration.create())
        .setKeyPattern("*")
        .setBatchSize(1000)
        .setOutputParallelization(true)
        .build();
  }
@Test
  public void testReadWithKeyPattern() {
    List<KV<String, String>> data = buildIncrementalData("pattern", 10);
    data.forEach(kv -> client.set(kv.getKey(), kv.getValue()));

    PCollection<KV<String, String>> read =
        p.apply("Read", RedisIO.read().withEndpoint(REDIS_HOST, port).withKeyPattern("pattern*"));
    PAssert.that(read).containsInAnyOrder(data);

    PCollection<KV<String, String>> readNotMatch =
        p.apply(
            "ReadNotMatch",
            RedisIO.read().withEndpoint(REDIS_HOST, port).withKeyPattern("foobar*"));
    PAssert.thatSingleton(readNotMatch.apply(Count.globally())).isEqualTo(0L);

    p.run();
  }