DRAFT: GUILD-625: Draft: Deimos changes required to support 1 app listener with multiple handlers #119
Conversation
…ith multiple handlers This is a proof of concept work for now. Outstanding work: - Formalize the top level configs and add decent defaults - Add support for the Consume::BatchConsumption - Add support for the AR stuff - Add tests - Clean up the code
eduardopoleoflipp
changed the title
| # placeholders names / values for now but we'll probably would want to move them | ||
| # to the Deimos.config object | ||
| max_bytes_per_partition = 524_288 | ||
| delivery = 'message' |
There was a problem hiding this comment.
Loosing granularity on this particular attribute is a bit of a bummer.
There was a problem hiding this comment.
Actually, looking at RubyKafka, and the Phobos PR, there's no reason we can't specify separate max_bytes_per_partition for each topic. Since we're using different subscribe calls, and each call takes that as an option, we can at least do it that way.
More important would be the ability to separate out batch from message processing on the Phobos side, because I think it is important to allow each topic to have a separate config for that. I think we need to still enable that in the Phobos config, which means a larger change than that PR has right now.
There was a problem hiding this comment.
ok, yes this will require that we provide a way to do per topic configuration in phobos. How about supporting something like this:
listeners:
- handler: DeimosHandler
topics:
topic1:
max_bytes_per_partition: 32400
delivery: batch
topic2:
max_bytes_per_partition: 524288
delivery: message
topic:
group_id: test
max_bytes_per_partition: 32400
delivery: message
start_from_beginning: trueIn the implementation we will then merge the specific topic configuration over the top level (default ones) to override them if they're defined. How does that sounds?
| # placeholders names / values for now but we'll probably would want to move them | ||
| # to the Deimos.config object | ||
| max_bytes_per_partition = 524_288 | ||
| delivery = 'message' |
There was a problem hiding this comment.
Actually, looking at RubyKafka, and the Phobos PR, there's no reason we can't specify separate max_bytes_per_partition for each topic. Since we're using different subscribe calls, and each call takes that as an option, we can at least do it that way.
More important would be the ability to separate out batch from message processing on the Phobos side, because I think it is important to allow each topic to have a separate config for that. I think we need to still enable that in the Phobos config, which means a larger change than that PR has right now.
| delivery = 'message' | ||
| backoff = { "min_ms"=> 500, "max_ms"=>10000 } | ||
| group_id = 'DeimosGroup' | ||
| handler = 'Deimos::Handler' |
There was a problem hiding this comment.
These 3 lines should probably be configurable somewhere.
| backoff = { "min_ms"=> 500, "max_ms"=>10000 } | ||
| group_id = 'DeimosGroup' | ||
| handler = 'Deimos::Handler' | ||
| max_concurrency = 1 # very relevant now since we'll probably need to scale this up |
There was a problem hiding this comment.
Wouldn't this be equal to the number of topics?
| end | ||
|
|
||
| consumer_object.class_name.constantize | ||
| end |
There was a problem hiding this comment.
So rather than putting this in the consumer class, I think this belongs in the handler class. I see the handler class as being a very simple pass-through that just sorts the messages by topic and passes the messages or batches (depending on configuration) onto the appropriate consumer class. Consumer class shouldn't need to change at all in this PR.
| # `consume_batch` -> use `delivery :inline_batch` | ||
| class Handler | ||
| include Consume::MessageConsumption | ||
| include Consume::BatchConsumption |
There was a problem hiding this comment.
I see this class as being much more lightweight - it should be a simple Phobos listener which calls the consumer class with the messages it received. The tracing etc. should still happen on the consumer side because we want the metrics to be tagged by topic.
This is a proof of concept work for now. Outstanding work: