feat: make allow_auto_topic_creation configurable

Author: zmstone

changelog.md

@@ -2,6 +2,7 @@
   - Fix 'failed' telemetry counter double-increment due to race condition. [#102](https://github.com/kafka4beam/wolff/pull/102)
   - Added producer process label `{wolff_producer, KafkaClientId, Topic, Partition}` [#102](https://github.com/kafka4beam/wolff/pull/102)
   - Upgrade to `kafka_protocol-4.2.9` for better CRC32C performance.
+  - Add client config `allow_auto_topic_creation` (default = false).
 
 * 4.0.13 (merge 1.5.19)
   - Handle `record_list_too_large` error returned from Kafka.

src/wolff_client.erl

@@ -129,7 +129,7 @@ check_connectivity(Hosts, ConnConfig) when Hosts =/= [] ->
 -spec check_if_topic_exists([host()], kpro:conn_config(), topic()) ->
         ok | {error, unknown_topic_or_partition | [#{host := binary(), reason := term()}] | any()}.
 check_if_topic_exists(Hosts, ConnConfig, Topic) when Hosts =/= [] ->
-  case get_metadata(Hosts, ConnConfig, Topic) of
+  case get_metadata(Hosts, ConnConfig, Topic, _IsAutoCreateAllowed = false) of
     {ok, {Pid, _}} ->
       ok = close_connection(Pid);
     {error, Errors} ->
@@ -148,7 +148,10 @@ safe_call(Pid, Call) ->
   catch exit : Reason -> {error, Reason}
   end.
 
-%% request client to send Pid the leader connection.
+%% @doc request client to send Pid the leader connection.
+%% The caller will receive message `{leader_connection, Pid}' if the leader is successfully connected,
+%% otherwise `{leader_connection, {down, Reason}}'.
+-spec recv_leader_connection(pid(), ?NO_GROUP | producer_group(), topic(), partition(), pid(), all_partitions | pos_integer()) -> ok.
 recv_leader_connection(Client, Group, Topic, Partition, Caller, MaxPartitions) ->
   gen_server:cast(Client, {recv_leader_connection, Group, Topic, Partition, Caller, MaxPartitions}).
 
@@ -288,7 +291,7 @@ ensure_metadata_conn(#{seed_hosts := Hosts, conn_config := ConnConfig, metadata_
   end.
 
 check_if_topic_exists2(Pid, Topic, Timeout) when is_pid(Pid) ->
-  case do_get_metadata(Pid, Topic, Timeout) of
+  case do_get_metadata(Pid, Topic, Timeout, _IsAutoCreateAllowed = false) of
     {ok, _} ->
       ok;
     {error, Reason} ->
@@ -394,9 +397,13 @@ ensure_leader_connections(St, Group, Topic, MaxPartitions) ->
 
 -spec ensure_leader_connections2(state(), producer_group(), topic(), max_partitions()) ->
           {ok, state()} | {error, term()}.
-ensure_leader_connections2(#{metadata_conn := Pid, conn_config := ConnConfig} = St, Group, Topic, MaxPartitions) when is_pid(Pid) ->
+ensure_leader_connections2(#{metadata_conn := Pid,
+                             conn_config := ConnConfig,
+                             config := Config
+                            } = St, Group, Topic, MaxPartitions) when is_pid(Pid) ->
   Timeout = maps:get(request_timeout, ConnConfig, ?DEFAULT_METADATA_TIMEOUT),
-  case do_get_metadata(Pid, Topic, Timeout) of
+  IsAutoCreateAllowed = maps:get(allow_auto_topic_creation, Config, false),
+  case do_get_metadata(Pid, Topic, Timeout, IsAutoCreateAllowed) of
     {ok, {Brokers, PartitionMetaList}} ->
       ensure_leader_connections3(St, Group, Topic, Pid, Brokers, PartitionMetaList, MaxPartitions);
     {error, _Reason} ->
@@ -406,8 +413,11 @@ ensure_leader_connections2(#{metadata_conn := Pid, conn_config := ConnConfig} =
       ensure_leader_connections2(St#{metadata_conn => down}, Group, Topic, MaxPartitions)
   end;
 ensure_leader_connections2(#{conn_config := ConnConfig,
-                             seed_hosts := SeedHosts} = St, Group, Topic, MaxPartitions) ->
-  case get_metadata(SeedHosts, ConnConfig, Topic, []) of
+                             seed_hosts := SeedHosts,
+                             config := Config
+                            } = St, Group, Topic, MaxPartitions) ->
+  IsAutoCreateAllowed = maps:get(allow_auto_topic_creation, Config, false),
+  case get_metadata(SeedHosts, ConnConfig, Topic, IsAutoCreateAllowed, []) of
     {ok, {ConnPid, {Brokers, PartitionMetaList}}} ->
       ensure_leader_connections3(St, Group, Topic, ConnPid, Brokers, PartitionMetaList, MaxPartitions);
     {error, unknown_topic_or_partition} ->
@@ -572,22 +582,22 @@ split_config(Config) ->
   {ConnCfg, MyCfg} = lists:partition(Pred, maps:to_list(Config)),
   {maps:from_list(ConnCfg), maps:from_list(MyCfg)}.
 
--spec get_metadata([_Host], _ConnConfig, topic()) ->
+-spec get_metadata([_Host], _ConnConfig, topic(), boolean()) ->
           {ok, {pid(), term()}} | {error, term()}.
-get_metadata(Hosts, _ConnectFun, _Topic) when Hosts =:= [] ->
+get_metadata(Hosts, _ConnectFun, _Topic, _IsAutoCreateAllowed) when Hosts =:= [] ->
   {error, no_hosts};
-get_metadata(Hosts, ConnectFun, Topic) ->
-  get_metadata(Hosts, ConnectFun, Topic, []).
+get_metadata(Hosts, ConnectFun, Topic, IsAutoCreateAllowed) ->
+  get_metadata(Hosts, ConnectFun, Topic, IsAutoCreateAllowed, []).
 
--spec get_metadata([_Host], _ConnConfig, topic(), [Error]) ->
+-spec get_metadata([_Host], _ConnConfig, topic(), boolean(), [Error]) ->
           {ok, {pid(), term()}} | {error, [Error] | term()}.
-get_metadata([], _ConnConfig, _Topic, Errors) ->
+get_metadata([], _ConnConfig, _Topic, _IsAutoCreateAllowed, Errors) ->
   {error, Errors};
-get_metadata([Host | Rest], ConnConfig, Topic, Errors) ->
+get_metadata([Host | Rest], ConnConfig, Topic, IsAutoCreateAllowed, Errors) ->
   case do_connect(Host, ConnConfig) of
     {ok, Pid} ->
       Timeout = maps:get(request_timeout, ConnConfig, ?DEFAULT_METADATA_TIMEOUT),
-      case do_get_metadata(Pid, Topic, Timeout) of
+      case do_get_metadata(Pid, Topic, Timeout, IsAutoCreateAllowed) of
         {ok, Result} ->
           {ok, {Pid, Result}};
         {error, Reason} ->
@@ -596,23 +606,55 @@ get_metadata([Host | Rest], ConnConfig, Topic, Errors) ->
           {error, Reason}
       end;
     {error, Reason} ->
-      get_metadata(Rest, ConnConfig, Topic, [Reason | Errors])
+      get_metadata(Rest, ConnConfig, Topic, IsAutoCreateAllowed, [Reason | Errors])
   end.
 
--spec do_get_metadata(connection(), topic(), timeout()) ->
+-spec do_get_metadata(connection(), topic(), timeout(), boolean()) ->
           {ok, {_Brokers, _Partitions}} | {error, term()}.
-do_get_metadata(Connection, Topic, Timeout) ->
+do_get_metadata(Connection, Topic, Timeout, IsAutoCreateAllowed) ->
   case kpro:get_api_versions(Connection) of
     {ok, Vsns} ->
       {_, Vsn} = maps:get(metadata, Vsns),
-      do_get_metadata2(Vsn, Connection, Topic, Timeout);
+      do_get_metadata2(Vsn, Connection, Topic, Timeout, IsAutoCreateAllowed);
     {error, Reason} ->
       {error, Reason}
   end.
 
--spec do_get_metadata2(_Vsn, connection(), topic(), timeout()) -> {ok, {_, _}} | {error, term()}.
-do_get_metadata2(Vsn, Connection, Topic, Timeout) ->
-  Req = kpro_req_lib:metadata(Vsn, [Topic], _IsAutoCreateAllowed = false),
+-spec retry(fun(() -> {ok, term()} | {error, term()}), timeout()) -> {ok, term()} | {error, term()}.
+retry(Fun, Timeout) ->
+  Deadline = erlang:monotonic_time(millisecond) + Timeout,
+  retry_loop(Fun, Deadline).
+
+-spec retry_loop(fun(() -> {ok, term()} | {error, term()}), integer()) -> {ok, term()} | {error, term()}.
+retry_loop(Fun, Deadline) ->
+  case Fun() of
+    {ok, Result} ->
+      {ok, Result};
+    {error, R} when R =:= unknown_topic_or_partition orelse R =:= leader_not_available ->
+      CurrentTime = erlang:monotonic_time(millisecond),
+      case CurrentTime >= Deadline of
+        true ->
+          {error, timeout};
+        false ->
+          %% Sleep for a short interval before retrying
+          RemainingTime = Deadline - CurrentTime,
+          SleepTime = min(1000, RemainingTime),
+          timer:sleep(SleepTime),
+          retry_loop(Fun, Deadline)
+      end;
+    {error, Reason} ->
+      {error, Reason}
+  end.
+
+-spec do_get_metadata2(_Vsn, connection(), topic(), timeout(), boolean()) -> {ok, {_, _}} | {error, term()}.
+do_get_metadata2(Vsn, Connection, Topic, Timeout, _IsAutoCreateAllowed = true) ->
+  retry(fun() -> do_get_metadata3(Vsn, Connection, Topic, Timeout, true) end, Timeout);
+do_get_metadata2(Vsn, Connection, Topic, Timeout, _IsAutoCreateAllowed = false) ->
+  do_get_metadata3(Vsn, Connection, Topic, Timeout, false).
+
+-spec do_get_metadata3(_Vsn, connection(), topic(), timeout(), boolean()) -> {ok, {_, _}} | {error, term()}.
+do_get_metadata3(Vsn, Connection, Topic, Timeout, IsAutoCreateAllowed) ->
+  Req = kpro_req_lib:metadata(Vsn, [Topic], IsAutoCreateAllowed),
   case kpro:request_sync(Connection, Req, Timeout) of
     {ok, #kpro_rsp{msg = Meta}} ->
       BrokersMeta = kpro:find(brokers, Meta),
@@ -671,7 +713,7 @@ parse_broker_meta(BrokerMeta) ->
   Port = kpro:find(port, BrokerMeta),
   {BrokerId, {Host, Port}}.
 
-log_warn(Msg, Report) -> logger:warning(Report#{msg => Msg}).
+log_warn(Msg, Report) -> logger:warning(Report#{msg => Msg, pid => self()}).
 
 do_connect(Host, ConnConfig) ->
     case kpro:connect(Host, ConnConfig) of

test/wolff_client_SUITE.erl

@@ -0,0 +1,164 @@
+-module(wolff_client_SUITE).
+
+-compile([export_all, nowarn_export_all]).
+
+-include("wolff.hrl").
+-include_lib("kafka_protocol/include/kpro.hrl").
+-include_lib("eunit/include/eunit.hrl").
+
+-define(HOSTS, [{"localhost", 9092}]).
+-define(config(Key), proplists:get_value(Key, Config)).
+
+all() ->
+  Exports = ?MODULE:module_info(exports),
+  [F || {F, _} <- Exports, lists:prefix("t_", atom_to_list(F))].
+
+init_per_suite(Config) ->
+  _ = application:stop(wolff),
+  application:ensure_all_started(wolff),
+  Config.
+
+end_per_suite(_Config) ->
+  application:stop(wolff),
+  ok.
+
+init_per_testcase(Case, Config) ->
+  ?MODULE:Case({init, Config}).
+
+end_per_testcase(Case, Config) ->
+  ?MODULE:Case({'end', Config}).
+
+t_recv_leader_connection_normal({init, Config}) ->
+  ClientId = atom_to_binary(?FUNCTION_NAME),
+  ClientCfg = client_config(),
+  {ok, ClientPid} = wolff:ensure_supervised_client(ClientId, ?HOSTS, ClientCfg),
+  [{clientid, ClientId}, {client, ClientPid} | Config];
+t_recv_leader_connection_normal({'end', Config}) ->
+  wolff:stop_and_delete_supervised_client(?config(clientid));
+t_recv_leader_connection_normal(Config) ->
+  Client = ?config(client),
+  Topic = <<"test-topic">>,
+  wolff_client:recv_leader_connection(Client, ?NO_GROUP, Topic, 0, self(), ?all_partitions),
+  receive
+    {leader_connection, Pid} ->
+      ?assert(is_pid(Pid))
+  end,
+  ok.
+
+t_recv_leader_connection_with_auto_create_retry_success({init, Config}) ->
+  ClientId = atom_to_binary(?FUNCTION_NAME),
+  ClientCfg = client_config_with_auto_create(),
+  {ok, ClientPid} = wolff:ensure_supervised_client(ClientId, ?HOSTS, ClientCfg),
+  [{clientid, ClientId}, {client, ClientPid} | Config];
+t_recv_leader_connection_with_auto_create_retry_success({'end', Config}) ->
+  wolff:stop_and_delete_supervised_client(?config(clientid));
+t_recv_leader_connection_with_auto_create_retry_success(Config) ->
+  %% Test retry mechanism when auto-creation is enabled and topic doesn't exist initially
+  ok = meck:new(kpro, [non_strict, no_history, no_link, passthrough]),
+  OldRetryCount = get(metadata_retry_count),
+  try
+    put(metadata_retry_count, 0),
+    meck:expect(kpro, request_sync,
+      fun(Connection, #kpro_req{api = metadata} = Req, Timeout) ->
+        case get(metadata_retry_count) of
+          0 ->
+            put(metadata_retry_count, 1),
+            {error, unknown_topic_or_partition};
+          1 ->
+            put(metadata_retry_count, 2),
+            {error, unknown_topic_or_partition};
+          _ ->
+            %% Return success after retries - use passthrough for real response
+            meck:passthrough([Connection, Req, Timeout])
+        end;
+      (Connection, Req, Timeout) ->
+        %% Passthrough all non-metadata requests
+        meck:passthrough([Connection, Req, Timeout])
+      end),
+    Client = ?config(client),
+    Topic = <<"test-topic">>,
+    wolff_client:recv_leader_connection(Client, ?NO_GROUP, Topic, 0, self(), ?all_partitions),
+    receive
+      {leader_connection, Pid} when is_pid(Pid) ->
+        ?assert(is_pid(Pid));
+      {leader_connection, {down, Reason}} ->
+        %% May get error if topic doesn't exist on real Kafka
+        ?assertNotEqual(undefined, Reason)
+    after
+      15000 ->
+        ?assert(false, "Expected leader connection message")
+    end,
+    ok
+  after
+    meck:unload(kpro),
+    case OldRetryCount of
+      undefined -> erase(metadata_retry_count);
+      _ -> put(metadata_retry_count, OldRetryCount)
+    end
+  end.
+
+t_recv_leader_connection_with_auto_create_retry_timeout({init, Config}) ->
+  ClientId = atom_to_binary(?FUNCTION_NAME),
+  ClientCfg = client_config_with_auto_create(),
+  {ok, ClientPid} = wolff:ensure_supervised_client(ClientId, ?HOSTS, ClientCfg),
+  [{clientid, ClientId}, {client, ClientPid} | Config];
+t_recv_leader_connection_with_auto_create_retry_timeout({'end', Config}) ->
+  wolff:stop_and_delete_supervised_client(?config(clientid));
+t_recv_leader_connection_with_auto_create_retry_timeout(Config) ->
+  %% Test retry mechanism timeout when auto-creation is enabled
+  ok = meck:new(kpro, [non_strict, no_history, no_link, passthrough]),
+  try
+    meck:expect(kpro, request_sync,
+      fun(_Connection, #kpro_req{api = metadata} = _Req, _Timeout) ->
+        %% Always return unknown_topic_or_partition to trigger retry timeout
+        {error, unknown_topic_or_partition};
+      (Connection, Req, Timeout) ->
+        %% Passthrough all non-metadata requests
+        meck:passthrough([Connection, Req, Timeout])
+      end),
+    Client = ?config(client),
+    Topic = <<"test-topic-timeout">>,
+    wolff_client:recv_leader_connection(Client, ?NO_GROUP, Topic, 0, self(), ?all_partitions),
+    receive
+      {leader_connection, {down, timeout}} ->
+        %% Should receive down message due to timeout
+        ok;
+      {leader_connection, {down, Reason}} ->
+        %% May receive timeout or another error reason depending on timing
+        ?assert(is_atom(Reason))
+    after
+      5000 ->
+        ?assert(false, "Expected leader connection down message with timeout")
+    end,
+    ok
+  after
+    meck:unload(kpro)
+  end.
+
+client_config() -> #{}.
+
+client_config_with_auto_create() ->
+  #{allow_auto_topic_creation => true,
+    request_timeout => 2000
+   }.
+
+client_config_without_auto_create() ->
+  #{allow_auto_topic_creation => false,
+    request_timeout => 1000
+   }.
+
+%% Helper function for meck mocking (similar to pattern used in other test files)
+with_meck(Mod, FnName, MockedFn, Action) ->
+  ok = meck:new(Mod, [non_strict, no_history, no_link, passthrough]),
+  ok = meck:expect(Mod, FnName, MockedFn),
+  try
+    Action()
+  after
+    meck:unload(Mod)
+  end.
+
+%% Helper function to create a mock connection for testing
+mock_connection() ->
+  %% Create a mock connection that can be used in tests
+  %% This is a simplified mock - in real tests you might want to use a more sophisticated mock
+  {mock_connection, self()}.