docs: private List<DataSegment> generateAndPushSegments(
        final TaskToolbox toolbox,
        final ParallelIndexTaskClient taskClient,
        final FirehoseFactory firehoseFactory,
        final File firehoseTempDir
    ) throws IOException, InterruptedException
    {
      final DataSchema dataSchema = ingestionSchema.getDataSchema();
      final GranularitySpec granularitySpec = dataSchema.getGranularitySpec();
      final FireDepartment fireDepartmentForMetrics =
          new FireDepartment(dataSchema, new RealtimeIOConfig(null, null, null), null);
      final FireDepartmentMetrics fireDepartmentMetrics = fireDepartmentForMetrics.getMetrics();
  
      if (toolbox.getMonitorScheduler() != null) {
        toolbox.getMonitorScheduler().addMonitor(
            new RealtimeMetricsMonitor(
                Collections.singletonList(fireDepartmentForMetrics),
                Collections.singletonMap(DruidMetrics.TASK_ID, new String[]{getId()})
            )
        );
      }
  
      // Initialize maxRowsPerSegment and maxTotalRows lazily
      final ParallelIndexTuningConfig tuningConfig = ingestionSchema.getTuningConfig();
      @Nullable final Integer maxRowsPerSegment = IndexTask.getValidMaxRowsPerSegment(tuningConfig);
      @Nullable final Long maxTotalRows = IndexTask.getValidMaxTotalRows(tuningConfig);
      final long pushTimeout = tuningConfig.getPushTimeout();
      final boolean explicitIntervals = granularitySpec.bucketIntervals().isPresent();
      final SegmentAllocator segmentAllocator = createSegmentAllocator(toolbox, taskClient, ingestionSchema);
  
      try (
          final Appenderator appenderator = newAppenderator(fireDepartmentMetrics, toolbox, dataSchema, tuningConfig);
          final BatchAppenderatorDriver driver = newDriver(appenderator, toolbox, segmentAllocator);
          final Firehose firehose = firehoseFactory.connect(dataSchema.getParser(), firehoseTempDir)
      ) {
        driver.startJob();
  
        final List<DataSegment> pushedSegments = new ArrayList<>();
  
        while (firehose.hasMore()) {
          try {
            final InputRow inputRow = firehose.nextRow();
  
            if (inputRow == null) {
              fireDepartmentMetrics.incrementThrownAway();
              continue;
            }
  
            if (!Intervals.ETERNITY.contains(inputRow.getTimestamp())) {
              final String errorMsg = StringUtils.format(
                  "Encountered row with timestamp that cannot be represented as a long: [%s]",
                  inputRow
              );
              throw new ParseException(errorMsg);
            }
  
            if (explicitIntervals) {
              final Optional<Interval> optInterval = granularitySpec.bucketInterval(inputRow.getTimestamp());
              if (!optInterval.isPresent()) {
                fireDepartmentMetrics.incrementThrownAway();
                continue;
              }
            }
  
            // Segments are created as needed, using a single sequence name. They may be allocated from the overlord
            // (in append mode) or may be created on our own authority (in overwrite mode).
            final String sequenceName = getId();
            final AppenderatorDriverAddResult addResult = driver.add(inputRow, sequenceName);
  
            if (addResult.isOk()) {
              if (addResult.isPushRequired(maxRowsPerSegment, maxTotalRows)) {
                // There can be some segments waiting for being published even though any rows won't be added to them.
                // If those segments are not published here, the available space in appenderator will be kept to be small
                // which makes the size of segments smaller.
                final SegmentsAndMetadata pushed = driver.pushAllAndClear(pushTimeout);
                pushedSegments.addAll(pushed.getSegments());
                log.info("Pushed segments[%s]", pushed.getSegments());
              }
            } else {
              throw new ISE("Failed to add a row with timestamp[%s]", inputRow.getTimestamp());
            }
  
            fireDepartmentMetrics.incrementProcessed();
          }
          catch (ParseException e) {
            if (tuningConfig.isReportParseExceptions()) {
              throw e;
            } else {
              fireDepartmentMetrics.incrementUnparseable();
            }
          }
        }
  
        final SegmentsAndMetadata pushed = driver.pushAllAndClear(pushTimeout);
        pushedSegments.addAll(pushed.getSegments());
        log.info("Pushed segments[%s]", pushed.getSegments());
  
        return pushedSegments;
      }
      catch (TimeoutException | ExecutionException e) {
        throw new RuntimeException(e);
      }
    }