Fix for #21

silviucpp · May 27, 2020 · e220c75 · e220c75
1 parent ab72c61
commit e220c75
Showing 1 changed file with 11 additions and 3 deletions.
diff --git a/src/erlkaf_consumer_group.erl b/src/erlkaf_consumer_group.erl
@@ -90,12 +90,20 @@ handle_info({revoke_partitions, Partitions}, #state{
     ?LOG_INFO("revoke partitions: ~p", [Partitions]),
     Pids = get_pids(ActiveTopicsMap, Partitions),
     ok = stop_consumers(Pids),
+    ?LOG_INFO("all existing consumers stopped for partitions: ~p", [Partitions]),
     ok = erlkaf_nif:consumer_partition_revoke_completed(ClientRef),
     {noreply, State#state{active_topics_map = #{}}};
 
-handle_info({'EXIT', FromPid, Reason} , State) when Reason =/= normal ->
-    ?LOG_WARNING("consumer ~p died with reason: ~p. restart consumer group ...", [FromPid, Reason]),
-    {stop, {error, Reason}, State};
+handle_info({'EXIT', FromPid, Reason}, #state{active_topics_map = ActiveTopics} = State) when Reason =/= normal ->
+
+    case maps:size(ActiveTopics) of
+        0 ->
+            ?LOG_WARNING("consumer ~p died with reason: ~p. no active topic (ignore message) ...", [FromPid, Reason]),
+            {noreply, State};
+        _ ->
+            ?LOG_WARNING("consumer ~p died with reason: ~p. restart consumer group ...", [FromPid, Reason]),
+            {stop, {error, Reason}, State}
+    end;
 
 handle_info(_Info, State) ->
     {noreply, State}.