Allow keyword configuration for producers

Author: ShapovalovaIrina

README.md

@@ -305,8 +305,8 @@ Multiple Producers configuration:
 
 ```elixir
 config :kaffe,
-  producers: %{
-    "producer_1" => [
+  producers: [
+    producer_1: [
       endpoints: [kafka1: 9092], # [hostname: port]
       topics: ["kafka-topic"],
   
@@ -319,11 +319,11 @@ config :kaffe,
         password: System.get_env("KAFFE_PRODUCER_PASSWORD")
       }
     ],
-    "producer_2" => [
+    producer_2: [
       endpoints: [kafka2: 9092], # [hostname: port]
       topics: ["another-kafka-topic"]
     ]
-  }
+  ]
 ```
 
 The `partition_strategy` setting can be one of:
@@ -375,7 +375,7 @@ There are several ways to produce:
     ```
 
     ```elixir
-    Kaffe.Producer.produce_sync_with_client("producer_1", "topic", [{"key1", "value1"}, {"key2", "value2"}])
+    Kaffe.Producer.produce_sync_with_client(:producer_1, "topic", [{"key1", "value1"}, {"key2", "value2"}])
     ```
 
 - `topic`/`partition`/`message_list` - Produce each message in the list to the given `topic`/`partition`.
@@ -387,7 +387,7 @@ There are several ways to produce:
     ```
 
     ```elixir
-    Kaffe.Producer.produce_sync_with_client("producer_1", "topic", 2, [{"key1", "value1"}, {"key2", "value2"}])
+    Kaffe.Producer.produce_sync_with_client(:producer_1, "topic", 2, [{"key1", "value1"}, {"key2", "value2"}])
     ```
 
 - `key`/`value` - The key/value will be produced to the first topic given to the producer when it was started. The partition will be selected with the chosen strategy or given function.
@@ -397,7 +397,7 @@ There are several ways to produce:
     ```
 
     ```elixir
-    Kaffe.Producer.produce_sync_with_client("producer_1", "key", "value")
+    Kaffe.Producer.produce_sync_with_client(:producer_1, "key", "value")
     ```
 
 - `topic`/`key`/`value` - The key/value will be produced to the given topic.
@@ -407,7 +407,7 @@ There are several ways to produce:
     ```
 
     ```elixir
-    Kaffe.Producer.produce_sync_with_client("producer_1", "whitelist", "key", "value")
+    Kaffe.Producer.produce_sync_with_client(:producer_1, "whitelist", "key", "value")
     ```
 
 - `topic`/`partition`/`key`/`value` - The key/value will be produced to the given topic/partition.
@@ -417,7 +417,7 @@ There are several ways to produce:
     ```
 
     ```elixir
-    Kaffe.Producer.produce_sync_with_client("producer_1", "whitelist", 2, "key", "value")
+    Kaffe.Producer.produce_sync_with_client(:producer_1, "whitelist", 2, "key", "value")
     ```
 
     **NOTE**: With this approach Kaffe will not calculate the next partition since it assumes you're taking over that job by giving it a specific partition.

config/test.exs

@@ -25,8 +25,8 @@ config :kaffe,
       }
     ]
   },
-  producers: %{
-    "producer_name" => [
+  producers: [
+    producer_name: [
       endpoints: [kafka: 9092],
       topics: ["kaffe-test"],
       sasl: %{
@@ -35,4 +35,4 @@ config :kaffe,
         password: System.get_env("KAFFE_PRODUCER_PASSWORD")
       }
     ]
-  }
+  ]

lib/kaffe.ex

@@ -18,7 +18,7 @@ defmodule Kaffe do
 
     if producers = Application.get_env(:kaffe, :producers) do
       producers
-      |> Map.keys()
+      |> Enum.map(&elem(&1, 0))
       |> Enum.each(fn config_key ->
         Logger.debug("event#start_producer_client=#{__MODULE__}_#{config_key}")
         config = Kaffe.Config.Producer.configuration(config_key)

lib/kaffe/config/producer.ex

@@ -60,7 +60,7 @@ defmodule Kaffe.Config.Producer do
 
   require Logger
 
-  @default_producer_config_key "producer"
+  @default_producer_config_key :producer
 
   def configuration(config_key) do
     %{
@@ -141,33 +141,33 @@ defmodule Kaffe.Config.Producer do
   def config_get!(config_key, key) do
     :kaffe
     |> Application.get_env(:producers)
-    |> Map.fetch!(config_key)
+    |> Access.fetch!(config_key)
     |> Keyword.fetch!(key)
   end
 
   def config_get(config_key, key, default) do
     :kaffe
     |> Application.get_env(:producers)
-    |> Map.fetch!(config_key)
+    |> Access.fetch!(config_key)
     |> Keyword.get(key, default)
   end
 
   def list_config_keys do
     :kaffe
     |> Application.get_env(:producers)
-    |> Map.keys()
+    |> Enum.map(&elem(&1, 0))
   end
 
   @doc """
   Sets :kaffe, :producers application env if :kaffe, :producer is present.
 
   Provides backward compatibility between single producer and multiple producers.
-  `#{@default_producer_config_key}` config key is used for multiple producers config.
+  `:#{@default_producer_config_key}` config key is used for multiple producers config.
   """
   @spec maybe_set_producers_env!() :: :ok
   def maybe_set_producers_env! do
     single_config = Application.get_env(:kaffe, :producer) || []
-    multiple_config = Application.get_env(:kaffe, :producers) || %{}
+    multiple_config = Application.get_env(:kaffe, :producers) || []
 
     if !Enum.empty?(single_config) and !Enum.empty?(multiple_config) do
       raise("""
@@ -178,7 +178,7 @@ defmodule Kaffe.Config.Producer do
     end
 
     if !Enum.empty?(single_config) and Enum.empty?(multiple_config) do
-      multiple_config = %{@default_producer_config_key => single_config}
+      multiple_config = [{@default_producer_config_key, single_config}]
 
       Logger.info("""
       Configuration for single producer is specified in :kaffe, :producer.

test/kaffe/config/producer_test.exs

@@ -1,7 +1,7 @@
 defmodule Kaffe.Config.ProducerTest do
   use ExUnit.Case, async: true
 
-  @default_producer "producer_name"
+  @default_producer :producer_name
 
   describe "configuration/0" do
     test "correct settings are extracted" do
@@ -75,19 +75,19 @@ defmodule Kaffe.Config.ProducerTest do
     test "correct settings are extracted for different producer clients" do
       config = Application.get_env(:kaffe, :producers)
 
-      multiple_producers_config = %{
-        "producer_1" => [
+      multiple_producers_config = [
+        producer_1: [
           endpoints: [kafka1: 9092],
           topics: ["kaffe-test-1"],
           sasl: %{mechanism: :plain, login: "Alice", password: "ecilA"},
           ssl: true
         ],
-        "producer_2" => [
+        producer_2: [
           endpoints: [kafka2: 9092],
           topics: ["kaffe-test-2"],
           compression: :zstd
         ]
-      }
+      ]
 
       Application.put_env(:kaffe, :producers, multiple_producers_config)
 
@@ -141,8 +141,28 @@ defmodule Kaffe.Config.ProducerTest do
         Application.put_env(:kaffe, :producers, config)
       end)
 
-      assert Kaffe.Config.Producer.configuration("producer_1") == expected_producer_config_1
-      assert Kaffe.Config.Producer.configuration("producer_2") == expected_producer_config_2
+      assert Kaffe.Config.Producer.configuration(:producer_1) == expected_producer_config_1
+      assert Kaffe.Config.Producer.configuration(:producer_2) == expected_producer_config_2
+    end
+
+    test "the same settings are extracted from map and keyword configuration" do
+      keyword_config = Application.get_env(:kaffe, :producers)
+      assert is_list(keyword_config)
+      assert keyword_config != []
+
+      producer_config_from_keyword = Kaffe.Config.Producer.configuration(@default_producer)
+
+      map_config = Map.new(keyword_config)
+
+      Application.put_env(:kaffe, :producers, map_config)
+
+      on_exit(fn ->
+        Application.put_env(:kaffe, :producers, keyword_config)
+      end)
+
+      producer_config_from_map = Kaffe.Config.Producer.configuration(@default_producer)
+
+      assert producer_config_from_keyword == producer_config_from_map
     end
   end
 
@@ -163,7 +183,8 @@ defmodule Kaffe.Config.ProducerTest do
 
     test "returns :ok if producers are configured with the :producers key" do
       producers_config = Application.get_env(:kaffe, :producers)
-      assert is_map(producers_config)
+      assert is_list(producers_config)
+      assert producers_config != []
 
       Application.delete_env(:kaffe, :producer)
 
@@ -175,7 +196,7 @@ defmodule Kaffe.Config.ProducerTest do
 
     test "returns :ok and sets :kaffe, :producers if producer is configured with the :producer key" do
       producers_config = Application.get_env(:kaffe, :producers)
-      producer_config = producers_config |> Map.values() |> List.first()
+      producer_config = producers_config |> Keyword.values() |> List.first()
 
       Application.delete_env(:kaffe, :producers)
       Application.put_env(:kaffe, :producer, producer_config)
@@ -188,12 +209,12 @@ defmodule Kaffe.Config.ProducerTest do
       assert :ok == Kaffe.Config.Producer.maybe_set_producers_env!()
 
       assert Application.get_env(:kaffe, :producer) == producer_config
-      assert Application.get_env(:kaffe, :producers) == %{"producer" => producer_config}
+      assert Application.get_env(:kaffe, :producers) == [producer: producer_config]
     end
 
     test "logs message if :kaffe, :producers was set with configuration from :kaffe, :producer" do
       producers_config = Application.get_env(:kaffe, :producers)
-      producer_config = producers_config |> Map.values() |> List.first()
+      producer_config = producers_config |> Keyword.values() |> List.first()
 
       Application.delete_env(:kaffe, :producers)
       Application.put_env(:kaffe, :producer, producer_config)
@@ -214,13 +235,13 @@ defmodule Kaffe.Config.ProducerTest do
              To ensure backward compatibility :kaffe, :producers was set to a map \
              with default producer name as the key and single producer config as the value:
 
-             config :kaffe, producers: %{\"producer\" => #{inspect(producer_config)}}
+             config :kaffe, producers: [producer: #{inspect(producer_config)}]
              """
     end
 
     test "raises error is both :producer and :producers keys are present" do
       producers_config = Application.get_env(:kaffe, :producers)
-      producer_config = producers_config |> Map.values() |> List.first()
+      producer_config = producers_config |> Keyword.values() |> List.first()
 
       Application.put_env(:kaffe, :producer, producer_config)
 

test/kaffe/producer_test.exs

@@ -3,7 +3,7 @@ defmodule Kaffe.ProducerTest do
 
   alias Kaffe.Producer
 
-  @default_client_config_key "producer_name"
+  @default_client_config_key :producer_name
   @default_client_name :kaffe_producer_client_producer_name
 
   setup do
@@ -97,25 +97,25 @@ defmodule Kaffe.ProducerTest do
     setup do
       current_producers_config = Application.get_env(:kaffe, :producers)
 
-      producers_config = %{
-        "producer_1" => [
+      producers_config = [
+        producer_1: [
           endpoints: [kafka1: 9092],
           topics: ["topic", "topic2"],
           partition_strategy: :md5
         ],
-        "producer_2" => [
+        producer_2: [
           endpoints: [kafka2: 9092],
           topics: ["topic"],
           partition_strategy: :md5
         ]
-      }
+      ]
 
       Application.put_env(:kaffe, :producers, producers_config)
       on_exit(fn -> Application.put_env(:kaffe, :producers, current_producers_config) end)
 
       producers =
         producers_config
-        |> Map.keys()
+        |> Keyword.keys()
         |> Enum.map(&{&1, Kaffe.Config.Producer.configuration(&1)})
 
       %{producers: producers}