public static Map<String, Map<String, Long>> getAllTaskMetadataLastUpdateTimeMs(
      HelixPropertyStore<ZNRecord> propertyStore) {
    Map<String, Map<String, Long>> tableTaskLastUpdateTimeMsMap = new HashMap<>();
    String propertyStorePathForMinionTaskMetadataPrefix =
        ZKMetadataProvider.getPropertyStorePathForMinionTaskMetadataPrefix();
    // the old and new path may exist at the same time
    List<String> tableNameWithTypeOrTaskTypes =
        propertyStore.getChildNames(propertyStorePathForMinionTaskMetadataPrefix, AccessOption.PERSISTENT);
    if (tableNameWithTypeOrTaskTypes == null || tableNameWithTypeOrTaskTypes.isEmpty()) {
      return tableTaskLastUpdateTimeMsMap;
    }
    for (String tableNameWithTypeOrTaskType : tableNameWithTypeOrTaskTypes) {
      String metadataNodeDirectParentPath =
          StringUtil.join("/", propertyStorePathForMinionTaskMetadataPrefix, tableNameWithTypeOrTaskType);
      List<String> metadataNodeNames =
          propertyStore.getChildNames(metadataNodeDirectParentPath, AccessOption.PERSISTENT);
      if (metadataNodeNames == null || metadataNodeNames.isEmpty()) {
        continue;
      }
      // the new path is MINION_TASK_METADATA/${tableNameWthType}/${taskType}
      // the old path is MINION_TASK_METADATA/${taskType}/${tableNameWthType}
      // The variable tableNameWithTypeOrTaskType stores the first level child name of MINION_TASK_METADATA, that's why
      // when it ends with OFFLINE or REALTIME, it is a new path.
      boolean isNewPath =
          tableNameWithTypeOrTaskType.endsWith(TableType.OFFLINE.toString()) || tableNameWithTypeOrTaskType.endsWith(
              TableType.REALTIME.toString());
      for (String metadataNodeName : metadataNodeNames) {
        String metadataNodePath = StringUtil.join("/", metadataNodeDirectParentPath, metadataNodeName);
        Stat stat = propertyStore.getStat(metadataNodePath, AccessOption.PERSISTENT);
        if (isNewPath) {
          saveOrUpdateTaskMetadataLastUpdateTime(tableNameWithTypeOrTaskType, metadataNodeName, stat.getMtime(),
              tableTaskLastUpdateTimeMsMap);
        } else {
          saveOrUpdateTaskMetadataLastUpdateTime(metadataNodeName, tableNameWithTypeOrTaskType, stat.getMtime(),
              tableTaskLastUpdateTimeMsMap);
        }
      }
    }
    return tableTaskLastUpdateTimeMsMap;
  }
@Test
  public void testGetAllTaskMetadataLastUpdateTimeMs() {
    // no task metadata exists
    HelixPropertyStore<ZNRecord> propertyStore = new FakePropertyStore();
    assertTrue(MinionTaskMetadataUtils.getAllTaskMetadataLastUpdateTimeMs(propertyStore).isEmpty());

    // only the old metadata path exists
    propertyStore = new FakePropertyStore();
    long tsBeforeSet = System.currentTimeMillis();
    propertyStore.set(OLD_MINION_METADATA_PATH, OLD_TASK_METADATA.toZNRecord(), EXPECTED_VERSION, ACCESS_OPTION);
    long tsAfterSet = System.currentTimeMillis();
    Map<String, Map<String, Long>> allTaskMetadataLastUpdateTimeMs =
        MinionTaskMetadataUtils.getAllTaskMetadataLastUpdateTimeMs(propertyStore);
    assertEquals(allTaskMetadataLastUpdateTimeMs.size(), 1);
    Map<String, Long> taskTypeLastUpdateMsMap = allTaskMetadataLastUpdateTimeMs.get(TABLE_NAME_WITH_TYPE);
    assertEquals(taskTypeLastUpdateMsMap.size(), 1);
    long lastUpdateTimeMs = taskTypeLastUpdateMsMap.get(TASK_TYPE);
    assertTrue(lastUpdateTimeMs >= tsBeforeSet && lastUpdateTimeMs <= tsAfterSet);

    // only the new metadata path exists
    propertyStore = new FakePropertyStore();
    tsBeforeSet = System.currentTimeMillis();
    propertyStore.set(NEW_MINION_METADATA_PATH, NEW_TASK_METADATA.toZNRecord(), EXPECTED_VERSION, ACCESS_OPTION);
    tsAfterSet = System.currentTimeMillis();
    allTaskMetadataLastUpdateTimeMs =
        MinionTaskMetadataUtils.getAllTaskMetadataLastUpdateTimeMs(propertyStore);
    assertEquals(allTaskMetadataLastUpdateTimeMs.size(), 1);
    taskTypeLastUpdateMsMap = allTaskMetadataLastUpdateTimeMs.get(TABLE_NAME_WITH_TYPE);
    assertEquals(taskTypeLastUpdateMsMap.size(), 1);
    lastUpdateTimeMs = taskTypeLastUpdateMsMap.get(TASK_TYPE);
    assertTrue(lastUpdateTimeMs >= tsBeforeSet && lastUpdateTimeMs <= tsAfterSet);

    // if two metadata paths exist at the same time, the newly updated one will be used.
    // the new metadata path is updated later
    propertyStore = new FakePropertyStore();
    propertyStore.set(OLD_MINION_METADATA_PATH, OLD_TASK_METADATA.toZNRecord(), EXPECTED_VERSION, ACCESS_OPTION);
    long tsAfterOldPathSet = System.currentTimeMillis();
    propertyStore.set(NEW_MINION_METADATA_PATH, NEW_TASK_METADATA.toZNRecord(), EXPECTED_VERSION, ACCESS_OPTION);
    long tsAfterNewPathSet = System.currentTimeMillis();
    allTaskMetadataLastUpdateTimeMs =
        MinionTaskMetadataUtils.getAllTaskMetadataLastUpdateTimeMs(propertyStore);
    assertEquals(allTaskMetadataLastUpdateTimeMs.size(), 1);
    taskTypeLastUpdateMsMap = allTaskMetadataLastUpdateTimeMs.get(TABLE_NAME_WITH_TYPE);
    assertEquals(taskTypeLastUpdateMsMap.size(), 1);
    lastUpdateTimeMs = taskTypeLastUpdateMsMap.get(TASK_TYPE);
    assertTrue(lastUpdateTimeMs >= tsAfterOldPathSet && lastUpdateTimeMs <= tsAfterNewPathSet);
    // the old metadata path is updated later
    propertyStore = new FakePropertyStore();
    propertyStore.set(NEW_MINION_METADATA_PATH, NEW_TASK_METADATA.toZNRecord(), EXPECTED_VERSION, ACCESS_OPTION);
    tsAfterNewPathSet = System.currentTimeMillis();
    propertyStore.set(OLD_MINION_METADATA_PATH, OLD_TASK_METADATA.toZNRecord(), EXPECTED_VERSION, ACCESS_OPTION);
    tsAfterOldPathSet = System.currentTimeMillis();
    allTaskMetadataLastUpdateTimeMs =
        MinionTaskMetadataUtils.getAllTaskMetadataLastUpdateTimeMs(propertyStore);
    assertEquals(allTaskMetadataLastUpdateTimeMs.size(), 1);
    taskTypeLastUpdateMsMap = allTaskMetadataLastUpdateTimeMs.get(TABLE_NAME_WITH_TYPE);
    assertEquals(taskTypeLastUpdateMsMap.size(), 1);
    lastUpdateTimeMs = taskTypeLastUpdateMsMap.get(TASK_TYPE);
    assertTrue(lastUpdateTimeMs >= tsAfterNewPathSet && lastUpdateTimeMs <= tsAfterOldPathSet);
  }