@Nullable
  public static ZNRecord fetchTaskMetadata(HelixPropertyStore<ZNRecord> propertyStore, String taskType,
      String tableNameWithType) {
    String newPath = ZKMetadataProvider.constructPropertyStorePathForMinionTaskMetadata(tableNameWithType, taskType);
    if (propertyStore.exists(newPath, AccessOption.PERSISTENT)) {
      return fetchTaskMetadata(propertyStore, newPath);
    } else {
      return fetchTaskMetadata(propertyStore,
          ZKMetadataProvider.constructPropertyStorePathForMinionTaskMetadataDeprecated(taskType, tableNameWithType));
    }
  }
@Test
  public void testFetchTaskMetadata() {
    // no metadata path exists
    HelixPropertyStore<ZNRecord> propertyStore = new FakePropertyStore();
    assertNull(MinionTaskMetadataUtils.fetchTaskMetadata(propertyStore, TASK_TYPE, TABLE_NAME_WITH_TYPE));

    // only the old metadata path exists
    propertyStore = new FakePropertyStore();
    propertyStore.set(OLD_MINION_METADATA_PATH, OLD_TASK_METADATA.toZNRecord(), EXPECTED_VERSION, ACCESS_OPTION);
    assertEquals(MinionTaskMetadataUtils.fetchTaskMetadata(propertyStore, TASK_TYPE, TABLE_NAME_WITH_TYPE),
        OLD_TASK_METADATA.toZNRecord());

    // only the new metadata path exists
    propertyStore = new FakePropertyStore();
    propertyStore.set(NEW_MINION_METADATA_PATH, NEW_TASK_METADATA.toZNRecord(), EXPECTED_VERSION, ACCESS_OPTION);
    assertEquals(MinionTaskMetadataUtils.fetchTaskMetadata(propertyStore, TASK_TYPE, TABLE_NAME_WITH_TYPE),
        NEW_TASK_METADATA.toZNRecord());

    // if two metadata paths exist at the same time, the new one will be used.
    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);
    assertEquals(MinionTaskMetadataUtils.fetchTaskMetadata(propertyStore, TASK_TYPE, TABLE_NAME_WITH_TYPE),
        NEW_TASK_METADATA.toZNRecord());
  }