protected SegmentGenerationTaskSpec generateTaskSpec(Map<String, String> taskConfigs, File localTempDir)
      throws Exception {
    SegmentGenerationTaskSpec taskSpec = new SegmentGenerationTaskSpec();
    URI inputFileURI = URI.create(taskConfigs.get(BatchConfigProperties.INPUT_DATA_FILE_URI_KEY));

    try (PinotFS inputFileFS = MinionTaskUtils.getInputPinotFS(taskConfigs, inputFileURI)) {
      File localInputTempDir = new File(localTempDir, "input");
      FileUtils.forceMkdir(localInputTempDir);
      File localOutputTempDir = new File(localTempDir, "output");
      FileUtils.forceMkdir(localOutputTempDir);
      taskSpec.setOutputDirectoryPath(localOutputTempDir.getAbsolutePath());

      //copy input path to local
      _eventObserver.notifyProgress(_pinotTaskConfig, String.format("Copying file: %s to local disk", inputFileURI));
      File localInputDataFile = new File(localInputTempDir, new File(inputFileURI.getPath()).getName());
      inputFileFS.copyToLocalFile(inputFileURI, localInputDataFile);
      taskSpec.setInputFilePath(localInputDataFile.getAbsolutePath());

      RecordReaderSpec recordReaderSpec = new RecordReaderSpec();
      recordReaderSpec.setDataFormat(taskConfigs.get(BatchConfigProperties.INPUT_FORMAT));
      recordReaderSpec.setClassName(taskConfigs.get(BatchConfigProperties.RECORD_READER_CLASS));
      recordReaderSpec.setConfigClassName(taskConfigs.get(BatchConfigProperties.RECORD_READER_CONFIG_CLASS));
      recordReaderSpec.setConfigs(IngestionConfigUtils.getConfigMapWithPrefix(taskConfigs,
          BatchConfigProperties.RECORD_READER_PROP_PREFIX));
      taskSpec.setRecordReaderSpec(recordReaderSpec);

      String authToken = taskConfigs.get(BatchConfigProperties.AUTH_TOKEN);

      String tableNameWithType = taskConfigs.get(BatchConfigProperties.TABLE_NAME);
      Schema schema;
      if (taskConfigs.containsKey(BatchConfigProperties.SCHEMA)) {
        schema = JsonUtils.stringToObject(JsonUtils.objectToString(taskConfigs.get(BatchConfigProperties.SCHEMA)),
            Schema.class);
      } else if (taskConfigs.containsKey(BatchConfigProperties.SCHEMA_URI)) {
        schema = SegmentGenerationUtils.getSchema(taskConfigs.get(BatchConfigProperties.SCHEMA_URI), authToken);
      } else {
        schema = getSchema(tableNameWithType);
      }
      taskSpec.setSchema(schema);
      TableConfig tableConfig;
      if (taskConfigs.containsKey(BatchConfigProperties.TABLE_CONFIGS)) {
        tableConfig = JsonUtils.stringToObject(taskConfigs.get(BatchConfigProperties.TABLE_CONFIGS), TableConfig.class);
      } else if (taskConfigs.containsKey(BatchConfigProperties.TABLE_CONFIGS_URI)) {
        tableConfig = SegmentGenerationUtils.getTableConfig(
            taskConfigs.get(BatchConfigProperties.TABLE_CONFIGS_URI), authToken);
      } else {
        tableConfig = getTableConfig(tableNameWithType);
      }
      taskSpec.setTableConfig(tableConfig);
      taskSpec.setSequenceId(Integer.parseInt(taskConfigs.get(BatchConfigProperties.SEQUENCE_ID)));
      if (taskConfigs.containsKey(BatchConfigProperties.FAIL_ON_EMPTY_SEGMENT)) {
        taskSpec.setFailOnEmptySegment(
            Boolean.parseBoolean(taskConfigs.get(BatchConfigProperties.FAIL_ON_EMPTY_SEGMENT)));
      }
      SegmentNameGeneratorSpec segmentNameGeneratorSpec = new SegmentNameGeneratorSpec();
      segmentNameGeneratorSpec.setType(taskConfigs.get(BatchConfigProperties.SEGMENT_NAME_GENERATOR_TYPE));
      segmentNameGeneratorSpec.setConfigs(IngestionConfigUtils.getConfigMapWithPrefix(taskConfigs,
          BatchConfigProperties.SEGMENT_NAME_GENERATOR_PROP_PREFIX));
      segmentNameGeneratorSpec.addConfig(SegmentGenerationTaskRunner.APPEND_UUID_TO_SEGMENT_NAME,
          taskConfigs.getOrDefault(BatchConfigProperties.APPEND_UUID_TO_SEGMENT_NAME, Boolean.toString(false)));
      taskSpec.setSegmentNameGeneratorSpec(segmentNameGeneratorSpec);
      taskSpec.setCustomProperty(BatchConfigProperties.INPUT_DATA_FILE_URI_KEY, inputFileURI.toString());

      return taskSpec;
    }
  }
@Test
  public void testGenerateTaskSpec()
      throws Exception {
    ClassLoader classLoader = getClass().getClassLoader();
    URL resourcesLoc = classLoader.getResource(".");
    assertNotNull(resourcesLoc);
    URL tableConfigUrl = classLoader.getResource("dummyTable.json");
    assertNotNull(tableConfigUrl);
    String tableConfig = FileUtils.readFileToString(new File(tableConfigUrl.getFile()), StandardCharsets.UTF_8);
    SegmentGenerationAndPushTaskExecutor executor = new SegmentGenerationAndPushTaskExecutor();
    Schema schema = new Schema.SchemaBuilder().build();
    FieldUtils.writeField(executor, "_eventObserver", new DefaultMinionEventObserver(), true);
    Map<String, String> configMap = new HashMap<>() {{
      put(BatchConfigProperties.INPUT_DATA_FILE_URI_KEY, resourcesLoc + "dummyTable.json");
      put(BatchConfigProperties.INPUT_FORMAT, "");
      put(BatchConfigProperties.RECORD_READER_CLASS, "AReaderClass");
      put(BatchConfigProperties.RECORD_READER_CONFIG_CLASS, "AReaderConfigClass");
      put(BatchConfigProperties.RECORD_READER_PROP_PREFIX + ".prop1", "value1");
      put(BatchConfigProperties.RECORD_READER_PROP_PREFIX + ".prop.2", "value2");
      put(BatchConfigProperties.AUTH_TOKEN, "not_used");
      put(BatchConfigProperties.TABLE_NAME, "not_used");
      put(BatchConfigProperties.SCHEMA, schema.toSingleLineJsonString());
      put(BatchConfigProperties.SCHEMA_URI, "not_used");
      put(BatchConfigProperties.TABLE_CONFIGS, tableConfig);
      put(BatchConfigProperties.TABLE_CONFIGS_URI, "not_used");
      put(BatchConfigProperties.SEQUENCE_ID, "42");
      put(BatchConfigProperties.FAIL_ON_EMPTY_SEGMENT, "true");
      put(BatchConfigProperties.SEGMENT_NAME_GENERATOR_TYPE, "inputtext");
      put(BatchConfigProperties.SEGMENT_NAME_GENERATOR_PROP_PREFIX + ".prop.seg.1", "valseg1");
      put(BatchConfigProperties.SEGMENT_NAME_GENERATOR_PROP_PREFIX + ".propseg2", "valseg2");
      put(BatchConfigProperties.APPEND_UUID_TO_SEGMENT_NAME, "true");
    }};
    SegmentGenerationTaskSpec spec = executor.generateTaskSpec(configMap, Paths.get(resourcesLoc.toURI()).toFile());
    assertEquals(spec.getSequenceId(), 42);
    assertEquals("file:" + spec.getInputFilePath(), resourcesLoc + "input/dummyTable.json");
    assertEquals(spec.getRecordReaderSpec().getClassName(), "AReaderClass");
    assertEquals(spec.getRecordReaderSpec().getConfigClassName(), "AReaderConfigClass");
    assertEqualsDeep(spec.getRecordReaderSpec().getConfigs(), Map.of("prop1", "value1", "prop.2", "value2"));
    assertTrue(spec.isFailOnEmptySegment());
    assertEquals(spec.getSegmentNameGeneratorSpec().getType(), "inputtext");
    assertEqualsDeep(spec.getSegmentNameGeneratorSpec().getConfigs(),
        Map.of("prop.seg.1", "valseg1", "propseg2", "valseg2", SegmentGenerationTaskRunner.APPEND_UUID_TO_SEGMENT_NAME,
            "true"));
  }