klarna-incubator / kflow Goto Github PK

-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
               }
   }.

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

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

{{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">>}}'

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

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