@Test
public void shouldWriteWithLatterTimestamp() {

  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(20).build(),
                  field("entity_id_primary", 2, Enum.INT32),
                  field("entity_id_secondary", "b", Enum.STRING),
                  field("feature_2", 222, Enum.INT32)))
          .addElements(
              createFeatureRow(
                  "myproject/feature_set",
                  Timestamp.newBuilder().setSeconds(10).build(),
                  field("entity_id_primary", 3, Enum.INT32),
                  field("entity_id_secondary", "c", Enum.STRING),
                  field("feature_2", 333, Enum.INT32)))
          .advanceWatermarkToInfinity();
  RedisKey keyA =
      createRedisKey(
          "myproject/feature_set",
          field("entity_id_primary", 1, Enum.INT32),
          field("entity_id_secondary", "a", Enum.STRING));
  RedisKey keyB =
      createRedisKey(
          "myproject/feature_set",
          field("entity_id_primary", 2, Enum.INT32),
          field("entity_id_secondary", "b", Enum.STRING));
  RedisKey keyC =
      createRedisKey(
          "myproject/feature_set",
          field("entity_id_primary", 3, Enum.INT32),
          field("entity_id_secondary", "c", Enum.STRING));
  sync.set(
      keyA.toByteArray(),
      createFeatureRow("", Timestamp.newBuilder().setSeconds(30).build()).toByteArray());
  sync.set(
      keyB.toByteArray(),
      createFeatureRow("", Timestamp.newBuilder().setSeconds(10).build()).toByteArray());
  sync.set(
      keyC.toByteArray(),
      createFeatureRow("", Timestamp.newBuilder().setSeconds(10).build()).toByteArray());
  p.apply(featureRowTestStream).apply(redisFeatureSink.writer());
  p.run();
  assertThat(
      sync.get(keyA.toByteArray()),
      equalTo(createFeatureRow("", Timestamp.newBuilder().setSeconds(30).build()).toByteArray()));
  assertThat(
      sync.get(keyB.toByteArray()),
      equalTo(
          createFeatureRow(
                  "",
                  Timestamp.newBuilder().setSeconds(20).build(),
                  Field.newBuilder()
                      .setName(hash("feature_1"))
                      .setValue(Value.getDefaultInstance())
                      .build(),
                  field(hash("feature_2"), 222, Enum.INT32))
              .toByteArray()));
  assertThat(
      sync.get(keyC.toByteArray()),
      equalTo(createFeatureRow("", Timestamp.newBuilder().setSeconds(10).build()).toByteArray()));
}