how to shut down benthos after consuming a csv

[ekallevig]

When I use the csv input and the stdout ouput, the csv will be consumed and then benthos will shut down. But when I use the elasticsearch output, benthos continues running after the csv has been consumed. Is it possible to have benthos shut down after completing the csv?

[Jeffail]

Hey @ekallevig, are there any other differences in the config that doesn't shut down? I'd expect a pipeline with an elasticsearch output to close gracefully like others would so there might be another component preventing shutdown, or possibly a bug somewhere.

[ekallevig]

No they're otherwise identical. I swap the output from stdout to elasticsearch and it doesn't shut down. Messages get sent successfully to sinks in both cases, just inconsistent shutdown behavior.

[ekallevig]

Here's the config where it doesn't shut down:

input:
  csv:
    paths:
      - test.csv
    batch_count: 350  
pipeline:
  threads: 1
  processors:
  - bloblang: |
      first_name = match first_name { "" => null, _ => this }
      last_name = match last_name { "" => null, _ => this }
      slug = slug
output:
  elasticsearch:
    urls:
      - http://localhost:9200
    index: myindex
    id: ${!json("slug")}
    type: _doc
    max_in_flight: 2
    sniff: false
    batching:
      count: 350
    timeout: 5s
logger:
  level: ALL
  format: json
  add_timestamp: true
  static_fields:
    '@service': benthos
metrics:
  http_server:
    prefix: benthos
    path_mapping: ""
tracer:
  none: {}
shutdown_timeout: 20s

vs the one where it DOES shut down after completion:

input:
  csv:
    paths:
      - test.csv
    batch_count: 350  
pipeline:
  threads: 1
  processors:
  - bloblang: |
      first_name = match first_name { "" => null, _ => this }
      last_name = match last_name { "" => null, _ => this }
      slug = slug
output:
  stdout:
    codec: lines
logger:
  level: ALL
  format: json
  add_timestamp: true
  static_fields:
    '@service': benthos
metrics:
  http_server:
    prefix: benthos
    path_mapping: ""
tracer:
  none: {}
shutdown_timeout: 20s

[ekallevig]

When I force it to shut down, it says Waiting for pending acks to resolve before shutting down. so maybe the csv input isnt properly acking a successful message and thus it stays open waiting for that?

[Jeffail]

Ah I see, it's likely because of the batching policy on the output as the final batched doesn't get filled, benthos will try and make a best attempt to resolve it. You can probably remove the batching field from the output entirely as you have it at the input level, but you could also add a period field and that why the final batch gets flushed with <350 messages:

output:
  elasticsearch:
    urls:
      - http://localhost:9200
    index: myindex
    id: ${!json("slug")}
    type: _doc
    max_in_flight: 2
    sniff: false
    batching:
      count: 350
      period: 1s
    timeout: 5s

Which version of benthos are you running right now? In cases like these where the input is exhausted we're meant to flush the batching mechanisms for a graceful shutdown. It should be fixed in the upcoming v4 release, but I thought we'd already fixed it for v3, would be interesting to take a look.

[ekallevig]

That worked, thanks! I'm using Version: 3.63.0