public static void persistTaskMetadata(HelixPropertyStore<ZNRecord> propertyStore, String taskType,
      BaseTaskMetadata taskMetadata, int expectedVersion) {
    String newPath =
        ZKMetadataProvider.constructPropertyStorePathForMinionTaskMetadata(taskMetadata.getTableNameWithType(),
            taskType);
    String oldPath = ZKMetadataProvider.constructPropertyStorePathForMinionTaskMetadataDeprecated(taskType,
        taskMetadata.getTableNameWithType());
    if (propertyStore.exists(newPath, AccessOption.PERSISTENT) || !propertyStore.exists(oldPath,
        AccessOption.PERSISTENT)) {
      persistTaskMetadata(newPath, propertyStore, taskType, taskMetadata, expectedVersion);
    } else {
      persistTaskMetadata(oldPath, propertyStore, taskType, taskMetadata, expectedVersion);
    }
  }
@Test
  public void testPersistTaskMetadata() {
    DummyTaskMetadata taskMetadata = new DummyTaskMetadata(TABLE_NAME_WITH_TYPE, 2000);

    // the metadata will be written to the new path
    HelixPropertyStore<ZNRecord> propertyStore = new FakePropertyStore();
    MinionTaskMetadataUtils.persistTaskMetadata(propertyStore, TASK_TYPE, taskMetadata, EXPECTED_VERSION);
    assertTrue(propertyStore.exists(NEW_MINION_METADATA_PATH, ACCESS_OPTION));
    assertFalse(propertyStore.exists(OLD_MINION_METADATA_PATH, ACCESS_OPTION));
    assertEquals(MinionTaskMetadataUtils.fetchTaskMetadata(propertyStore, TASK_TYPE, TABLE_NAME_WITH_TYPE),
        taskMetadata.toZNRecord());

    // the metadata will be written to the old path if only the old path exists
    propertyStore = new FakePropertyStore();
    propertyStore.set(OLD_MINION_METADATA_PATH, OLD_TASK_METADATA.toZNRecord(), EXPECTED_VERSION, ACCESS_OPTION);
    MinionTaskMetadataUtils.persistTaskMetadata(propertyStore, TASK_TYPE, taskMetadata, EXPECTED_VERSION);
    assertFalse(propertyStore.exists(NEW_MINION_METADATA_PATH, ACCESS_OPTION));
    assertTrue(propertyStore.exists(OLD_MINION_METADATA_PATH, ACCESS_OPTION));
    assertEquals(MinionTaskMetadataUtils.fetchTaskMetadata(propertyStore, TASK_TYPE, TABLE_NAME_WITH_TYPE),
        taskMetadata.toZNRecord());

    // the metadata will be written to the new path if only the new path exists
    propertyStore = new FakePropertyStore();
    propertyStore.set(NEW_MINION_METADATA_PATH, NEW_TASK_METADATA.toZNRecord(), EXPECTED_VERSION, ACCESS_OPTION);
    MinionTaskMetadataUtils.persistTaskMetadata(propertyStore, TASK_TYPE, taskMetadata, EXPECTED_VERSION);
    assertTrue(propertyStore.exists(NEW_MINION_METADATA_PATH, ACCESS_OPTION));
    assertFalse(propertyStore.exists(OLD_MINION_METADATA_PATH, ACCESS_OPTION));
    assertEquals(MinionTaskMetadataUtils.fetchTaskMetadata(propertyStore, TASK_TYPE, TABLE_NAME_WITH_TYPE),
        taskMetadata.toZNRecord());

    // the metadata will be written to the new path if both paths exist
    propertyStore = new FakePropertyStore();
    propertyStore.set(OLD_MINION_METADATA_PATH, OLD_TASK_METADATA.toZNRecord(), EXPECTED_VERSION, ACCESS_OPTION);
    propertyStore.set(NEW_MINION_METADATA_PATH, NEW_TASK_METADATA.toZNRecord(), EXPECTED_VERSION, ACCESS_OPTION);
    MinionTaskMetadataUtils.persistTaskMetadata(propertyStore, TASK_TYPE, taskMetadata, EXPECTED_VERSION);
    assertTrue(propertyStore.exists(NEW_MINION_METADATA_PATH, ACCESS_OPTION));
    assertTrue(propertyStore.exists(OLD_MINION_METADATA_PATH, ACCESS_OPTION));
    assertEquals(propertyStore.get(NEW_MINION_METADATA_PATH, new Stat(), ACCESS_OPTION), taskMetadata.toZNRecord());
    assertEquals(MinionTaskMetadataUtils.fetchTaskMetadata(propertyStore, TASK_TYPE, TABLE_NAME_WITH_TYPE),
        taskMetadata.toZNRecord());
  }