Setting a batching period policy blocks graceful termination

It appears that configuring batch with a period policy blocks graceful termination of a Benthos process. Configuration and sample output shown below.

When I remove the period policy and only have batch_size, I am able to safely terminate the benthos process and observe the pending batch is flushed correctly.

Config

metrics:
  prometheus:
    prefix: benthos
    path_mapping: ""

input:
  file:
    paths: [${DATA_PATH}/input/pypi/*.jsonl]
    codec: lines
  processors:
    - label: json_to_csv
      bloblang: |
        map escape_csv {
          root = if this.re_match("[\"\n,]+") {
            "\"" + this.replace("\"", "\"\"") + "\""
          } else {
            this
          }
        }

        meta = deleted()

        root = [
          (this.timestamp | "").string().apply("escape_csv"),
          (this.project | "").string().apply("escape_csv"),
          (this.url | "").string().apply("escape_csv"),
          (this.details.installer.name | "").string().apply("escape_csv"),
          (this.details.installer.version | "").string().apply("escape_csv"),
          (this.details.implementation.name | "").string().apply("escape_csv"),
          (this.details.implementation.version | "").string().apply("escape_csv"),
          (this.details.system.name | "").string().apply("escape_csv"),
          (this.details.system.release | "").string().apply("escape_csv")
        ].join(",")
output:
  broker:
    outputs:
      - drop: {}
        processors:
          - log:
              message: "dropping a batch"
    batching:
      period: 1m
      byte_size: 20_000_000
      processors:
        - insert_part:
            index: 0
            content: "timestamp,project,url,installer_name,installer_version,implementation_name,implementation_version,system_name,system_release"
        - archive:
            format: lines
        - bloblang: |
            meta batch_hash = content().hash("xxhash64").encode("hex")

Actual output

{"@timestamp":"2021-12-09T16:50:58Z","@service":"benthos","component":"benthos.input","level":"INFO","message":"Consuming from file '/data/input/pypi/pypi_results_000000000000.jsonl'"}
{"@timestamp":"2021-12-09T16:50:58Z","@service":"benthos","component":"benthos.output","level":"INFO","message":"Dropping messages."}
{"@timestamp":"2021-12-09T16:50:58Z","@service":"benthos","component":"benthos","level":"INFO","message":"Launching a benthos instance, use CTRL+C to close."}
{"@timestamp":"2021-12-09T16:50:58Z","@service":"benthos","component":"benthos","level":"INFO","message":"Listening for HTTP requests at: http://0.0.0.0:4195"}
{"@timestamp":"2021-12-09T16:51:03Z","@service":"benthos","component":"benthos.output.processor.0","level":"INFO","message":"dropping a batch"}
^C{"@timestamp":"2021-12-09T16:51:05Z","@service":"benthos","component":"benthos","level":"INFO","message":"Received SIGTERM, the service is closing."}
{"@timestamp":"2021-12-09T16:51:20Z","@service":"benthos","component":"benthos","level":"INFO","message":"Unable to fully drain buffered messages within target time."}
{"@timestamp":"2021-12-09T16:51:25Z","@service":"benthos","component":"benthos","level":"ERROR","message":"Failed to stop stream gracefully within target time."}

Expected output

(observed when I remove the period policy)

{"@timestamp":"2021-12-09T16:54:50Z","@service":"benthos","component":"benthos.input","level":"INFO","message":"Consuming from file '/data/input/pypi/pypi_results_000000000000.jsonl'"}
{"@timestamp":"2021-12-09T16:54:50Z","@service":"benthos","component":"benthos.output","level":"INFO","message":"Dropping messages."}
{"@timestamp":"2021-12-09T16:54:50Z","@service":"benthos","component":"benthos","level":"INFO","message":"Launching a benthos instance, use CTRL+C to close."}
{"@timestamp":"2021-12-09T16:54:50Z","@service":"benthos","component":"benthos","level":"INFO","message":"Listening for HTTP requests at: http://0.0.0.0:4195"}
{"@timestamp":"2021-12-09T16:54:55Z","@service":"benthos","component":"benthos.output.processor.0","level":"INFO","message":"dropping a batch"}
^C{"@timestamp":"2021-12-09T16:54:56Z","@service":"benthos","component":"benthos","level":"INFO","message":"Received SIGTERM, the service is closing."}
{"@timestamp":"2021-12-09T16:54:56Z","@service":"benthos","component":"benthos.output.processor.0","level":"INFO","message":"dropping a batch"}

Start by reproducing the issue with the provided Benthos configuration, comparing shutdown behavior with and without the batching period policy. Trace the batching and graceful-termination paths, then verify that SIGTERM flushes the pending batch and exits within the target time when the period is configured.