docs: private TaskStatus generateAndPublishSegments(
        final TaskToolbox toolbox,
        final DataSchema dataSchema,
        final ShardSpecs shardSpecs,
        final Map<Interval, String> versions,
        final FirehoseFactory firehoseFactory,
        final File firehoseTempDir,
        @Nullable final Integer maxRowsPerSegment,
        @Nullable final Long maxTotalRows
    ) throws IOException, InterruptedException
    {
      final GranularitySpec granularitySpec = dataSchema.getGranularitySpec();
      final FireDepartment fireDepartmentForMetrics =
          new FireDepartment(dataSchema, new RealtimeIOConfig(null, null, null), null);
      buildSegmentsFireDepartmentMetrics = fireDepartmentForMetrics.getMetrics();
  
      if (toolbox.getMonitorScheduler() != null) {
        final TaskRealtimeMetricsMonitor metricsMonitor = TaskRealtimeMetricsMonitorBuilder.build(
            this,
            fireDepartmentForMetrics,
            buildSegmentsMeters
        );
        toolbox.getMonitorScheduler().addMonitor(metricsMonitor);
      }
  
      final IndexIOConfig ioConfig = ingestionSchema.getIOConfig();
      final IndexTuningConfig tuningConfig = ingestionSchema.tuningConfig;
      final long pushTimeout = tuningConfig.getPushTimeout();
      final boolean isGuaranteedRollup = isGuaranteedRollup(ioConfig, tuningConfig);
  
      final SegmentAllocator segmentAllocator;
      if (isGuaranteedRollup) {
        // Overwrite mode, guaranteed rollup: segments are all known in advance and there is one per sequenceName.
        final Map<String, SegmentIdWithShardSpec> lookup = new HashMap<>();
  
        for (Map.Entry<Interval, List<ShardSpec>> entry : shardSpecs.getMap().entrySet()) {
          for (ShardSpec shardSpec : entry.getValue()) {
            final String version = findVersion(versions, entry.getKey());
            lookup.put(
                Appenderators.getSequenceName(entry.getKey(), version, shardSpec),
                new SegmentIdWithShardSpec(getDataSource(), entry.getKey(), version, shardSpec)
            );
          }
        }
  
        segmentAllocator = (row, sequenceName, previousSegmentId, skipSegmentLineageCheck) -> lookup.get(sequenceName);
      } else if (ioConfig.isAppendToExisting()) {
        // Append mode: Allocate segments as needed using Overlord APIs.
        segmentAllocator = new ActionBasedSegmentAllocator(
            toolbox.getTaskActionClient(),
            dataSchema,
            (schema, row, sequenceName, previousSegmentId, skipSegmentLineageCheck) -> new SegmentAllocateAction(
                schema.getDataSource(),
                row.getTimestamp(),
                schema.getGranularitySpec().getQueryGranularity(),
                schema.getGranularitySpec().getSegmentGranularity(),
                sequenceName,
                previousSegmentId,
                skipSegmentLineageCheck
            )
        );
      } else {
        // Overwrite mode, non-guaranteed rollup: We can make up our own segment ids but we don't know them in advance.
        final Map<Interval, AtomicInteger> counters = new HashMap<>();
  
        segmentAllocator = (row, sequenceName, previousSegmentId, skipSegmentLineageCheck) -> {
          final DateTime timestamp = row.getTimestamp();
          Optional<Interval> maybeInterval = granularitySpec.bucketInterval(timestamp);
          if (!maybeInterval.isPresent()) {
            throw new ISE("Could not find interval for timestamp [%s]", timestamp);
          }
  
          final Interval interval = maybeInterval.get();
          if (!shardSpecs.getMap().containsKey(interval)) {
            throw new ISE("Could not find shardSpec for interval[%s]", interval);
          }
  
          final int partitionNum = counters.computeIfAbsent(interval, x -> new AtomicInteger()).getAndIncrement();
          return new SegmentIdWithShardSpec(
              getDataSource(),
              interval,
              findVersion(versions, interval),
              new NumberedShardSpec(partitionNum, 0)
          );
        };
      }
  
      final TransactionalSegmentPublisher publisher = (segments, commitMetadata) -> {
        final SegmentTransactionalInsertAction action = new SegmentTransactionalInsertAction(segments);
        return toolbox.getTaskActionClient().submit(action);
      };
  
      try (
          final Appenderator appenderator = newAppenderator(
              buildSegmentsFireDepartmentMetrics,
              toolbox,
              dataSchema,
              tuningConfig
          );
          final BatchAppenderatorDriver driver = newDriver(appenderator, toolbox, segmentAllocator);
          final Firehose firehose = firehoseFactory.connect(dataSchema.getParser(), firehoseTempDir)
      ) {
        driver.startJob();
  
        while (firehose.hasMore()) {
          try {
            final InputRow inputRow = firehose.nextRow();
  
            if (inputRow == null) {
              buildSegmentsMeters.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);
            }
  
            final Optional<Interval> optInterval = granularitySpec.bucketInterval(inputRow.getTimestamp());
            if (!optInterval.isPresent()) {
              buildSegmentsMeters.incrementThrownAway();
              continue;
            }
  
            final String sequenceName;
  
            if (isGuaranteedRollup) {
              // Sequence name is based solely on the shardSpec, and there will only be one segment per sequence.
              final Interval interval = optInterval.get();
              final ShardSpec shardSpec = shardSpecs.getShardSpec(interval, inputRow);
              sequenceName = Appenderators.getSequenceName(interval, findVersion(versions, interval), shardSpec);
            } else {
              // 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).
              sequenceName = getId();
            }
            final AppenderatorDriverAddResult addResult = driver.add(inputRow, sequenceName);
  
            if (addResult.isOk()) {
              // incremental segment publishment is allowed only when rollup don't have to be perfect.
              if (!isGuaranteedRollup && 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);
                log.info("Pushed segments[%s]", pushed.getSegments());
              }
            } else {
              throw new ISE("Failed to add a row with timestamp[%s]", inputRow.getTimestamp());
            }
  
            if (addResult.getParseException() != null) {
              handleParseException(addResult.getParseException());
            } else {
              buildSegmentsMeters.incrementProcessed();
            }
          }
          catch (ParseException e) {
            handleParseException(e);
          }
        }
  
        final SegmentsAndMetadata pushed = driver.pushAllAndClear(pushTimeout);
        log.info("Pushed segments[%s]", pushed.getSegments());
  
        final SegmentsAndMetadata published = awaitPublish(
            driver.publishAll(publisher),
            pushTimeout
        );
  
        ingestionState = IngestionState.COMPLETED;
        if (published == null) {
          log.error("Failed to publish segments, aborting!");
          errorMsg = "Failed to publish segments.";
          toolbox.getTaskReportFileWriter().write(getTaskCompletionReports());
          return TaskStatus.failure(
              getId(),
              errorMsg
          );
        } else {
          log.info(
              "Processed[%,d] events, unparseable[%,d], thrownAway[%,d].",
              buildSegmentsMeters.getProcessed(),
              buildSegmentsMeters.getUnparseable(),
              buildSegmentsMeters.getThrownAway()
          );
          log.info("Published segments: %s", Lists.transform(published.getSegments(), DataSegment::getId));
  
          toolbox.getTaskReportFileWriter().write(getTaskCompletionReports());
          return TaskStatus.success(getId());
        }
      }
      catch (TimeoutException | ExecutionException e) {
        throw new RuntimeException(e);
      }
    }