Add support of per-topic configuration

[romul]

Closes #130 & #14

Configuration format now looks like:

config :kaffe,
  consumers: %{
    "subscriber_1" => [
      endpoints: [kafka: 9092],
      topics: ["interesting-topic"],
      consumer_group: "your-app-consumer-group",
      message_handler: MessageHandler,
      offset_reset_policy: :reset_to_latest,
      max_bytes: 100_000,
      min_bytes: 10_000,
      max_wait_time: 1_000,
      worker_allocation_strategy: :worker_per_topic_partition
    ],
    "subscriber_2" => [
      endpoints: [kafka: 9092],
      topics: ["topic-2"],
      consumer_group: "your-app-consumer-group",
      message_handler: AnotherMessageHandler,
      offset_reset_policy: :reset_to_latest,
      max_bytes: 50_000,
      worker_allocation_strategy: :worker_per_topic_partition
    ]
}

[saneery]

@fatcatt316 can you look at this? very useful changes 🙏

[fatcatt316]

@saneery Thanks for letting us know. We'll aim to check this out before too long.

[apfranzen]

Hi @romul, really appreciate your contributions here. It would be useful to allow configuration per Consumer to be more granular. Things are looking good and mix test passes beautifully. e2e is not passing for me, so I was hoping we could look into that piece a bit? I left some starter adjustments that I am hoping will be helpful. Wanting to get this PR finished up and merged!

[apfranzen]

This test isn't passing for me, is it passing for you? It is passing on master, so I dug in a bit to try to get it shored up with the new Consumer config structure. I didn't quite get the test to succeed, but I think there was some setup left in limbo from the previous Consumer config structure. Here's some adjustments in setup that seem to move it in the right direction, but curious what you think?

    # Retrieve the consumers map
    consumers = Application.get_env(:kaffe, :consumers)

    # Get the first key and its associated configuration
    first_key = consumers |> Map.keys() |> List.first()
    consumer_config = Map.get(consumers, first_key)

    # Set up the first consumer
    Application.put_env(:kaffe, :consumers, %{"s1" => consumer_config})
    {:ok, _pid} = Kaffe.GroupMemberSupervisor.start_link("s1")

    # Set up the second consumer
    Application.put_env(:kaffe, :consumers, %{"s2" => consumer_config})
    {:ok, _pid} = Kaffe.GroupMemberSupervisor.start_link("s2")

[romul]

Yeah, looks like I missed to run e2e tests. This one is definitely incorrect atm.

I guess the code should be:

    # Retrieve the consumers map
    consumers = Application.get_env(:kaffe, :consumers)

    # Get the first key and its associated configuration
    subscriber_key = consumers |> Map.keys() |> List.first()
    consumer_config = Map.get(consumers, subscriber_key)
    Application.put_env(:kaffe, :consumers, %{"s1" => consumer_config, "s2" => consumer_config})

    # Set up the first consumer
    {:ok, _pid} = Kaffe.GroupMemberSupervisor.start_link("s1")

    # Set up the second consumer
    {:ok, _pid} = Kaffe.GroupMemberSupervisor.start_link("s2")

    Process.sleep(consumer_config[:rebalance_delay_ms] + 100)

I'll check it tomorrow

[romul]

@apfranzen It was a tricky one. This test was failed, b/c signature of Subscriber.subscribe was changed and therefore TestSubscriber didn't work. Fixed here: romul@0ea9639

[apfranzen]

Some very old 😅 packages got updated. Know that kafka_protocol was very out of date, which was depending on snappyer. No issues with the updates, just for my understanding, could you please comment quickly on the rationale/process for which packages got updated?

[romul]

@apfranzen I'll take a look at tests & deps and do rebase on master.
Definitely a lot of water has flowed under the bridge since 2022
Btw, at my company we use our fork with this change and it works like a charm)

[romul]

@apfranzen The only rationale was just holding deps more up to date. I agree that this change could be removed from this PR. But I suppose update is wanted by package users, even to the more recent versions than they were at 2022. For example, I updated my company fork to brod 4.0 already.

Do you want me to remove mix.lock change from this PR?

[apfranzen]

@romul No, keeping those updates in this change makes sense, especially with the knowledge that these versions have been stable in your usage. Thanks!

[romul]

@apfranzen I made rebase on the latest master. All tests pass. So now it can be merged safely ;-)

[apfranzen]

@romul this has been merged to master! 🎉 Thank you for your contributions and improving Kaffe!