_group_config = [] ? about kaffe HOT 8 CLOSED

@stephanos It looks like that setting is only relevant if we also used offset_commit_policy: commit_to_kafka_v2. Currently, Kaffe commits offsets after the handle_messages function is returned. I don't think it would be possible to support that option without other changes to Kaffe.

Am I understanding that correctly?

I'm looking at the Brod code here.

from kaffe.

Outside of this specific question, though, we should see which of those properties we could support (assuming I'm reading things right in this case).

from kaffe.

🤔I don't know enough about brod or kaffe to answer that question.

What I found was that in an integration test that sent a message to Kafka and waited for it to be acked in Kafka always took exactly 5s. I could see that the message was acked in our handler code but it was only sent to Kafka every 5s. This matches the default time for offset_commit_interval_seconds exactly, so my naive assumption was that I can just lower that setting to 100ms during testing. But, to my surprise, it didn't do anything. It still took 5s. So - based on what I observed - I think this setting is useful and is being used right now.

Please note that this is happening when using supervisor(Kaffe.GroupMemberSupervisor, []). Based on what I understand from the source code, it would work correctly when using worker(Kaffe.Consumer, []) (since the group_config setting is passed on to brod). But we need the group member supervisor in our code.

from kaffe.

Very interesting! Do you want to try to pass it in and see if you get the results you expect?

Looking at the definition of consumer_group_config here, I do think you're on to something.

I think you'd just need to grab the config like this:

  defp group_config do
    Kaffe.Config.Consumer.configuration.group_config
  end

Having it verified in your test would ensure that it would do the right thing.

from kaffe.

I just tried it: I can see that my test went from ~5s to 1s! The ack is definitely sent faster now.

This is all I changed:

{:ok, pid} = group_coordinator().start_link(subscriber_name, consumer_group,
      [topic], group_config(), __MODULE__, self())

and

defp group_config do
    Kaffe.Config.Consumer.configuration.group_config
 end

as well as

config :kaffe,
  consumer: [
    rebalance_delay_ms: 100,
    offset_commit_interval_seconds: 1,
    endpoints: [localhost: 9092]
  ],
  ...

from kaffe.

@stephanos Excellent! Would you be interested in submitting a PR? I think this is a nice addition.

from kaffe.

I will!

from kaffe.

closed, continued in #70

from kaffe.