Can't seem to get output retry working

[nickzelei]

Hi all,
Am exploring the retry plugin and I can't seem to get it to actually work.
Trying to understand if I'm doing something wrong, or if this is actually a bug.

I'm syncing data to a SQL database and want it to fail if there is something wrong (i.e. bad connection, schema mapping, data, etc). Right now, the sql_insert will retry forever. This led me to try out the retry, with no success.

Here is a sample config I whipped up for easy repro. In a production scenario. I'm using postgres, but the same issue applies here with sqlite.

input:
  label: ""
  generate:
    count: 1
    interval: ""
    mapping: |
      root = {
        "id": uuid_v4(),
        "created_at": now(),
        "updated_at": now(),
      }
output:
  label: ""
  retry:
    max_retries: 1
    output:
      sql_insert:
        driver: sqlite
        dsn: ./foo.db
        table: users
        columns: ['id', 'created_at', 'updated_at']
        args_mapping: root = [this.id, this.created_at, this.updated_at]
        # init_statement: |
        # CREATE TABLE IF NOT EXISTS users (id varchar NOT NULL, created_at varchar NOT NULL, updated_at varchar NOT NULL);

I've intentionally commented out the init_statement in the output so that it will fail. The output succeeds just fine with that uncommented of course, but I'm trying to get it to fail to showcase my issue. :)

The output when running benthos with this config:

benthos -c benthos.yaml
INFO Running main config from specified file       @service=benthos path=benthos.yaml version=4.22.0
INFO Listening for HTTP requests at: http://0.0.0.0:4195  @service=benthos
INFO Launching a benthos instance, use CTRL+C to close  @service=benthos
ERRO Failed to send message to sql_insert: SQL logic error: no such table: users (1)  @service=benthos label="" path=root.output
WARN Failed to send message: SQL logic error: no such table: users (1)  @service=benthos label="" path=root.output
ERRO Failed to send message to sql_insert: SQL logic error: no such table: users (1)  @service=benthos label="" path=root.output
ERRO Failed to send message: SQL logic error: no such table: users (1)  @service=benthos label="" path=root.output
ERRO Failed to send message to sql_insert: SQL logic error: no such table: users (1)  @service=benthos label="" path=root.output
WARN Failed to send message: SQL logic error: no such table: users (1)  @service=benthos label="" path=root.output
ERRO Failed to send message to sql_insert: SQL logic error: no such table: users (1)  @service=benthos label="" path=root.output
ERRO Failed to send message: SQL logic error: no such table: users (1)  @service=benthos label="" path=root.output
ERRO Failed to send message to sql_insert: SQL logic error: no such table: users (1)  @service=benthos label="" path=root.output
WARN Failed to send message: SQL logic error: no such table: users (1)  @service=benthos label="" path=root.output
ERRO Failed to send message to sql_insert: SQL logic error: no such table: users (1)  @service=benthos label="" path=root.output
ERRO Failed to send message: SQL logic error: no such table: users (1)  @service=benthos label="" path=root.output
ERRO Failed to send message to sql_insert: SQL logic error: no such table: users (1)  @service=benthos label="" path=root.output
WARN Failed to send message: SQL logic error: no such table: users (1)  @service=benthos label="" path=root.output
ERRO Failed to send message to sql_insert: SQL logic error: no such table: users (1)  @service=benthos label="" path=root.output
ERRO Failed to send message: SQL logic error: no such table: users (1)  @service=benthos label="" path=root.output
ERRO Failed to send message to sql_insert: SQL logic error: no such table: users (1)  @service=benthos label="" path=root.output
WARN Failed to send message: SQL logic error: no such table: users (1)  @service=benthos label="" path=root.output
ERRO Failed to send message to sql_insert: SQL logic error: no such table: users (1)  @service=benthos label="" path=root.output
ERRO Failed to send message: SQL logic error: no such table: users (1)  @service=benthos label="" path=root.output
ERRO Failed to send message to sql_insert: SQL logic error: no such table: users (1)  @service=benthos label="" path=root.output
WARN Failed to send message: SQL logic error: no such table: users (1)  @service=benthos label="" path=root.output
ERRO Failed to send message to sql_insert: SQL logic error: no such table: users (1)  @service=benthos label="" path=root.output
ERRO Failed to send message: SQL logic error: no such table: users (1)  @service=benthos label="" path=root.output
ERRO Failed to send message to sql_insert: SQL logic error: no such table: users (1)  @service=benthos label="" path=root.output
WARN Failed to send message: SQL logic error: no such table: users (1)  @service=benthos label="" path=root.output
ERRO Failed to send message to sql_insert: SQL logic error: no such table: users (1)  @service=benthos label="" path=root.output
ERRO Failed to send message: SQL logic error: no such table: users (1)  @service=benthos label="" path=root.output
ERRO Failed to send message to sql_insert: SQL logic error: no such table: users (1)  @service=benthos label="" path=root.output
WARN Failed to send message: SQL logic error: no such table: users (1)  @service=benthos label="" path=root.output
ERRO Failed to send message to sql_insert: SQL logic error: no such table: users (1)  @service=benthos label="" path=root.output
ERRO Failed to send message: SQL logic error: no such table: users (1)  @service=benthos label="" path=root.output

This continues forever until killing the process, where-as I would expect it to fail after the first try since I have the max_retries set to 1.

[Jeffail]

Hey @nickzelei you need to specify an alternative output to send the failed data or it'll be retried indefinitely. To do that stick your sql output inside a https://www.benthos.dev/docs/components/outputs/fallback, and then add an alternative output after it. For example if you want to just drop the data add the https://www.benthos.dev/docs/components/outputs/drop output as the second one.

[nickzelei]

Hi @Jeffail - thanks for the quick reply! I appreciate it.

Your suggestion works, but then the benthos process returns with a successful exit code instead of not. Was moreso hoping for a way for bethos to fail in the case where something went wrong. If it is configured to drop all of the data, then from the sync standpoint, it was a success.

To give a little more context, I am syncing to multiple data sources in a broker fan-out fashion and was hoping to have benthos fail if one of the output streams has an issue or isn't syncable in some fashion.

EDIT: to give a little more color here as well, I'm invoking benthos within a go process like this:

	streambldr := service.NewStreamBuilder()

	err := streambldr.SetYAML(req.BenthosConfig)
	if err != nil {
		return nil, fmt.Errorf("unable to convert benthos config to yaml for stream builder: %w", err)
	}
	stream, err := streambldr.Build()
	if err != nil {
		return nil, fmt.Errorf("unable to build benthos stream: %w", err)
	}

	err = stream.Run(ctx)
	if err != nil {
		return nil, fmt.Errorf("unable to run benthos stream: %w", err)
	}

My initial thought here was that if something with the stream was wrong, the stream.Run would return an error and I could key off of that.

[Jeffail]

Ah okay, not in this case, .Run would only return an error if the service is told to stop and can't within a certain amount of time, as this'd indicate that resources aren't fully cleaned up. In the sense of delivery guarantees within the stream processor itself those are all made explicit in your config, and when you run generate into something that might fail, and it fails, the stream processor will only ever continue to attempt to deliver the data. You can choose to drop the data if you aren't interested in delivery guarantees, or ideally you'd redirect the data to some form of dead-letter queue, but it needs to be made explicit otherwise when the input source doesnt support nacks itself you'll get endless retries.

However, when you're invoking the stream processor programmatically you can do things slightly differently and insert data instead via https://pkg.go.dev/github.com/benthosdev/benthos/[email protected]/public/service#StreamBuilder.AddProducerFunc, which returns an error if the message is nacked (immediately ending the retry loop).

It'd mean you need to implement the generate logic yourself but in your case it's just a single message so not difficult. If you find yourself wanting to use other benthos inputs but still access and intercept output nacks programmatically you could be cheeky and have two stream builders, one where you have your input config and use https://pkg.go.dev/github.com/benthosdev/benthos/[email protected]/public/service#StreamBuilder.AddConsumerFunc to extract data from the input and then a second one where you use https://pkg.go.dev/github.com/benthosdev/benthos/[email protected]/public/service#StreamBuilder.AddProducerFunc to inject the data and implement your own custom nack behaviour.

[nickzelei]

Amazing, thank you for the detailed response.

Yea in the real world my inputs are coming from one sql database table and am trying to synchronize it with another. The problem I was hitting though is that I would cancel the context of Run (due to some SQL error), but because the sql_input is retrying endlessly, that keeps going and going until I actually restart the overall process. Peering into the code, I think that's because a lot of the processors use an internal context.Background() so aren't a part of the Run ctx itself (at least with my cursory peering, but it seems to match up with the behavior.)

I was separately looking into the metrics benthos emits and was wondering if I could just surface those metrics to know after the fact if any of the inputs failed. That would work too I think.

Otherwise, I might get cheeky and implement my own Consumer/Producer funcs to get the logic I am after.

Thanks again for the help and the links!