Unable to delete message from branch

[mariamjamal94]

I am using benthos for the first time and am miserably stuck.
My problem is that in case an http processor 'errors' (e.g. 500 response), I delete the root in catch mapping i.e. root = deleted()
While unit testing this behavior, I expect the output batches to be empty. However it fails, expecting the output batch to be the same as the input batch.

Processor:

   - label: "rule_engine_request"
     branch:
       request_map: |
         root = {"test1": "test1"}
         root = if meta("validation_error").not_null().catch("") != "" {
           deleted()
         }
       processors:
         - log:
             level: INFO
             message: "Sending request to rule engine: ${!this}"
         - label: "make_rule_engine_http_request"
           http:
             url: http://httpbin.org/status/500
             verb: POST
             headers:
               Authorization: "Bearer ${RULE_ENGINE_TOKEN}"
             successful_on:
               - 200
             drop_on:
               - 404
               - 500
               - 502
               - 503
         - catch:
             - log:
                 message: "Rule engine request failed unexpectedly with code ${! meta(\"http_status_code\")}"
             - mapping: root = deleted()
       result_map: root = this

My unit test is very simple:

  - name: tests error from rule engine request deletes root
    target_processors: 'rule_engine_request'
    mocks:
      make_rule_engine_http_request:
        mapping: throw("some error")
    metadata:
      - "http_status_code": "500"
    input_batch:
      - json_content:
          "test1": "test1"
    output_batches: []

When I run the test, I get:
unexpected batch: [{"test1":"test1"}]

Can someone help me what am I missing or if my understanding of root is completely wrong?
Slighty related: is this even the right way to mock failed http processor with a 500 response?

[mihaitodor]

Hey @mariamjamal94 👋 If you run the tests with --log=debug you'll see some hints. It should show you a message saying that "Child processors failed: child processors resulted in zero messages", which basically means that the number of messages going inside a branch processor needs to be equal to the number of messages coming out of it. Because of this, you can't just delete a message in there. What you can do, for example, is to set meta error = error() and then, in the result_map, you can have root = if metadata("error") != null { throw(metadata("error")) } else { this }. Something like this:

input:
  generate:
    count: 1
    mapping: root = ""
  processors:
    - branch:
        processors:
          - mapping: throw("Kaboom!")

          - catch:
              - mapping: meta error = error()

        result_map: |
          root = if metadata("error") != null { throw(metadata("error")) } else { this }
    - catch:
        - mapping: root = deleted()

tests:
  - name: tests branch processor
    target_processors: "/input/processors"
    input_batch:
      - json_content:
          foo: "bar"

Also, not sure why you have tests[].metadata in there, but you should use tests[].input_batches[][].metadata instead. I was kinda' expecting to get an error since you're using an undocumented field, but maybe it's some leftover.

[mihaitodor]

Hey @maxarndt I’m not sure what’s happening in your case without seeing the config you’re running. In general, the branch processor needs to somehow map each message it outputs onto the origin message. If it produces more or fewer messages, it will error out. You can implement your own custom one if really needed, but, in general, archive and unarchive steps can be moved outside of the branch processor and deletion can be done outside based on metadata fields or on catch processors. Hope that helps.

[maxarndt]

Hey @mihaitodor, I copied the config you posted above to a file and executed it with rpk connect run config.yaml.
Hope this helps you reconstructing what the issue might be. :-)

Here you can see the full output:

❯ rpk connect run config.yaml
INFO Running main config from specified file       @service=redpanda-connect benthos_version=4.38.0 path=config.yaml
INFO Listening for HTTP requests at: http://0.0.0.0:4195  @service=redpanda-connect
INFO Input type generate is now active             @service=redpanda-connect label="" path=root.input
INFO Launching a Redpanda Connect instance, use CTRL+C to close  @service=redpanda-connect
ERRO failed assignment (line 1): Kaboom!           @service=redpanda-connect label="" path=root.input.processors.0.branch.processors.0
ERRO Branch error: result mapping failed: failed assignment (line 1): failed assignment (line 1): Kaboom!  @service=redpanda-connect label="" path=root.input.processors.0
INFO Output type stdout is now active              @service=redpanda-connect label="" path=root.output
INFO Pipeline has terminated. Shutting down the service  @service=redpanda-connect

[mihaitodor]

@maxarndt I had a look and the errors you're seeing there are expected. One is generated by the mapping processor which contains an explicit throw("Kaboom!") and the other one is coming from throw(metadata("error")) inside the result_map of the branch processor. This makes the entire branch processor fail, so the catch processor detects the error flag on the message and runs the mapping processor which does root = deleted(). Hope this helps.

[maxarndt]

@mihaitodor, it's good to hear that this is expected behavior. To be honest, I would not expect errors to be printed out that way when they are caught.
When I check the logs of Benthos, I feel uncomfortable if I see errors. However, if they are caught, I should not be concerned about them.

[mihaitodor]

That is part of a wider discussion about having more fine-grained control over logs which I expect to be addressed at some point (this one should definitely be fixed ASAP). However, the use of exceptions for control flow is an anti-pattern (see here and here). In general, I'd like to see an error in the logs when some operation fails due to IO or when a logical bug is encountered, but I wouldn't use errors for normal flow and I'd design my pipelines in such a way that errors are handled as soon as possible (see the try, catch, switch and retry processors for example) and can't happen frequently. The throw() function is handy for simulating errors, but, otherwise, I don't really use it. Having a metrics-based observability stack in place is crucial to ensure that a failing pipeline is detected in realtime and the appropriate actions are taken (redeployment, alerting etc). Error logs are mostly useful for troubleshooting, like when replaying some events which haven't been processed correctly.