feat(dlq): Added produce policy

[rahul-kumar-saini]

I'm not sure if I did this correctly but eventually our DLQ policies need to produce invalid messages to a dead letter topic.

• Introduced a new Policy which has the sole purpose of producing a given message to a dead letter topic
• This could be extended by other policies such as the count policy to start putting the invalid messages into the dead letter topic instead of just ignoring them
• Modified InvalidMessage Exception to also hold the original topic the message was produced to along with the timestamp the exception was thrown

[evanh]

This looks good to me.

[rahul-kumar-saini]

A few questions I have before this is merged:

• Since the exception can contain multiple messages, should they be produced individually? Currently it would just produce one message to the DLQ containing all of them.
• Is there an easy way to test this policy? I'm not sure how to test that a message is produced to a topic so I haven't written any tests yet

Note:

• I modified the count policy to inherit from this one, it seemed like the logical next step anyway - should be reviewed

[lynnagara]

My understanding was that we'd be having a one-to-one relationship between the original messages and those produced on the dead letter topic. So if 2 of the original messages failed processing, even if they were processed by the strategy together, we'd end up with 2 messages on the dead letter topic - one for each. But this implementation is grouping them together into a single message. Any particular reason for the change (or was my original understanding incorrect)?

[rahul-kumar-saini]

@lynnagara
Yeah that was my question from my comment above your review, I'll change it to produce individually but I wasn't sure what to do.

[lynnagara]

It's a bit unintuitive to me that CountInvalidMessagePolicy sort of flips between being an ignore policy and a produce policy depending on what arguments you pass to it. ProduceInvalidMessagePolicy is a bit like a CountInvalidMessagePolicy with a limit of 0 so I'm kind of unclear on what additional utility it provides.

Some ideas that could help tackle this:

• Make the ignore and produce policies totally separate and making the user explicitly pick which one they want instead of passing a producer to CountInvalidMessagePolicy if they want it to produce. You could pass the limit to those other policies so they both still support the counting behavior.

• Instead of passing the topic and producer to the count policy and making it inherit from ProduceInvalidMessagePolicy, how about passing a function instead, or even another policy class (that then does the producing or ignoring). The count policy would be more generic and could do anything the user wants after the limit is reached (not just ignoring or producing) and might be able to accomodate more custom behavior for different use cases.

What do you think?

[rahul-kumar-saini]

@lynnagara

[...] or even another policy class (that then does the producing or ignoring). The count policy would be more generic and could do anything the user wants after the limit is reached (not just ignoring or producing) and might be able to accomodate more custom behavior for different use cases.

I like this idea of chaining an additional policy to define whatever the user would like to happen with messages staying under the limit, I'll modify the policies to support it.

[fpacifici]

Getting closer, thanks

[fpacifici]

Please fix the to_dict method to avoid turning everything into strings.
Otherwise the rest is ok.

[fpacifici]

Please address the comments Lyn added before merging