@Test
public void shouldDeduplicateRowsWithinBatch() {

  TestStream<FeatureRow> featureRowTestStream =
      TestStream.create(ProtoCoder.of(FeatureRow.class))
          .addElements(
              createFeatureRow(
                  "myproject/feature_set",
                  Timestamp.newBuilder().setSeconds(20).build(),
                  field("entity_id_primary", 1, Enum.INT32),
                  field("entity_id_secondary", "a", Enum.STRING),
                  field("feature_2", 111, Enum.INT32)))
          .addElements(
              createFeatureRow(
                  "myproject/feature_set",
                  Timestamp.newBuilder().setSeconds(10).build(),
                  field("entity_id_primary", 1, Enum.INT32),
                  field("entity_id_secondary", "a", Enum.STRING),
                  field("feature_2", 222, Enum.INT32)))
          .addElements(
              createFeatureRow(
                  "myproject/feature_set",
                  Timestamp.getDefaultInstance(),
                  field("entity_id_primary", 1, Enum.INT32),
                  field("entity_id_secondary", "a", Enum.STRING),
                  field("feature_2", 333, Enum.INT32)))
          .advanceWatermarkToInfinity();
  p.apply(featureRowTestStream).apply(redisFeatureSink.writer());
  p.run();
  RedisKey expectedKey =
      createRedisKey(
          "myproject/feature_set",
          field("entity_id_primary", 1, Enum.INT32),
          field("entity_id_secondary", "a", Enum.STRING));
  FeatureRow expectedValue =
      createFeatureRow(
          "",
          Timestamp.newBuilder().setSeconds(20).build(),
          Field.newBuilder()
              .setName(hash("feature_1"))
              .setValue(Value.getDefaultInstance())
              .build(),
          field(hash("feature_2"), 111, Enum.INT32));
  byte[] actual = sync.get(expectedKey.toByteArray());
  assertThat(actual, equalTo(expectedValue.toByteArray()));
}