Sending specific exceptions only to DLQ via new failure-strategy

[millerjp]

I am building a streaming platform and I have certain use cases where certain errors need to go to DLQ but others I want to fail on. For example, custom application exception is thrown, send it to DLQ otherwise fail. The current DLQ failure strategy is all or nothing which is not appropriate for this type of handling.

I implemented a new failure strategy to handle this and would like to add it back to the project as I think it can be useful for others to use. It's not complex - basically takes in a list of exceptions and if the reason is one of them, sends it to DLQ otherwise fails.

eg

        failure-strategy: exception-dead-letter-queue
        dead-letter-queue:
          topic: "my.dlq.topic"
          value:
            serializer: org.apache.kafka.common.serialization.BytesSerializer
          exception-classes: "my.specific.Exception,my.specific.AnotherException"

Would be happy to contribute it back.

[cescoffier]

I like that idea.
My only concern is that our failure strategies are not composable, but I think it's fine for now.

@ozangunalp WDYT?

[ozangunalp]

I think this deserves to be integrated into the existing dead-letter-queue strategy. The only fallback strategy that makes sense is fail-stop, as ignore can be implemented with standard error handling.

If we go down the composition path we can imagine a separate failure strategy that'll instantiate others and filter failure reasons according to it's configuration

[millerjp]

@ozangunalp So would the change be if a list of exceptions is provided to the existing DLQ strategy then it uses them (with fail-stop if not in the list) or if no list of exceptions is provided it follows the current behaviour?

In terms of fallback strategy - could this not be passed in also? (although might be too much for this small change) eg

        failure-strategy: dead-letter-queue
        dead-letter-queue:
          topic: "my.dlq.topic"
          value:
            serializer: org.apache.kafka.common.serialization.BytesSerializer
          exception-classes: "my.specific.Exception,my.specific.AnotherException"`
          fallback-strategy: fail

Happy to make the updates to support it either way?