@VisibleForTesting
  static void validateTaskConfigs(TableConfig tableConfig, Schema schema) {
    TableTaskConfig taskConfig = tableConfig.getTaskConfig();
    if (taskConfig != null) {
      for (Map.Entry<String, Map<String, String>> taskConfigEntry : taskConfig.getTaskTypeConfigsMap().entrySet()) {
        String taskTypeConfigName = taskConfigEntry.getKey();
        Map<String, String> taskTypeConfig = taskConfigEntry.getValue();
        // Task configuration cannot be null.
        Preconditions.checkNotNull(taskTypeConfig,
            String.format("Task configuration for \"%s\" cannot be null!", taskTypeConfigName));
        // Schedule key for task config has to be set.
        if (taskTypeConfig.containsKey(SCHEDULE_KEY)) {
          String cronExprStr = taskTypeConfig.get(SCHEDULE_KEY);
          try {
            CronScheduleBuilder.cronSchedule(cronExprStr);
          } catch (Exception e) {
            throw new IllegalStateException(
                String.format("Task %s contains an invalid cron schedule: %s", taskTypeConfigName, cronExprStr), e);
          }
        }
        boolean isAllowDownloadFromServer = Boolean.parseBoolean(
            taskTypeConfig.getOrDefault(TableTaskConfig.MINION_ALLOW_DOWNLOAD_FROM_SERVER,
                String.valueOf(TableTaskConfig.DEFAULT_MINION_ALLOW_DOWNLOAD_FROM_SERVER)));
        if (isAllowDownloadFromServer) {
          Preconditions.checkState(tableConfig.getValidationConfig().getPeerSegmentDownloadScheme() != null,
              String.format("Table %s has task %s with allowDownloadFromServer set to true, but "
                      + "peerSegmentDownloadScheme is not set in the table config", tableConfig.getTableName(),
                  taskTypeConfigName));
        }
        // Task Specific validation for REALTIME_TO_OFFLINE_TASK_TYPE
        // TODO task specific validate logic should directly call to PinotTaskGenerator API
        if (taskTypeConfigName.equals(REALTIME_TO_OFFLINE_TASK_TYPE)) {
          // check table is not upsert
          Preconditions.checkState(tableConfig.getUpsertMode() == UpsertConfig.Mode.NONE,
              "RealtimeToOfflineTask doesn't support upsert table!");
          // check no malformed period
          TimeUtils.convertPeriodToMillis(taskTypeConfig.getOrDefault("bufferTimePeriod", "2d"));
          TimeUtils.convertPeriodToMillis(taskTypeConfig.getOrDefault("bucketTimePeriod", "1d"));
          TimeUtils.convertPeriodToMillis(taskTypeConfig.getOrDefault("roundBucketTimePeriod", "1s"));
          // check mergeType is correct
          Preconditions.checkState(ImmutableSet.of("CONCAT", "ROLLUP", "DEDUP")
                  .contains(taskTypeConfig.getOrDefault("mergeType", "CONCAT").toUpperCase()),
              "MergeType must be one of [CONCAT, ROLLUP, DEDUP]!");
          // check no mis-configured columns
          Set<String> columnNames = schema.getColumnNames();
          for (Map.Entry<String, String> entry : taskTypeConfig.entrySet()) {
            if (entry.getKey().endsWith(".aggregationType")) {
              Preconditions.checkState(columnNames.contains(StringUtils.removeEnd(entry.getKey(), ".aggregationType")),
                  String.format("Column \"%s\" not found in schema!", entry.getKey()));
              try {
                // check that it's a valid aggregation function type
                AggregationFunctionType aft = AggregationFunctionType.getAggregationFunctionType(entry.getValue());
                // check that a value aggregator is available
                if (!AVAILABLE_CORE_VALUE_AGGREGATORS.contains(aft)) {
                  throw new IllegalArgumentException("ValueAggregator not enabled for type: " + aft.toString());
                }
              } catch (IllegalArgumentException e) {
                String err =
                    String.format("Column \"%s\" has invalid aggregate type: %s", entry.getKey(), entry.getValue());
                throw new IllegalStateException(err);
              }
            }
          }
        } else if (taskTypeConfigName.equals(UPSERT_COMPACTION_TASK_TYPE)) {
          // check table is realtime
          Preconditions.checkState(tableConfig.getTableType() == TableType.REALTIME,
              "UpsertCompactionTask only supports realtime tables!");
          // check upsert enabled
          Preconditions.checkState(tableConfig.isUpsertEnabled(), "Upsert must be enabled for UpsertCompactionTask");

          // check no malformed period
          if (taskTypeConfig.containsKey("bufferTimePeriod")) {
            TimeUtils.convertPeriodToMillis(taskTypeConfig.get("bufferTimePeriod"));
          }
          // check invalidRecordsThresholdPercent
          if (taskTypeConfig.containsKey("invalidRecordsThresholdPercent")) {
            Preconditions.checkState(Double.parseDouble(taskTypeConfig.get("invalidRecordsThresholdPercent")) >= 0
                    && Double.parseDouble(taskTypeConfig.get("invalidRecordsThresholdPercent")) <= 100,
                "invalidRecordsThresholdPercent must be >= 0 and <= 100");
          }
          // check invalidRecordsThresholdCount
          if (taskTypeConfig.containsKey("invalidRecordsThresholdCount")) {
            Preconditions.checkState(Long.parseLong(taskTypeConfig.get("invalidRecordsThresholdCount")) >= 1,
                "invalidRecordsThresholdCount must be >= 1");
          }
          // check that either invalidRecordsThresholdPercent or invalidRecordsThresholdCount was provided
          Preconditions.checkState(
              taskTypeConfig.containsKey("invalidRecordsThresholdPercent") || taskTypeConfig.containsKey(
                  "invalidRecordsThresholdCount"),
              "invalidRecordsThresholdPercent or invalidRecordsThresholdCount or both must be provided");
          String validDocIdsType = taskTypeConfig.getOrDefault("validDocIdsType", "snapshot");
          if (validDocIdsType.equals(ValidDocIdsType.SNAPSHOT.toString())) {
            UpsertConfig upsertConfig = tableConfig.getUpsertConfig();
            Preconditions.checkNotNull(upsertConfig, "UpsertConfig must be provided for UpsertCompactionTask");
            Preconditions.checkState(upsertConfig.isEnableSnapshot(), String.format(
                "'enableSnapshot' from UpsertConfig must be enabled for UpsertCompactionTask with validDocIdsType = "
                    + "%s", validDocIdsType));
          } else if (validDocIdsType.equals(ValidDocIdsType.IN_MEMORY_WITH_DELETE.toString())) {
            UpsertConfig upsertConfig = tableConfig.getUpsertConfig();
            Preconditions.checkNotNull(upsertConfig, "UpsertConfig must be provided for UpsertCompactionTask");
            Preconditions.checkNotNull(upsertConfig.getDeleteRecordColumn(), String.format(
                "deleteRecordColumn must be provided for " + "UpsertCompactionTask with validDocIdsType = %s",
                validDocIdsType));
          }
        }
      }
    }
  }
@Test
  public void testUpsertCompactionTaskConfig() {
    Schema schema =
        new Schema.SchemaBuilder().setSchemaName(TABLE_NAME).addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
            .addDateTime(TIME_COLUMN, FieldSpec.DataType.LONG, "1:MILLISECONDS:EPOCH", "1:MILLISECONDS")
            .setPrimaryKeyColumns(Lists.newArrayList("myCol")).build();
    Map<String, String> upsertCompactionTaskConfig =
        ImmutableMap.of("bufferTimePeriod", "5d", "invalidRecordsThresholdPercent", "1", "invalidRecordsThresholdCount",
            "1");
    UpsertConfig upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
    upsertConfig.setEnableSnapshot(true);
    TableConfig tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
        .setUpsertConfig(upsertConfig)
        .setTaskConfig(new TableTaskConfig(ImmutableMap.of("UpsertCompactionTask", upsertCompactionTaskConfig)))
        .build();

    TableConfigUtils.validateTaskConfigs(tableConfig, schema);

    // test with invalidRecordsThresholdPercents as 0
    upsertCompactionTaskConfig = ImmutableMap.of("invalidRecordsThresholdPercent", "0");
    TableConfig zeroPercentTableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
        .setUpsertConfig(upsertConfig)
        .setTaskConfig(new TableTaskConfig(ImmutableMap.of("UpsertCompactionTask", upsertCompactionTaskConfig)))
        .build();
    TableConfigUtils.validateTaskConfigs(zeroPercentTableConfig, schema);

    // test with invalid invalidRecordsThresholdPercents as -1 and 110
    upsertCompactionTaskConfig = ImmutableMap.of("invalidRecordsThresholdPercent", "-1");
    TableConfig negativePercentTableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
        .setUpsertConfig(upsertConfig)
        .setTaskConfig(new TableTaskConfig(ImmutableMap.of("UpsertCompactionTask", upsertCompactionTaskConfig)))
        .build();
    Assert.assertThrows(IllegalStateException.class,
        () -> TableConfigUtils.validateTaskConfigs(negativePercentTableConfig, schema));
    upsertCompactionTaskConfig = ImmutableMap.of("invalidRecordsThresholdPercent", "110");
    TableConfig hundredTenPercentTableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
        .setUpsertConfig(new UpsertConfig(UpsertConfig.Mode.FULL))
        .setTaskConfig(new TableTaskConfig(ImmutableMap.of("UpsertCompactionTask", upsertCompactionTaskConfig)))
        .build();
    Assert.assertThrows(IllegalStateException.class,
        () -> TableConfigUtils.validateTaskConfigs(hundredTenPercentTableConfig, schema));

    // test with invalid invalidRecordsThresholdCount
    upsertCompactionTaskConfig = ImmutableMap.of("invalidRecordsThresholdCount", "0");
    TableConfig invalidCountTableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
        .setUpsertConfig(new UpsertConfig(UpsertConfig.Mode.FULL))
        .setTaskConfig(new TableTaskConfig(ImmutableMap.of("UpsertCompactionTask", upsertCompactionTaskConfig)))
        .build();
    Assert.assertThrows(IllegalStateException.class,
        () -> TableConfigUtils.validateTaskConfigs(invalidCountTableConfig, schema));

    // test without invalidRecordsThresholdPercent or invalidRecordsThresholdCount
    upsertCompactionTaskConfig = ImmutableMap.of("bufferTimePeriod", "5d");
    TableConfig invalidTableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
        .setUpsertConfig(new UpsertConfig(UpsertConfig.Mode.FULL))
        .setTaskConfig(new TableTaskConfig(ImmutableMap.of("UpsertCompactionTask", upsertCompactionTaskConfig)))
        .build();
    Assert.assertThrows(IllegalStateException.class,
        () -> TableConfigUtils.validateTaskConfigs(invalidTableConfig, schema));
  }