Merge pull request #99 from kafka4beam/250907-sync-main-1.5

250907 sync main 1.5

Author: zmstone

changelog.md

@@ -1,3 +1,9 @@
+* 4.0.12 (merge 1.5.18)
+  - Partition metadata handling.
+    - Fixed an issue introduced in 4.0.7 (1.5.15) where temporarily missing partitions in the metadata response could leave a `wolff_producer` process permanently disconnected.
+      `wolff_producer` now always attempts to reconnect, even if a partition disappears and reappears.
+    - Ensured that partition numbers in the metadata response are always sequential, even when Kafka returns malformed metadata.
+
 * 4.0.11
   - Upgrade to `kafka_protocol-4.2.8`. Fixed build speed and link issue for crc32c.
 

rebar.config

@@ -30,3 +30,10 @@
                              {tag, "v2.2.0-emqx-1"}}}]}
    ]}]}.
 {shell, [{apps, [wolff]}]}.
+{profiles, [
+  {test, [
+    {deps, [
+      {meck, "1.0.0"}
+    ]}
+  ]}
+]}.

src/wolff_client.erl

@@ -149,8 +149,8 @@ safe_call(Pid, Call) ->
   end.
 
 %% request client to send Pid the leader connection.
-recv_leader_connection(Client, Group, Topic, Partition, Pid, MaxPartitions) ->
-  gen_server:cast(Client, {recv_leader_connection, Group, Topic, Partition, Pid, MaxPartitions}).
+recv_leader_connection(Client, Group, Topic, Partition, Caller, MaxPartitions) ->
+  gen_server:cast(Client, {recv_leader_connection, Group, Topic, Partition, Caller, MaxPartitions}).
 
 release_leader_conns(Client, Group, Topic) ->
     gen_server:cast(Client, {release_leader_conns, Group, Topic}).
@@ -214,15 +214,17 @@ handle_cast({recv_leader_connection, Group, Topic, Partition, Caller, MaxConnect
       %% then get the real EXIT reason in the retry attempts
       St = flush_exit_signals(St1),
       Partitions = do_get_leader_connections(St, Group, Topic),
-      MaybePid = case lists:keyfind(Partition, 1, Partitions) of
-        {_, Pid} ->
-          Pid;
+      _ = case lists:keyfind(Partition, 1, Partitions) of
+        {_, MaybePid} ->
+          erlang:send(Caller, ?leader_connection(MaybePid));
         false ->
-          %% this happens as a race between metadata refresh and partition producer shutdown
+          log_warn("partition_missing_in_metadata_response", #{topic => Topic, partition => Partition}),
+          %% This happens as a race between metadata refresh and partition producer shutdown
           %% partition producer will be shutdown by wolff_producers after metadata refresh is complete
-          partition_missing_in_metadata_response
+          %% Or a malformed metadata response with the last partition missing.
+          Reason = partition_missing_in_metadata_response,
+          erlang:send(Caller, ?leader_connection(?conn_down(Reason)))
       end,
-      _ = erlang:send(Caller, ?leader_connection(MaybePid)),
       {noreply, St};
     {error, Reason} ->
       _ = erlang:send(Caller, ?leader_connection(?conn_down(Reason))),
@@ -617,9 +619,13 @@ do_get_metadata2(Vsn, Connection, Topic, Timeout) ->
       Brokers = [parse_broker_meta(M) || M <- BrokersMeta],
       [TopicMeta] = kpro:find(topics, Meta),
       ErrorCode = kpro:find(error_code, TopicMeta),
-      Partitions = kpro:find(partitions, TopicMeta),
+      Partitions0 = kpro:find(partitions, TopicMeta),
       case ErrorCode =:= ?no_error of
+        true when Partitions0 =:= [] ->
+          %% unsure if this is possible
+          {error, no_partitions_metadata};
         true ->
+          Partitions = fix_partition_metadata(Topic, Partitions0),
           {ok, {Brokers, Partitions}};
         false ->
           {error, ErrorCode} %% no such topic ?
@@ -628,6 +634,36 @@ do_get_metadata2(Vsn, Connection, Topic, Timeout) ->
       {error, Reason}
   end.
 
+%% The partitions count cache only caches the total count but not
+%% a list of partition numbers, so we need to fix the "holes" in
+%% the partition sequence [0, N), if any.
+%% There is no partition count returned in metadata response,
+%% so we must rely on the max partition number to guess the total
+%% number of partitions.
+fix_partition_metadata(Topic, PartitionMetaList) ->
+  Partitions = lists:sort(lists:map(fun(M) -> kpro:find(partition_index, M) end, PartitionMetaList)),
+  Missing = fix_partition_metadata_loop(Partitions, 0, []),
+  case Missing =:= [] of
+    true ->
+      PartitionMetaList;
+    false ->
+      log_warn("partitions_missing_in_metadata_response", #{topic => Topic, partitions => Missing}),
+      PartitionMetaList ++
+      lists:map(fun(P) ->
+        #{partition_index => P,
+          error_code => partition_missing_in_metadata_response
+        }
+      end, Missing)
+  end.
+
+fix_partition_metadata_loop([], _, Missing) ->
+  lists:reverse(Missing);
+fix_partition_metadata_loop([P | Partitions], P, Missing) ->
+  fix_partition_metadata_loop(Partitions, P + 1, Missing);
+fix_partition_metadata_loop([P0 | _] = Partitions, P, Missing) ->
+  true = (P0 > P),
+  fix_partition_metadata_loop(Partitions, P + 1, [P | Missing]).
+
 -spec parse_broker_meta(kpro:struct()) -> {integer(), host()}.
 parse_broker_meta(BrokerMeta) ->
   BrokerId = kpro:find(node_id, BrokerMeta),

src/wolff_producer.erl

@@ -720,12 +720,29 @@ log_connection_down(Topic, Partition, Conn, Reason) ->
            "connection_to_partition_leader_error",
            #{conn => Conn, reason => Reason}).
 
-ensure_delayed_reconnect(#{config := #{reconnect_delay_ms := Delay0} = Config,
-                           client_id := ClientId,
-                           topic := Topic,
-                           partition := Partition,
-                           reconnect_timer := ?no_timer
-                          } = St, DelayStrategy) ->
+is_timer_on(?no_timer) ->
+  false;
+is_timer_on({_, Ref}) when is_reference(Ref) ->
+  %% started by timer:apply_after
+  is_timer_on(Ref);
+is_timer_on(Ref) when is_reference(Ref) ->
+  erlang:read_timer(Ref) =/= false.
+
+ensure_delayed_reconnect(St, DelayStrategy) ->
+  Tref = maps:get(reconnect_timer, St, ?no_timer),
+  case is_timer_on(Tref) of
+    true ->
+      St;
+    false ->
+      do_ensure_delayed_reconnect(St, DelayStrategy)
+  end.
+
+do_ensure_delayed_reconnect(
+  #{config := #{reconnect_delay_ms := Delay0} = Config,
+    client_id := ClientId,
+    topic := Topic,
+    partition := Partition
+   } = St, DelayStrategy) ->
   Attempts = maps:get(reconnect_attempts, St, 0),
   MaxPartitions = maps:get(max_partitions, Config, all_partitions),
   Attempts > 0 andalso Attempts rem 10 =:= 0 andalso
@@ -755,10 +772,7 @@ ensure_delayed_reconnect(#{config := #{reconnect_delay_ms := Delay0} = Config,
       %% call timer:apply_after for both cases, do not use send_after here
       {ok, Tref} = timer:apply_after(Delay, erlang, send, [self(), ?reconnect]),
       St#{reconnect_timer => Tref, reconnect_attempts => Attempts + 1}
-  end;
-ensure_delayed_reconnect(St, _Delay) ->
-  %% timer already started
-  St.
+  end.
 
 evaluate_pending_ack_funs(PendingAcks, [], _BaseOffset) -> PendingAcks;
 evaluate_pending_ack_funs(PendingAcks, [{CallId, BatchSize} | Rest], BaseOffset) ->
@@ -950,7 +964,7 @@ maybe_reply_queued(?SEND_REQ({Caller, Ref}, _, _)) ->
     erlang:send(Caller, {Ref, ?queued}).
 
 eval_ack_cb(?ACK_CB(AckFun, Partition), BaseOffset) when is_function(AckFun, 2) ->
-  ok = AckFun(Partition, BaseOffset); %% backward compatible
+  ok = AckFun(Partition, BaseOffset);
 eval_ack_cb(?ACK_CB({F, A}, Partition), BaseOffset) when is_function(F) ->
   true = is_function(F, length(A) + 2),
   ok = erlang:apply(F, [Partition, BaseOffset | A]);

src/wolff_producers.erl

@@ -685,7 +685,6 @@ start_producer_and_insert_pid(ClientId, Group, Topic, Partition, Config) ->
                                         ?conn_down(to_be_discovered), Config),
   ok = insert_producers(ClientId, Group, Topic, #{Partition => Pid}).
 
-%% Config is not used so far.
 start_partition_refresh_timer(Config) ->
   IntervalSeconds = maps:get(partition_count_refresh_interval_seconds, Config,
                              ?partition_count_refresh_interval_seconds),

test/wolff_supervised_tests.erl

@@ -412,7 +412,109 @@ test_partition_count_decrease() ->
   ok = application:stop(wolff),
   ok.
 
-test_topic_recreate_test_() ->
+%% The last partition is temporarily missing from metadata response.
+%% The producer should retry to recover
+last_partition_missing_in_metadata_response_test_() ->
+  {timeout, 30, %% it takes time to alter topic via cli in docker container
+   fun() -> test_partition_missing_in_metadata_response(2) end}.
+
+%% The partition in the middle is temporarily missing from metadata response.
+%% The producer should retry to recover
+mid_partition_missing_in_metadata_response_test_() ->
+  {timeout, 30, %% it takes time to alter topic via cli in docker container
+   fun() -> test_partition_missing_in_metadata_response(1) end}.
+
+test_partition_missing_in_metadata_response(ThePartition) ->
+  ClientId = <<"test-temporarily-malformed-metadata-response">>,
+  %% ensure the topic does not exist
+  Topic = <<"test-topic-5">>,
+  delete_topic(Topic),
+  Partitions = 3,
+  create_topic(Topic, Partitions),
+  io:format(user, "created topic ~s with ~p partitions\n", [Topic, Partitions]),
+  _ = application:stop(wolff), %% ensure stopped
+  {ok, _} = application:ensure_all_started(wolff),
+  %% always refresh metadata
+  ClientCfg = #{min_metadata_refresh_interval => 0},
+  {ok, ClientPid} = wolff:ensure_supervised_client(ClientId, ?HOSTS, ClientCfg),
+  {ok, Connections} = get_leader_connections(ClientPid, Topic),
+  ?assertEqual(Partitions, length(Connections)),
+  %% always send it to the partition which is going to be missing
+  Partitioner = fun(_, _) -> ThePartition end,
+  ProducerCfg = #{required_acks => all_isr,
+                  partitioner => Partitioner,
+                  reconnect_delay_ms => 0,
+                  %% 0 means no aut-refresh
+                  partition_count_refresh_interval_seconds => 0
+                 },
+  {ok, Producers} = wolff:ensure_supervised_producers(ClientId, Topic, ProducerCfg),
+  ?assertEqual(Partitions, wolff_producers:get_partition_cnt(ClientId, ?NO_GROUP, Topic)),
+  Msg = #{key => ?KEY, value => <<"value">>},
+  ?assertMatch({ThePartition, _}, wolff:send_sync(Producers, [Msg], 3000)),
+  %% mock a bad
+  %% Kill partition leader connection
+  Pid = get_partition_leader_connection(ClientPid, Topic, ThePartition),
+  exit(Pid, kill),
+  %% mock kafka_protcol to return metadata with ThePartition missing
+  meck:new(kpro, [passthrough, no_history]),
+  meck:expect(kpro, request_sync,
+    fun(Connection, Req, Timeout) ->
+      {ok, Rsp} = meck:passthrough([Connection, Req, Timeout]),
+      case Rsp of
+        #kpro_rsp{msg = #{topics := [#{partitions := PM0} = TopicMeta]} = Meta} ->
+          PM = lists:filter(fun(#{partition_index := P}) -> P =/= ThePartition end, PM0),
+          NewRsp = Rsp#kpro_rsp{msg =Meta#{topics := [TopicMeta#{partitions := PM}]}},
+          {ok, NewRsp};
+        _ ->
+          {ok, Rsp}
+      end
+  end),
+  ?assertNot(is_pid(get_partition_leader_connection(ClientPid, Topic, ThePartition))),
+  %% the request will be buffered, but the call times out
+  Msg2 = #{key => ?KEY, value => <<"v">>},
+  ?assertError(timeout, wolff:send_sync(Producers, [Msg2], 100)),
+  %% Wait for auto recover
+  Tester = self(),
+  meck:expect(kpro, send,
+    fun(Conn, Req) ->
+      Tester ! {sent, Conn},
+      meck:passthrough([Conn, Req])
+    end),
+  %% Now remove the injected error (malformed metadata response)
+  %% wolff_producer should now be able to recover
+  meck:expect(kpro, request_sync,
+    fun(Connection, Req, Timeout) ->
+      meck:passthrough([Connection, Req, Timeout])
+    end),
+  receive
+    {sent, NewConn} ->
+      ?assertEqual(NewConn, get_partition_leader_connection(ClientPid, Topic, ThePartition))
+  after
+    2000 ->
+      %% wolff_producer retry delay is 0, but it randomize with extra 0-1000ms
+      error(timeout)
+  end,
+  meck:unload(kpro),
+  ok = fetch_and_match(ClientPid, Topic, ThePartition, 0, [Msg, Msg2]),
+  %% cleanup
+  ok = wolff:stop_and_delete_supervised_producers(Producers),
+  ?assertEqual([], supervisor:which_children(wolff_producers_sup)),
+  ok = wolff:stop_and_delete_supervised_client(ClientId),
+  ?assertEqual([], supervisor:which_children(wolff_client_sup)),
+  ok = application:stop(wolff),
+  ok.
+
+get_partition_leader_connection(Client, Topic, Partition) ->
+  ok = wolff_client:recv_leader_connection(Client, ?NO_GROUP, Topic, Partition, self(), all_partitions),
+  receive
+    {leader_connection, Pid} ->
+      Pid
+  after
+    20000 ->
+      error(timeout)
+  end.
+
+topic_recreate_test_() ->
   {timeout, 30, %% it takes time to alter topic via cli in docker container
    fun test_topic_recreate/0}.