@Test
public void shouldWriteToRedis() {

  HashMap<RedisKey, FeatureRow> kvs = new LinkedHashMap<>();
  kvs.put(
      createRedisKey("myproject/fs", field("entity", 1, Enum.INT64)),
      createFeatureRow(
          null, Timestamp.getDefaultInstance(), field(hash("feature"), "one", Enum.STRING)));
  kvs.put(
      createRedisKey("myproject/fs", field("entity", 2, Enum.INT64)),
      createFeatureRow(
          null, Timestamp.getDefaultInstance(), field(hash("feature"), "two", Enum.STRING)));
  List<FeatureRow> featureRows =
      ImmutableList.of(
          createFeatureRow(
              "myproject/fs",
              null,
              field("entity", 1, Enum.INT64),
              field("feature", "one", Enum.STRING)),
          createFeatureRow(
              "myproject/fs",
              null,
              field("entity", 2, Enum.INT64),
              field("feature", "two", Enum.STRING)));
  p.apply(Create.of(featureRows)).apply(redisFeatureSink.writer());
  p.run();
  kvs.forEach(
      (key, value) -> {
        byte[] actual = sync.get(key.toByteArray());
        assertThat(actual, equalTo(value.toByteArray()));
      });
}