class ImportJob {

  @SuppressWarnings("UnusedReturnValue")
  public static PipelineResult runPipeline(ImportOptions options) throws IOException {
    /*
     * Steps:
     * 1. Read FeatureSetSpec messages from kafka
     * 2. Read messages from Feast Source as FeatureRow
     * 3. Validate the feature rows to ensure the schema matches what is registered to the system
     * 4. Distribute rows across stores by subscription
     * 5. Write FeatureRow to the corresponding Store
     * 6. Write elements that failed to be processed to a dead letter queue.
     * 7. Write metrics to a metrics sink
     * 8. Send ack on receiving FeatureSetSpec
     */

    PipelineOptionsValidator.validate(ImportOptions.class, options);
    Pipeline pipeline = Pipeline.create(options);

    log.info("Starting import job with settings: \n{}", options.toString());

    List<Store> stores = SpecUtil.parseStoreJsonList(options.getStoresJson());
    Source source = SpecUtil.parseSourceJson(options.getSourceJson());
    SpecsStreamingUpdateConfig specsStreamingUpdateConfig =
        SpecUtil.parseSpecsStreamingUpdateConfig(options.getSpecsStreamingUpdateConfigJson());

    // Step 1. Read FeatureSetSpecs from Spec source
    PCollection<KV<FeatureSetReference, FeatureSetSpec>> featureSetSpecs =
        pipeline.apply(
            "ReadFeatureSetSpecs",
            ReadFeatureSetSpecs.newBuilder()
                .setSource(source)
                .setStores(stores)
                .setSpecsStreamingUpdateConfig(specsStreamingUpdateConfig)
                .build());

    PCollectionView<Map<String, Iterable<FeatureSetSpec>>> globalSpecView =
        featureSetSpecs
            .apply(MapElements.via(new ReferenceToString()))
            .apply("GlobalSpecView", View.asMultimap());

    // Step 2. Read messages from Feast Source as FeatureRow.
    PCollectionTuple convertedFeatureRows =
        pipeline.apply(
            "ReadFeatureRowFromSource",
            ReadFromSource.newBuilder()
                .setSource(source)
                .setSuccessTag(FEATURE_ROW_OUT)
                .setFailureTag(DEADLETTER_OUT)
                .setKafkaConsumerProperties(
                    options.getKafkaConsumerProperties() == null
                        ? new HashMap<>()
                        : options.getKafkaConsumerProperties())
                .build());

    // Step 3. Process and validate incoming FeatureRows
    PCollectionTuple validatedRows =
        convertedFeatureRows
            .get(FEATURE_ROW_OUT)
            .apply(
                ProcessAndValidateFeatureRows.newBuilder()
                    .setDefaultProject(options.getDefaultFeastProject())
                    .setFeatureSetSpecs(globalSpecView)
                    .setSuccessTag(FEATURE_ROW_OUT)
                    .setFailureTag(DEADLETTER_OUT)
                    .build());

    Map<Store, TupleTag<FeatureRow>> storeTags =
        stores.stream()
            .map(s -> Pair.of(s, new TupleTag<FeatureRow>()))
            .collect(Collectors.toMap(Pair::getLeft, Pair::getRight));

    // Step 4. Allocate validated rows to stores by store subscription
    PCollectionTuple storeAllocatedRows =
        validatedRows
            .get(FEATURE_ROW_OUT)
            .apply(
                FeatureRowToStoreAllocator.newBuilder()
                    .setStores(stores)
                    .setStoreTags(storeTags)
                    .build());

    PCollectionList<FeatureSetReference> sinkReadiness = PCollectionList.empty(pipeline);

    for (Store store : stores) {
      FeatureSink featureSink = getFeatureSink(store);

      sinkReadiness = sinkReadiness.and(featureSink.prepareWrite(featureSetSpecs));
      PCollection<FeatureRow> rowsForStore =
          storeAllocatedRows.get(storeTags.get(store)).setCoder(ProtoCoder.of(FeatureRow.class));

      // Step 5. Write metrics of successfully validated rows
      rowsForStore.apply(
          "WriteInflightMetrics", WriteInflightMetricsTransform.create(store.getName()));

      // Step 6. Write FeatureRow to the corresponding Store.
      WriteResult writeFeatureRows =
          rowsForStore.apply("WriteFeatureRowToStore", featureSink.writer());

      // Step 7. Write FailedElements to a dead letter table in BigQuery.
      if (options.getDeadLetterTableSpec() != null) {
        // TODO: make deadletter destination type configurable
        DeadletterSink deadletterSink =
            new BigQueryDeadletterSink(options.getDeadLetterTableSpec());

        writeFeatureRows
            .getFailedInserts()
            .apply("WriteFailedElements_WriteFeatureRowToStore", deadletterSink.write());
      }

      // Step 8. Write metrics to a metrics sink.
      writeFeatureRows
          .getSuccessfulInserts()
          .apply("WriteSuccessMetrics", WriteSuccessMetricsTransform.create(store.getName()));

      writeFeatureRows
          .getFailedInserts()
          .apply("WriteFailureMetrics", WriteFailureMetricsTransform.create(store.getName()));
    }

    if (options.getDeadLetterTableSpec() != null) {
      DeadletterSink deadletterSink = new BigQueryDeadletterSink(options.getDeadLetterTableSpec());

      convertedFeatureRows
          .get(DEADLETTER_OUT)
          .apply("WriteFailedElements_ReadFromSource", deadletterSink.write());

      validatedRows
          .get(DEADLETTER_OUT)
          .apply("WriteFailedElements_ValidateRows", deadletterSink.write());
    }

    sinkReadiness
        .apply(Flatten.pCollections())
        .apply(
            "WriteAck",
            WriteFeatureSetSpecAck.newBuilder()
                .setSinksCount(stores.size())
                .setSpecsStreamingUpdateConfig(specsStreamingUpdateConfig)
                .build());

    return pipeline.run();
  }

}