@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 testTagLocationAndPartitionPathUpdateEnabled() throws Exception {
    final String newCommitTime = "001";
    final String oldPartitionPath = "1970/01/01";
    final int numRecords = 10;

    List<HoodieRecord> newRecords = dataGen.generateInserts(newCommitTime, numRecords);
    List<HoodieRecord> oldRecords = new LinkedList();
    for (HoodieRecord newRecord: newRecords) {
      HoodieKey key = new HoodieKey(newRecord.getRecordKey(), oldPartitionPath);
      HoodieRecord hoodieRecord = new HoodieAvroRecord(key, (HoodieRecordPayload) newRecord.getData());
      oldRecords.add(hoodieRecord);
    }

    JavaRDD<HoodieRecord> newWriteRecords = jsc().parallelize(newRecords, 1);
    JavaRDD<HoodieRecord> oldWriteRecords = jsc().parallelize(oldRecords, 1);

    HoodieWriteConfig config = getConfigBuilder(100, true, false).build();
    SparkRDDWriteClient writeClient = getHoodieWriteClient(config);
    writeClient.startCommitWithTime(newCommitTime);
    JavaRDD<WriteStatus> writeStatues = writeClient.upsert(oldWriteRecords, newCommitTime);
    writeClient.commit(newCommitTime, writeStatues);
    assertNoWriteErrors(writeStatues.collect());
    metaClient = HoodieTableMetaClient.reload(metaClient);
    HoodieTable hoodieTable = HoodieSparkTable.create(config, context, metaClient);
    SparkHoodieHBaseIndex index = new SparkHoodieHBaseIndex(config);
    List<HoodieRecord> pathChangeRecords = tagLocation(index, newWriteRecords, hoodieTable).collect();
    assertEquals(numRecords * 2, pathChangeRecords.stream().count());
    assertEquals(numRecords, pathChangeRecords.stream().filter(HoodieRecord::isCurrentLocationKnown).count());

    String newCommitTime1 = "002";
    writeClient.startCommitWithTime(newCommitTime1);
    JavaRDD<WriteStatus> writeStatues1 = writeClient.upsert(newWriteRecords, newCommitTime1);
    writeClient.commit(newCommitTime1, writeStatues1);
    assertNoWriteErrors(writeStatues1.collect());
    assertEquals(numRecords * 2, writeStatues1.map(writeStatus -> writeStatus.getTotalRecords()).reduce(Long::sum));
    assertNotEquals(0, writeStatues1.filter(writeStatus -> writeStatus.getPartitionPath().equals(oldPartitionPath)).count());
    metaClient = HoodieTableMetaClient.reload(metaClient);
    hoodieTable = HoodieSparkTable.create(config, context, metaClient);
    List<HoodieRecord> pathChangeRecords1 = tagLocation(index, newWriteRecords, hoodieTable).collect();
    assertEquals(numRecords, pathChangeRecords1.stream().filter(HoodieRecord::isCurrentLocationKnown).count());
  }