public final void processFiles(List<Path> sourcePaths, FeatureGroup writer, PlanetilerConfig config) {
    var timer = stats.startStage(sourceName);
    long featureCount = getFeatureCount(sourcePaths);
    int readThreads = config.featureReadThreads();
    int writeThreads = config.featureWriteThreads();
    int processThreads = config.featureProcessThreads();
    Envelope latLonBounds = config.bounds().latLon();
    AtomicLong featuresRead = new AtomicLong(0);
    AtomicLong featuresWritten = new AtomicLong(0);

    var pipeline = WorkerPipeline.start(sourceName, stats)
      .readFromTiny("source_paths", sourcePaths)
      .addWorker("read", readThreads, readPaths())
      .addBuffer("process_queue", 1000, 1)
      .<SortableFeature>addWorker("process", processThreads, (prev, next) -> {
        var featureCollectors = new FeatureCollector.Factory(config, stats);
        try (FeatureRenderer renderer = newFeatureRenderer(writer, config, next)) {
          for (SourceFeature sourceFeature : prev) {
            featuresRead.incrementAndGet();
            FeatureCollector features = featureCollectors.get(sourceFeature);
            if (sourceFeature.latLonGeometry().getEnvelopeInternal().intersects(latLonBounds)) {
              try {
                profile.processFeature(sourceFeature, features);
                for (FeatureCollector.Feature renderable : features) {
                  renderer.accept(renderable);
                }
              } catch (Exception e) {
                LOGGER.error("Error processing " + sourceFeature, e);
              }
            }
          }
        }
      })
      // output large batches since each input may map to many tiny output features (i.e. slicing ocean tiles)
      // which turns enqueueing into the bottleneck
      .addBuffer("write_queue", 50_000, 1_000)
      .sinkTo("write", writeThreads, prev -> {
        try (var threadLocalWriter = writer.writerForThread()) {
          for (var item : prev) {
            featuresWritten.incrementAndGet();
            threadLocalWriter.accept(item);
          }
        }
      });

    var loggers = ProgressLoggers.create()
      .addRatePercentCounter("read", featureCount, featuresRead, true)
      .addRateCounter("write", featuresWritten)
      .addFileSize(writer)
      .newLine()
      .addProcessStats()
      .newLine()
      .addPipelineStats(pipeline);

    pipeline.awaitAndLog(loggers, config.logInterval());

    // hook for profile to do any post-processing after this source is read
    try (
      var threadLocalWriter = writer.writerForThread();
      var featureRenderer = newFeatureRenderer(writer, config, threadLocalWriter)
    ) {
      profile.finish(sourceName, new FeatureCollector.Factory(config, stats), featureRenderer);
    } catch (IOException e) {
      LOGGER.warn("Error closing writer", e);
    }
    timer.stop();
  }
Renders map features for all elements from this data source based on the mapping logic defined in {@code profile}.

@param sourcePaths list of paths associated with this source
@param writer      consumer for rendered features
@param config      user-defined parameters controlling number of threads and log interval
planetiler-core/src/main/java/com/onthegomap/planetiler/reader/SourceFeatureProcessor.java