From 8d4b5e54ddebc2f614c132ea7725a824ff2ad32e Mon Sep 17 00:00:00 2001 From: urmastalimaa Date: Tue, 25 Jun 2024 17:25:06 +0300 Subject: [PATCH] Add spawn_dedicated_client to v2 group subscriber consumer config When `spawn_dedicated_client` is set to `{true, Bootstrap}`, a separate client is spawned for each partition subscriber, allowing the subscribers themselves to block the fetch loop with setting min_bytes > 0, without affecting one another. --- src/brod_group_subscriber_v2.erl | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/src/brod_group_subscriber_v2.erl b/src/brod_group_subscriber_v2.erl index 1fb84014..cc5592b9 100644 --- a/src/brod_group_subscriber_v2.erl +++ b/src/brod_group_subscriber_v2.erl @@ -503,7 +503,15 @@ maybe_start_worker( _MemberId ) -> {ok, pid()}. start_worker(Client, Topic, MessageType, Partition, ConsumerConfig, StartOptions) -> - {ok, Pid} = brod_topic_subscriber:start_link( Client + Client1 = case proplists:get_value(spawn_dedicated_client, ConsumerConfig, false) of + false -> + Client; + {true, {Endpoints, ClientConfig} = _Bootstrap} -> + ClientId = list_to_atom(atom_to_list(Client) ++ binary_to_list(<<"-group-subscriber-v2-">>) ++ binary_to_list(Topic) ++ integer_to_list(Partition)), + {ok, _Pid} = brod_client:start_link(Endpoints, ClientId, ClientConfig), + ClientId + end, + {ok, Pid} = brod_topic_subscriber:start_link( Client1 , Topic , [Partition] , ConsumerConfig