@Test
public void loadTest() {

  List<FeatureRow> rows =
      IntStream.range(0, 10000)
          .mapToObj(
              i ->
                  createFeatureRow(
                      "myproject/feature_set",
                      Timestamp.newBuilder().setSeconds(20).build(),
                      field("entity_id_primary", i, Enum.INT32),
                      field("entity_id_secondary", "a", Enum.STRING),
                      field("feature_1", "text", Enum.STRING),
                      field("feature_2", 111, Enum.INT32)))
          .collect(Collectors.toList());
  p.apply(Create.of(rows)).apply(redisFeatureSink.writer());
  p.run();
  List<byte[]> outcome =
      IntStream.range(0, 10000)
          .mapToObj(
              i ->
                  createRedisKey(
                          "myproject/feature_set",
                          field("entity_id_primary", i, Enum.INT32),
                          field("entity_id_secondary", "a", Enum.STRING))
                      .toByteArray())
          .map(sync::get)
          .collect(Collectors.toList());
  assertThat(outcome, hasSize(10000));
  assertThat("All rows were stored", outcome.stream().allMatch(Objects::nonNull));
}