Should kflow handle adding partition to a topic?

[NAR]

Hello!

I setup a minimal kflow server using this kflow_config module:

-module(kflow_config).

-export([pipes/0]).

pipes() ->
  [bar()].

bar() ->
  PipeSpec = [{map, fun(_Offset, Msg) ->
                        io:format("~p ~p:~p bar Msg: '~p'~n", [calendar:local_time(), ?MODULE, ?LINE, Msg]),
                        {ok, Msg}
                    end},
              {map, fun(_Offset, Msg2) ->
                        io:format("~p ~p:~p Msg2: '~p'~n", [calendar:local_time(), ?MODULE, ?LINE, Msg2]),
                        {ok, Msg2}
                    end}
             ],
  #{ start => {kflow_kafka_consumer_v2, start_link}
   , args  => #{ group_id  => <<"foo_group">>
               , topics    => [<<"foobar">>]
               , id        => ?FUNCTION_NAME
               , pipe_spec => PipeSpec
               , auto_commit => false
               }
   }.

First I create a topic with two partitions:

kafka-topics --create --topic foobar --partitions 2 -replication-factor 1  --bootstrap-server localhost:9091

Then produce some messages using brod:

brod:produce_sync(brod_client_1, <<"foobar">>, 1, <<"">>, <<"test1 from brod">>).
brod:produce_sync(brod_client_1, <<"foobar">>, 0, <<"">>, <<"test0 from brod">>).

The messages are printed on the kflow side:

{{2021,1,20},{18,5,50}} kflow_config:44 bar Msg: '#{headers => [],key => <<>>,
                                                    partition => 1,
                                                    value =>
                                                        <<"test1 from brod">>}'
{{2021,1,20},{18,5,50}} kflow_config:50 Msg2: '{ok,
                                                #{headers => [],key => <<>>,
                                                  partition => 1,
                                                  value =>
                                                   <<"test1 from brod">>}}'
{{2021,1,20},{18,5,59}} kflow_config:44 bar Msg: '#{headers => [],key => <<>>,
                                                    partition => 0,
                                                    value =>
                                                        <<"test0 from brod">>}'
{{2021,1,20},{18,5,59}} kflow_config:50 Msg2: '{ok,
                                                #{headers => [],key => <<>>,
                                                  partition => 0,
                                                  value =>
                                                   <<"test0 from brod">>}}'

Everything works great. Then I add a new partition to the topic:

kafka-topics --alter --topic foobar --partitions 3 --bootstrap-server localhost:9091

and try to send a message into this new partition:

brod:produce_sync(brod_client_1, <<"foobar">>, 2, <<"what is it">>, <<"test2 from brod">>).

Unfortunately this message is not received on the kflow side. Is it supposed to work at all or am I trying something that's not supported (or even doesn't make sense)?

[k32]

Hi!

This reminds me of this issue: kafka4beam/brod#349 (Kflow is based on brod, so unfortunately it inherits this problem). Have you tried restarting the workflow or the whole kflow app? It should trigger consumer group rebalance, and messages from the new partition should be consumed.

[NAR]

Hello!

I think this is exactly the problem, after restarting kflow I get the messages from the newly added partitions. Thanks!

[k32]

P.S. Fixing the linked issue in brod will automatically fix this issue as well, since kflow partition workers are supervised by brod.