Simplify kafka retransmit workflow (#5)

Author: k32

Makefile

@@ -2,7 +2,7 @@ PICS=$(patsubst %.uml,%.png,$(wildcard doc/images/*.uml))
 
 ENV_FILE=.env
 
-MIN_COVERAGE := 80
+MIN_COVERAGE := 75
 BUILD_DIR := $(CURDIR)/_build
 CONCUERROR := $(BUILD_DIR)/Concuerror/bin/concuerror
 CONCUERROR_RUN := $(CONCUERROR) -x snabbkaffe_collector -x snabbkaffe -x snabbkaffe_nemesis -x snabbkaffe_sup \

src/application/kflow_produce_to_kafka.erl

@@ -87,7 +87,7 @@ init(Config = #{topic := Topic}) ->
     }.
 
 %% @private
-map(Offset, Msg = #{partition := P, values := VV}, State) ->
+map(Offset, Msg = #{partition := P, value := Val}, State) ->
   #s{ client = Client
     , topic  = Topic
     } = State,
@@ -97,7 +97,7 @@ map(Offset, Msg = #{partition := P, values := VV}, State) ->
        , partition => P
        , offset    => Offset
        }),
-  case brod:produce_sync_offset(Client, Topic, P, <<>>, VV) of
+  case brod:produce_sync_offset(Client, Topic, P, <<>>, Val) of
     {ok, OutOffset} ->
       prometheus_gauge:set( <<"kflow_kafka_producer_offset">>
                           , [Topic, P]

src/framework/kflow_kafka_commit.erl

@@ -11,6 +11,10 @@
 
 -include("kflow_int.hrl").
 
+-ifndef(TEST).
+-define(TEST, false).
+-endif.
+
 %% kflow_gen callbacks:
 -export([ init/2
         , handle_message/3
@@ -97,7 +101,8 @@ handle_message( Msg = #kflow_msg{ offset                 = Offset
                     , last_committed_offset  => LastCommittedOffset
                     , group_id               => GroupId
                     , route                  => Route
-                    })
+                    }),
+      ?TEST andalso error(offset_tracking_bug)
   end,
   {ok, [Msg], State#{last_committed_offset => SafeOffset}}.
 

src/workflows/kflow_kafka_retransmit.erl

@@ -82,16 +82,9 @@ pipe_spec(Config) ->
   PartFun = maps:get(part_fun, Config, fun partition_by_key/2),
   BufferConfig = maps:with([max_size, max_messages], Config),
   Preprocess ++
-    [ %% First, choose what partition the message should end up in the downstream topic:
-      {map, ?MODULE,
-       {PartFun, NPartitions}}
-      %% Then separate messages by partition:
-    , {demux,
-       fun(_Offset, #{?out_part := P}) -> P end}
-      %% Group messages in chunks:
-    , {aggregate, kflow_group_kafka_messages,
-       BufferConfig}
-      %% And finally push chunks to another topic:
+    [ %% Choose what partition the message should end up in the downstream topic:
+      {map, ?MODULE, {PartFun, NPartitions}}
+      %% Push chunks to another topic:
     , {map, kflow_produce_to_kafka,
        #{ topic  => ToTopic
         , client => ToClient