@Override
  public <R> HoodieData<HoodieRecord<R>> tagLocation(
      HoodieData<HoodieRecord<R>> records, HoodieEngineContext context,
      HoodieTable hoodieTable) {
    return HoodieJavaRDD.of(HoodieJavaRDD.getJavaRDD(records)
        .mapPartitionsWithIndex(locationTagFunction(hoodieTable.getMetaClient()), true));
  }
@Test
  public void testEnsureTagLocationUsesCommitTimeline() throws Exception {
    // Load to memory
    HoodieWriteConfig config = getConfigBuilder(100, false, false)
        .withRollbackUsingMarkers(false).build();
    SparkHoodieHBaseIndex index = new SparkHoodieHBaseIndex(config);
    try (SparkRDDWriteClient writeClient = getHoodieWriteClient(config)) {

      String commitTime1 = writeClient.startCommit();
      JavaRDD<HoodieRecord> writeRecords1 = generateAndCommitRecords(writeClient, 20, commitTime1);

      // rollback the commit - leaves a clean file in timeline.
      writeClient.rollback(commitTime1);

      // create a second commit with 20 records
      metaClient = HoodieTableMetaClient.reload(metaClient);
      generateAndCommitRecords(writeClient, 20);

      // Now tagLocation for the first set of rolledback records, hbaseIndex should tag them
      metaClient = HoodieTableMetaClient.reload(metaClient);
      HoodieTable hoodieTable = HoodieSparkTable.create(config, context, metaClient);
      JavaRDD<HoodieRecord> javaRDD1 = tagLocation(index, writeRecords1, hoodieTable);
      assert (javaRDD1.filter(HoodieRecord::isCurrentLocationKnown).collect().size() == 20);
    }
  }