Benthos schedule input does not exhaust non-batch inputs

[sagikazarmark]

Preflight Checklist

• I have searched the issue tracker for an issue that matches the one I want to file, without success.
• I am not looking for support or already pursued the available support channels without success.

OpenMeter Version

v1.0.0-beta.53

Expected Behavior

When wrapping a non-batch input in schedule input, only one document is being read from the input instead of reading all.

Actual Behavior

All documents should be read

Steps To Reproduce

input:
  schedule:
    interval: "10s"
    input:
      sql_select:
        driver: postgres
        dsn: postgres://postgres:postgres@service_a:5432/service_a?sslmode=disable
        table: usage
        columns:
          - id
          - realm_id
          - subject
          - time
        where: time >= ?

        # Interval is 10 seconds, but we go back 15 seconds to make sure no events are missed.
        # Deduplication should take care of duplicate events
        # Tweak this based on your needs (ie. your priority of duplicated event ingestion vs potentially missed events)
        args_mapping: 'root = [ (now().ts_unix() - 15).ts_format(format: "2006-01-02 15:04:05", tz: "UTC") ]'

Additional Information

Workaround: use generate input with sql_select processor and unarchive processor.

Also see redpanda-data/connect#2314