fix: differentiate producers to the same topic with aliases

README.md

@@ -54,6 +54,14 @@ ok = wolff:stop_producers(Producers).
 ok = wolff:stop_client(Client).
 ```
 
+If you want to use more than one producer pointing to the same topic, be sure to define an unique `alias` for each one to avoid clashes.
+
+```erlang
+Topic = <<"test-topic">>.
+{ok, Producers1} = wolff:start_producers(Client, Topic, ProducerCfg#{alias => <<"a1">>}).
+{ok, Producers2} = wolff:start_producers(Client, Topic, ProducerCfg#{alias => <<"a2">>}).
+```
+
 ### Async Produce with Callback
 
 ```
@@ -182,7 +190,7 @@ defining telemetry events. Wolff defines such telemetry events. Users of Wolff
 can attach functions to the events, for example, to record when a message has
 been successfully sent to Kafka. Wolff's telemetry events are described in the
 `wolff_metrics` module. One can read more about how to attach code to the
-events in [Beam Telemetry's documentation](https://github.com/beam-telemetry/telemetry). 
+events in [Beam Telemetry's documentation](https://github.com/beam-telemetry/telemetry).
 The third parameter of the Beam Telemetry handler function is a meta data map.
 One can send a custom meta data map for each Kafka producer instance by setting
 the Kafka producer configuration parameter `telemetry_meta_data` to the map one

changelog.md

@@ -1,3 +1,6 @@
+* 2.0.0
+  - Added the `alias` producer config option to make producers to the same topic be independent.
+
 * 1.10.4 (merge 1.5.14)
   - Split batch if `message_too_large` error code is received.
     Prior to this fix, `wolff_producer` would retry the same batch indefinitely for any error code received from Kafka (`message_too_large` included).

docker-compose.yml

@@ -3,6 +3,10 @@ services:
   zookeeper:
     image: wurstmeister/zookeeper
     container_name: wolff-zk
+    ulimits:
+      nofile:
+        soft: 65536
+        hard: 65536
   kafka_1:
     depends_on:
       - zookeeper

include/wolff.hrl

@@ -25,4 +25,9 @@
 %% is inserted to cache the partition count.
 -define(WOLFF_PRODUCERS_GLOBAL_TABLE, wolff_producers_global).
 
+-define(ALIASED_TOPIC(ALIAS, TOPIC), {ALIAS, TOPIC}).
+-define(NO_ALIAS, no_alias).
+
+-define(all_partitions, all_partitions).
+
 -endif.

src/wolff.app.src

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

src/wolff.appup.src

@@ -1,5 +1,5 @@
 %% -*- mode: erlang; -*-
-{"1.10.4",
+{"2.0.0",
   [
   ],
   [

src/wolff.erl

@@ -14,6 +14,8 @@
 
 -module(wolff).
 
+-include("wolff.hrl").
+
 %% Supervised client management APIs
 -export([ensure_supervised_client/3,
          stop_and_delete_supervised_client/1
@@ -26,10 +28,7 @@
 
 %% Supervised producer management APIs
 -export([ensure_supervised_producers/3,
-         stop_and_delete_supervised_producers/1,
-         stop_and_delete_supervised_producers/2,
-         %% /3 is deprecated, call /2 instead
-         stop_and_delete_supervised_producers/3
+         stop_and_delete_supervised_producers/1
         ]).
 
 %% Messaging APIs
@@ -49,7 +48,6 @@
               name/0, offset_reply/0, topic/0]).
 
 -deprecated({check_if_topic_exists, 3}).
--deprecated({stop_and_delete_supervised_producers, 3}).
 
 -type client_id() :: binary().
 -type host() :: kpro:endpoint().
@@ -97,15 +95,6 @@ stop_producers(Producers) ->
 ensure_supervised_producers(ClientId, Topic, ProducerCfg) ->
   wolff_producers:start_supervised(ClientId, Topic, ProducerCfg).
 
-%% @hidden Deprecated.
--spec stop_and_delete_supervised_producers(client_id(), topic(), name()) -> ok.
-stop_and_delete_supervised_producers(ClientId, Topic, _Name) ->
-    stop_and_delete_supervised_producers(ClientId, Topic).
-
-%% @doc Ensure supervised producers are stopped then deleted.
-stop_and_delete_supervised_producers(ClientId, Topic) ->
-  wolff_producers:stop_supervised(ClientId, Topic).
-
 %% @doc Ensure supervised producers are stopped then deleted.
 -spec stop_and_delete_supervised_producers(wolff_producers:producers()) -> ok.
 stop_and_delete_supervised_producers(Producers) ->