public static Map<String, Map<String, String>> getLevelToConfigMap(Map<String, String> taskConfig) {
    Map<String, Map<String, String>> levelToConfigMap = new TreeMap<>();
    for (Map.Entry<String, String> entry : taskConfig.entrySet()) {
      String key = entry.getKey();
      for (String configKey : VALID_CONFIG_KEYS) {
        if (key.endsWith(configKey)) {
          String level = key.substring(0, key.length() - configKey.length() - 1);
          levelToConfigMap.computeIfAbsent(level, k -> new TreeMap<>()).put(configKey, entry.getValue());
        }
      }
    }
    return levelToConfigMap;
  }
@Test
  public void testGetLevelToConfigMap() {
    Map<String, String> taskConfig = new HashMap<>();
    taskConfig.put("daily.bucketTimePeriod", "1d");
    taskConfig.put("daily.bufferTimePeriod", "3d");
    taskConfig.put("daily.maxNumRecordsPerSegment", "1000000");
    taskConfig.put("monthly.bucketTimePeriod", "30d");
    taskConfig.put("monthly.bufferTimePeriod", "10d");
    taskConfig.put("monthly.roundBucketTimePeriod", "7d");
    taskConfig.put("monthly.mergeType", "rollup");
    taskConfig.put("monthly.maxNumRecordsPerTask", "5000000");
    taskConfig.put("monthly.maxNumParallelBuckets", "5");

    Map<String, Map<String, String>> levelToConfigMap = MergeRollupTaskUtils.getLevelToConfigMap(taskConfig);
    assertEquals(levelToConfigMap.size(), 2);

    Map<String, String> dailyConfig = levelToConfigMap.get("daily");
    assertNotNull(dailyConfig);
    assertEquals(dailyConfig.size(), 3);
    assertEquals(dailyConfig.get(MergeTask.BUCKET_TIME_PERIOD_KEY), "1d");
    assertEquals(dailyConfig.get(MergeTask.BUFFER_TIME_PERIOD_KEY), "3d");
    assertEquals(dailyConfig.get(MergeTask.MAX_NUM_RECORDS_PER_SEGMENT_KEY), "1000000");

    Map<String, String> monthlyConfig = levelToConfigMap.get("monthly");
    assertNotNull(monthlyConfig);
    assertEquals(monthlyConfig.size(), 6);
    assertEquals(monthlyConfig.get(MergeTask.BUCKET_TIME_PERIOD_KEY), "30d");
    assertEquals(monthlyConfig.get(MergeTask.BUFFER_TIME_PERIOD_KEY), "10d");
    assertEquals(monthlyConfig.get(MergeTask.ROUND_BUCKET_TIME_PERIOD_KEY), "7d");
    assertEquals(monthlyConfig.get(MergeTask.MERGE_TYPE_KEY), "rollup");
    assertEquals(monthlyConfig.get(MergeTask.MAX_NUM_RECORDS_PER_TASK_KEY), "5000000");
    assertEquals(monthlyConfig.get(MergeTask.MAX_NUM_PARALLEL_BUCKETS), "5");
  }