Merge pull request #98 from zmstone/280829-handle-partial-metata-response

250829 handle partial metata response

Author: zmstone

.github/workflows/ci.yml

@@ -12,7 +12,6 @@ jobs:
       matrix:
         otp:
           - '25.3'
-          - '24.3'
         kafka:
           - '2.4'
           - '1.1'

changelog.md

@@ -1,3 +1,9 @@
+* 1.5.18
+  - Partition metadata handling.
+    - Fixed an issue introduced in 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.
+
 * 1.5.17
   - Avoid `badfun` error when release hot upgrade.
 * 1.5.16

rebar.config

@@ -20,3 +20,10 @@
                deprecated_functions]}.
 
 {shell, [{apps, [wolff]}]}.
+{profiles, [
+  {test, [
+    {deps, [
+      {meck, "1.0.0"}
+    ]}
+  ]}
+]}.

src/wolff.app.src

@@ -1,6 +1,6 @@
 {application, wolff,
  [{description, "Kafka's publisher"},
-  {vsn, "1.5.17"},
+  {vsn, "1.5.18"},
   {registered, []},
   {applications,
    [kernel,

src/wolff.appup.src

@@ -1,10 +1,7 @@
 %% -*- mode: erlang -*-
-{"1.5.17",
+{"1.5.18",
   [
-    {"1.5.16",
-     [ {load_module, wolff_producer, brutal_purge, soft_purge, []}
-     ]},
-    {<<"1\\.5\\.1[1-5]">>,
+    {<<"1\\.5\\.1[1-7]">>,
      [ {load_module, wolff_client, brutal_purge, soft_purge, []}
      , {load_module, wolff_producers, brutal_purge, soft_purge, []}
      , {load_module, wolff_producer, brutal_purge, soft_purge, []}
@@ -42,10 +39,7 @@
     }
   ],
   [
-    {"1.5.16",
-     [ {load_module, wolff_producer, brutal_purge, soft_purge, []}
-     ]},
-    {<<"1\\.5\\.1[1-3]">>,
+    {<<"1\\.5\\.1[1-7]">>,
      [ {load_module, wolff_client, brutal_purge, soft_purge, []}
      , {load_module, wolff_producers, brutal_purge, soft_purge, []}
      , {load_module, wolff_producer, brutal_purge, soft_purge, []}

src/wolff_client.erl

@@ -113,8 +113,8 @@ safe_call(Pid, Call) ->
   end.
 
 %% request client to send Pid the leader connection.
-recv_leader_connection(Client, Topic, Partition, Pid) ->
-  gen_server:cast(Client, {recv_leader_connection, Topic, Partition, Pid}).
+recv_leader_connection(Client, Topic, Partition, Caller) ->
+  gen_server:cast(Client, {recv_leader_connection, Topic, Partition, Caller}).
 
 delete_producers_metadata(Client, Topic) ->
     gen_server:cast(Client, {delete_producers_metadata, Topic}).
@@ -169,15 +169,17 @@ handle_cast({recv_leader_connection, Topic, Partition, Caller}, St0) ->
     {ok, St1} ->
       St = flush_exit_signals(St1),
       Partitions = do_get_leader_connections(St, 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 ~s-~p", [Topic, 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))),
@@ -464,9 +466,13 @@ do_get_metadata2(Vsn, Connection, Topic) ->
       Brokers = [parse_broker_meta(M) || M <- BrokersMeta],
       [TopicMeta] = kpro:find(topic_metadata, Meta),
       ErrorCode = kpro:find(error_code, TopicMeta),
-      Partitions = kpro:find(partition_metadata, TopicMeta),
+      Partitions0 = kpro:find(partition_metadata, 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 ?
@@ -475,6 +481,36 @@ do_get_metadata2(Vsn, Connection, Topic) ->
       {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, M) end, PartitionMetaList)),
+  Missing = fix_partition_metadata_loop(Partitions, 0, []),
+  case Missing =:= [] of
+    true ->
+      PartitionMetaList;
+    false ->
+      log_warn("partitions_missing_in_metadata_response ~s: ~w", [Topic, Missing]),
+      PartitionMetaList ++
+      lists:map(fun(P) ->
+        #{partition => 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

@@ -591,13 +591,30 @@ log_connection_down(Topic, Partition, _, noproc) ->
 log_connection_down(Topic, Partition, _, Reason) ->
   log_info(Topic, Partition, "connection to partition leader is down~nreason=~p", [Reason]).
 
-ensure_delayed_reconnect(#{config := #{reconnect_delay_ms := Delay0},
-                           client_id := ClientId,
-                           topic := Topic,
-                           partition := Partition,
-                           reconnect_timer := ?no_timer,
-                           conn := Conn
-                          } = 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},
+    client_id := ClientId,
+    topic := Topic,
+    partition := Partition,
+    conn := Conn
+   } = St, DelayStrategy) ->
   Attempts = maps:get(reconnect_attempts, St, 0),
   Attempts > 0 andalso Attempts rem 10 =:= 0 andalso
     log_error(Topic, Partition, "still disconnected after ~p reconnect attempts, conn=~p~n", [Attempts, Conn]),
@@ -621,10 +638,7 @@ ensure_delayed_reconnect(#{config := #{reconnect_delay_ms := Delay0},
       %% 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, [{?no_caller_ack, BatchSize} | Rest], BaseOffset) ->

src/wolff_producers.erl

@@ -351,7 +351,6 @@ start_producer_and_insert_pid(Ets, ClientId, Topic, Partition, Config) ->
   ets:insert(Ets, {Partition, Pid}),
   ok.
 
-%% 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

@@ -210,7 +210,7 @@ partition_count_decrease_test_() ->
 
 test_partition_count_decrease() ->
   io:format(user, "test_partition_count_decrease\n", []),
-  ClientId = <<"test-add-more-partitions">>,
+  ClientId = <<"test-recreate-topic-with-fewer-partitions">>,
   %% ensure the topic does not exist
   Topic = <<"test-topic-4">>,
   delete_topic(Topic),
@@ -273,7 +273,112 @@ 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) ->
+  io:format(user, "test_partition_missing_in_metadata_response\n", []),
+  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} = wolff_client: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,
+  Name = ?FUNCTION_NAME,
+  ProducerCfg = #{required_acks => all_isr,
+                  partitioner => Partitioner,
+                  reconnect_delay_ms => 0,
+                  name => Name,
+                  %% 0 means no aut-refresh
+                  partition_count_refresh_interval_seconds => 0
+                 },
+  {ok, Producers} = wolff:ensure_supervised_producers(ClientId, Topic, ProducerCfg),
+  ?assertEqual(Partitions, length(ets:tab2list(Name))),
+  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 = #{topic_metadata := [#{partition_metadata := PM0} = TopicMeta]} = Meta} ->
+          PM = lists:filter(fun(#{partition := P}) -> P =/= ThePartition end, PM0),
+          NewRsp = Rsp#kpro_rsp{msg =Meta#{topic_metadata := [TopicMeta#{partition_metadata := 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(ClientId, Topic, Name),
+  ?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, Topic, Partition, self()),
+  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}.