fix: make sure wolff_client:check_topic_exists_with_client_pid will t…

Makefile

@@ -1,4 +1,4 @@
-KAFKA_IMAGE_VERSION ?= 1.1.3
+KAFKA_IMAGE_VERSION ?= 1.2.1
 export KAFKA_IMAGE_VERSION
 KAFKA_VERSION ?= 4.0.0
 export KAFKA_VERSION

changelog.md

@@ -1,3 +1,6 @@
+* 4.1.2
+  - Made sure `wolff_client:check_topic_exists_with_client_pid/2` triggers topic creation when `allow_auto_topic_creation` is set to `true`.
+
 * 4.1.1
   - Do not linger before enqueue when queue is not on disk.
     Since 4.0.0, the linger is moved from after the queue to before the queue to optimize IOPS in disk mode.

scripts/docker-compose-kraft.yml

@@ -30,6 +30,7 @@ services:
       ADVERTISED_HOSTNAME: ${KAFKA_1_IP}
       INNER_HOSTNAME: ${KAFKA_1_IP}
       VOTERS: 1@${KAFKA_1_IP}:9090
+      AUTO_CREATE_TOPICS_ENABLE: "true"
   kafka_2:
     image: ${KAFKA_IMAGE_TAG}
     container_name: 'kafka-2'
@@ -57,3 +58,4 @@ services:
       ADVERTISED_HOSTNAME: ${KAFKA_2_IP}
       INNER_HOSTNAME: ${KAFKA_2_IP}
       VOTERS: 1@${KAFKA_1_IP}:9090
+      AUTO_CREATE_TOPICS_ENABLE: "true"

scripts/docker-compose.yml

@@ -34,6 +34,7 @@ services:
       SASL_PLAINTEXT_PORT: 9095
       ADVERTISED_HOSTNAME: ${KAFKA_1_IP}
       ZOOKEEPER_CONNECT: ${ZOOKEEPER_IP}:2181
+      AUTO_CREATE_TOPICS_ENABLE: "true"
   kafka_2:
     depends_on:
       - zookeeper
@@ -55,3 +56,4 @@ services:
       SASL_PLAINTEXT_PORT: 9095
       ADVERTISED_HOSTNAME: ${KAFKA_2_IP}
       ZOOKEEPER_CONNECT: ${ZOOKEEPER_IP}:2181
+      AUTO_CREATE_TOPICS_ENABLE: "true"

scripts/setup-test-env.sh

@@ -56,7 +56,7 @@ case $KAFKA_VERSION in
     ;;
 esac
 
-KAFKA_IMAGE_VERSION="${KAFKA_IMAGE_VERSION:-1.1.3}"
+KAFKA_IMAGE_VERSION="${KAFKA_IMAGE_VERSION:-1.2.1}"
 export KAFKA_IMAGE_TAG="zmstone/kafka:${KAFKA_IMAGE_VERSION}-${KAFKA_VERSION}"
 echo "Using $KAFKA_IMAGE_TAG"
 

src/wolff_client.erl

@@ -167,11 +167,12 @@ handle_call(Call, From, #{connect := _Fun} = St) ->
     handle_call(Call, From, upgrade(St));
 handle_call(get_id, _From, #{client_id := Id} = St) ->
   {reply, Id, St};
-handle_call({check_if_topic_exists, Topic}, _From, #{conn_config := ConnConfig} = St0) ->
+handle_call({check_if_topic_exists, Topic}, _From, #{config := Config, conn_config := ConnConfig} = St0) ->
+  IsAutoCreateAllowed = maps:get(allow_auto_topic_creation, Config, false),
   case ensure_metadata_conn(St0) of
     {ok, #{metadata_conn := ConnPid} = St} ->
       Timeout = maps:get(request_timeout, ConnConfig, ?DEFAULT_METADATA_TIMEOUT),
-      {reply, check_if_topic_exists2(ConnPid, Topic, Timeout), St};
+      {reply, check_if_topic_exists2(ConnPid, Topic, Timeout, IsAutoCreateAllowed), St};
     {error, Reason} ->
       {reply, {error, Reason}, St0}
   end;
@@ -290,8 +291,8 @@ ensure_metadata_conn(#{seed_hosts := Hosts, conn_config := ConnConfig, metadata_
       end
   end.
 
-check_if_topic_exists2(Pid, Topic, Timeout) when is_pid(Pid) ->
-  case do_get_metadata(Pid, Topic, Timeout, _IsAutoCreateAllowed = false) of
+check_if_topic_exists2(Pid, Topic, Timeout, IsAutoCreateAllowed) when is_pid(Pid) ->
+  case do_get_metadata(Pid, Topic, Timeout, IsAutoCreateAllowed) of
     {ok, _} ->
       ok;
     {error, Reason} ->

test/wolff_client_SUITE.erl

@@ -28,6 +28,20 @@ init_per_testcase(Case, Config) ->
 end_per_testcase(Case, Config) ->
   ?MODULE:Case({'end', Config}).
 
+t_check_topic_exists_trigger_auto_creation({init, Config}) ->
+  ClientId = atom_to_binary(?FUNCTION_NAME),
+  ClientCfg = #{allow_auto_topic_creation => true,
+                request_timeout => 5000
+               },
+  {ok, ClientPid} = wolff:ensure_supervised_client(ClientId, ?HOSTS, ClientCfg),
+  [{clientid, ClientId}, {client, ClientPid} | Config];
+t_check_topic_exists_trigger_auto_creation({'end', Config}) ->
+  wolff:stop_and_delete_supervised_client(?config(clientid));
+t_check_topic_exists_trigger_auto_creation(Config) ->
+  Topic = iolist_to_binary(["tmp-", integer_to_list(erlang:system_time(microsecond))]),
+  Client = ?config(client),
+  ?assertEqual(ok, wolff_client:check_topic_exists_with_client_pid(Client, Topic)).
+
 t_recv_leader_connection_normal({init, Config}) ->
   ClientId = atom_to_binary(?FUNCTION_NAME),
   ClientCfg = client_config(),