@Test
public void runPipeline_ShouldWriteToRedisCorrectlyGivenValidSpecAndFeatureRow()
    throws IOException, InterruptedException {

  Source featureSource =
      Source.newBuilder()
          .setType(SourceType.KAFKA)
          .setKafkaSourceConfig(
              KafkaSourceConfig.newBuilder()
                  .setBootstrapServers(kafkaContainer.getBootstrapServers())
                  .setTopic(KAFKA_TOPIC)
                  .build())
          .build();
  IngestionJobProto.SpecsStreamingUpdateConfig specsStreamingUpdateConfig =
      IngestionJobProto.SpecsStreamingUpdateConfig.newBuilder()
          .setSource(
              KafkaSourceConfig.newBuilder()
                  .setBootstrapServers(kafkaContainer.getBootstrapServers())
                  .setTopic(KAFKA_SPECS_TOPIC)
                  .build())
          .setAck(
              KafkaSourceConfig.newBuilder()
                  .setBootstrapServers(kafkaContainer.getBootstrapServers())
                  .setTopic(KAFKA_SPECS_ACK_TOPIC)
                  .build())
          .build();
  FeatureSetSpec spec =
      FeatureSetSpec.newBuilder()
          .setName("feature_set")
          .setProject("myproject")
          .addEntities(
              EntitySpec.newBuilder()
                  .setName("entity_id_primary")
                  .setValueType(Enum.INT32)
                  .build())
          .addEntities(
              EntitySpec.newBuilder()
                  .setName("entity_id_secondary")
                  .setValueType(Enum.STRING)
                  .build())
          .addFeatures(
              FeatureSpec.newBuilder()
                  .setName("feature_1")
                  .setValueType(Enum.STRING_LIST)
                  .build())
          .addFeatures(
              FeatureSpec.newBuilder().setName("feature_2").setValueType(Enum.STRING).build())
          .addFeatures(
              FeatureSpec.newBuilder().setName("feature_3").setValueType(Enum.INT64).build())
          .setSource(featureSource)
          .build();
  FeatureSet featureSet = FeatureSet.newBuilder().setSpec(spec).build();
  Store redis =
      Store.newBuilder()
          .setName(StoreType.REDIS.toString())
          .setType(StoreType.REDIS)
          .setRedisConfig(
              RedisConfig.newBuilder()
                  .setHost(redisContainer.getHost())
                  .setPort(redisContainer.getFirstMappedPort())
                  .build())
          .addSubscriptions(
              Subscription.newBuilder()
                  .setProject(spec.getProject())
                  .setName(spec.getName())
                  .build())
          .build();
  ImportOptions options = PipelineOptionsFactory.create().as(ImportOptions.class);
  options.setSpecsStreamingUpdateConfigJson(
      JsonFormat.printer().print(specsStreamingUpdateConfig));
  options.setSourceJson(JsonFormat.printer().print(featureSource));
  options.setStoresJson(Collections.singletonList(JsonFormat.printer().print(redis)));
  options.setDefaultFeastProject("myproject");
  options.setProject("");
  options.setBlockOnRun(false);
  List<Pair<String, FeatureRow>> input = new ArrayList<>();
  Map<RedisKey, FeatureRow> expected = new HashMap<>();
  LOGGER.info("Generating test data ...");
  IntStream.range(0, IMPORT_JOB_SAMPLE_FEATURE_ROW_SIZE)
      .forEach(
          i -> {
            FeatureRow randomRow = TestUtil.createRandomFeatureRow(featureSet.getSpec());
            RedisKey redisKey = TestUtil.createRedisKey(featureSet.getSpec(), randomRow);
            input.add(Pair.of("", randomRow));
            List<FieldProto.Field> fields =
                randomRow.getFieldsList().stream()
                    .filter(
                        field ->
                            spec.getFeaturesList().stream()
                                .map(FeatureSpec::getName)
                                .collect(Collectors.toList())
                                .contains(field.getName()))
                    .map(
                        field ->
                            field.toBuilder().setName(TestUtil.hash(field.getName())).build())
                    .collect(Collectors.toList());
            randomRow =
                randomRow
                    .toBuilder()
                    .clearFields()
                    .addAllFields(fields)
                    .clearFeatureSet()
                    .build();
            expected.put(redisKey, randomRow);
          });
  LOGGER.info("Starting Import Job with the following options: {}", options.toString());
  PipelineResult pipelineResult = ImportJob.runPipeline(options);
  Thread.sleep(Duration.standardSeconds(IMPORT_JOB_READY_DURATION_SEC).getMillis());
  assertThat(pipelineResult.getState(), equalTo(State.RUNNING));
  LOGGER.info("Publishing {} Feature Row messages to Kafka ...", input.size());
  TestUtil.publishToKafka(
      kafkaContainer.getBootstrapServers(),
      KAFKA_SPECS_TOPIC,
      ImmutableList.of(Pair.of(getFeatureSetStringRef(spec), spec)),
      ByteArraySerializer.class,
      KAFKA_PUBLISH_TIMEOUT_SEC);
  TestUtil.publishToKafka(
      kafkaContainer.getBootstrapServers(),
      KAFKA_TOPIC,
      input,
      ByteArraySerializer.class,
      KAFKA_PUBLISH_TIMEOUT_SEC);
  TestUtil.waitUntilAllElementsAreWrittenToStore(
      pipelineResult,
      Duration.standardSeconds(IMPORT_JOB_MAX_RUN_DURATION_SEC),
      Duration.standardSeconds(IMPORT_JOB_CHECK_INTERVAL_DURATION_SEC));
  LOGGER.info("Validating the actual values written to Redis ...");
  RedisClient redisClient =
      RedisClient.create(
          new RedisURI(
              redisContainer.getHost(),
              redisContainer.getFirstMappedPort(),
              java.time.Duration.ofMillis(2000)));
  StatefulRedisConnection<byte[], byte[]> connection = redisClient.connect(new ByteArrayCodec());
  RedisCommands<byte[], byte[]> sync = connection.sync();
  for (Map.Entry<RedisKey, FeatureRow> entry : expected.entrySet()) {
    RedisKey key = entry.getKey();
    FeatureRow expectedValue = entry.getValue();
    // Ensure ingested key exists.
    byte[] actualByteValue = sync.get(key.toByteArray());
    assertThat("Key not found in Redis: " + key, actualByteValue, notNullValue());
    // Ensure value is a valid serialized FeatureRow object.
    FeatureRow actualValue = null;
    actualValue = FeatureRow.parseFrom(actualByteValue);
    // Ensure the retrieved FeatureRow is equal to the ingested FeatureRow.
    assertThat(actualValue, equalTo(expectedValue));
  }
  redisClient.shutdown();
}