Broadway Kafka can't connect through SASL

Questions / Help
, ,
1

Hi guys, I’m facing an issue trying to connect to kafka using the broadway_kafka connector, I need to pass a user and pass but I’m getting an error that it couldn’t join the given group, looking the output I see the arguments are not being taken as expected, looks like the password position is being held as the callback, even when I pass the :plain key in config:

Output config on start:

mfargs: {:brod_client, :start_link,
   [
     [{"xxxx.aws.confluent.cloud", 9092}],
     HubspotWorkers.HubspotDuplicaetDomainsWorker.Broadway.Producer_0.Client,
     [
       sasl: {:plain, "user",
        #Function<11.44977799/0 in :brod_utils.init_sasl_opt/1>}
     ]
   ]},
  restart_type: {:permanent, 10},
  shutdown: 5000,
  child_type: :worker

Output error:

[warning] :brod_client [#PID<0.474.0>] HubspotWorkers.HubspotDuplicaetDomainsWorker.Broadway.Producer_0.Client is terminating
reason: [
  {{"xxxxxx.aws.confluent.cloud", 9092},
   {{{:kpro_req, #Reference<0.3725464545.3257925635.50906>, :api_versions, 0,
      false, []}, :closed},
    [
      {:kpro_lib, :send_and_recv_raw, 4,
       [
         file: '/Users/gastonrey/dev/elixir/hubspot_workers/deps/kafka_protocol/src/kpro_lib.erl',
         line: 70
       ]},
      {:kpro_lib, :send_and_recv, 5,
       [
         file: '/Users/gastonrey/dev/elixir/hubspot_workers/deps/kafka_protocol/src/kpro_lib.erl',
         line: 81
       ]},
      {:kpro_connection, :query_api_versions, 4,
       [
         file: '/Users/gastonrey/dev/elixir/hubspot_workers/deps/kafka_protocol/src/kpro_connection.erl',
         line: 246
       ]},
      {:kpro_connection, :init_connection, 3,
       [
         file: '/Users/gastonrey/dev/elixir/hubspot_workers/deps/kafka_protocol/src/kpro_connection.erl',
         line: 233
       ]},
      {:kpro_connection, :init, 4,
       [
         file: '/Users/gastonrey/dev/elixir/hubspot_workers/deps/kafka_protocol/src/kpro_connection.erl',
         line: 170
       ]},
      {:proc_lib, :init_p_do_apply, 3, [file: 'proc_lib.erl', line: 240]}
    ]}}
]

[error] GenServer HubspotWorkers.HubspotDuplicaetDomainsWorker.Broadway.Producer_0.Client terminating
** (stop) [{{"pkc-zpjg0.eu-central-1.aws.confluent.cloud", 9092}, {{{:kpro_req, #Reference<0.3725464545.3257925635.50906>, :api_versions, 0, false, []}, :closed}, [{:kpro_lib, :send_and_recv_raw, 4, [file: '/Users/gastonrey/dev/elixir/hubspot_workers/deps/kafka_protocol/src/kpro_lib.erl', line: 70]}, {:kpro_lib, :send_and_recv, 5, [file: '/Users/gastonrey/dev/elixir/hubspot_workers/deps/kafka_protocol/src/kpro_lib.erl', line: 81]}, {:kpro_connection, :query_api_versions, 4, [file: '/Users/gastonrey/dev/elixir/hubspot_workers/deps/kafka_protocol/src/kpro_connection.erl', line: 246]}, {:kpro_connection, :init_connection, 3, [file: '/Users/gastonrey/dev/elixir/hubspot_workers/deps/kafka_protocol/src/kpro_connection.erl', line: 233]}, {:kpro_connection, :init, 4, [file: '/Users/gastonrey/dev/elixir/hubspot_workers/deps/kafka_protocol/src/kpro_connection.erl', line: 170]}, {:proc_lib, :init_p_do_apply, 3, [file: 'proc_lib.erl', line: 240]}]}}]
    (brod 3.16.5) /Users/gastonrey/dev/elixir/hubspot_workers/deps/brod/src/brod_client.erl:622: :brod_client.ensure_metadata_connection/1
    (brod 3.16.5) /Users/gastonrey/dev/elixir/hubspot_workers/deps/brod/src/brod_client.erl:343: :brod_client.handle_info/2
    (stdlib 4.3.1) gen_server.erl:1123: :gen_server.try_dispatch/4
    (stdlib 4.3.1) gen_server.erl:1200: :gen_server.handle_msg/6
    (stdlib 4.3.1) proc_lib.erl:240: :proc_lib.init_p_do_apply/3
Last message: :init
State: {:state, HubspotWorkers.HubspotDuplicaetDomainsWorker.Broadway.Producer_0.Client, [{"xxxxxx.aws.confluent.cloud", 9092}], :undefined, [], :undefined, :undefined, [sasl: {:plain, "xxxxx", #Function<11.44977799/0 in :brod_utils.init_sasl_opt/1>}], HubspotWorkers.HubspotDuplicaetDomainsWorker.Broadway.Producer_0.Client}
[info] Group member (hubspot_duplicate_domains_1,coor=#PID<0.478.0>,cb=#PID<0.473.0>,generation=0):
failed to join group
reason: {:client_down,
 [
   {{"xxxxxx.aws.confluent.cloud", 9092},
    {{{:kpro_req, #Reference<0.3725464545.3257925635.50906>, :api_versions, 0,
       false, []}, :closed},
     [
       {:kpro_lib, :send_and_recv_raw, 4,
        [
          file: '/Users/gastonrey/dev/elixir/hubspot_workers/deps/kafka_protocol/src/kpro_lib.erl',
          line: 70
        ]},
       {:kpro_lib, :send_and_recv, 5,
        [
          file: '/Users/gastonrey/dev/elixir/hubspot_workers/deps/kafka_protocol/src/kpro_lib.erl',
          line: 81
        ]},
       {:kpro_connection, :query_api_versions, 4,
        [
          file: '/Users/gastonrey/dev/elixir/hubspot_workers/deps/kafka_protocol/src/kpro_connection.erl',
          line: 246
        ]},
       {:kpro_connection, :init_connection, 3,
        [
          file: '/Users/gastonrey/dev/elixir/hubspot_workers/deps/kafka_protocol/src/kpro_connection.erl',
          line: 233
        ]},
       {:kpro_connection, :init, 4,
        [
          file: '/Users/gastonrey/dev/elixir/hubspot_workers/deps/kafka_protocol/src/kpro_connection.erl',
          line: 170
        ]},
       {:proc_lib, :init_p_do_apply, 3, [file: 'proc_lib.erl', line: 240]}
     ]}}
 ]}
[info] Group member (hubspot_duplicate_domains_1,coor=#PID<0.478.0>,cb=#PID<0.473.0>,generation=0):
Leaving group, reason: :shutdown

My config:

{BroadwayKafka.Producer,
           [
             client_config: [sasl: {:plain, "user", "password"}],
             hosts: "xxxxx.aws.confluent.cloud:9092",
             group_id: "xxxx_1",
             topics: ["xxxx.events.json"]
           ]},

What is wrong? Am using broadway_kafka 0.4

2

Solved, forgot the ssl: true, so config is like:

client_config: [
  ssl: true,
  sasl:
 {:plain, "user", "pass"}
 ]