How to stop a Broadway Kafka pipeline?

[aravindanck]

Team, I tried Broadway.stop(__ MODULE __) in a broadway_kafka pipeline. But the pipeline gets restarted after a minute by the brod supervisor :brod_sup. The restart strategy is fixed to permanent in brod library (see ref) - could it be the cause? Is there a way I can start my broadway kafka producer with 'restart' arg set to temporary/transient?

:supervisor: {:local, :brod_sup}
    :started: [
  pid: #PID<0.1456.0>,
  id: TestBroadway.Broadway.Producer_0.Client,
  mfargs: {:brod_client, :start_link,
   [[localhost: 9092], TestBroadway.Broadway.Producer_0.Client, []]},
  restart_type: {:permanent, 10},
  shutdown: 5000,
  child_type: :worker
]

Reference: https://github.com/kafka4beam/brod/blob/master/src/brod_sup.erl#L164

[amacciola]

@aravindanck how i do this is by starting all my BroadwayPipelines under a DynamicSupervisor.

Then when i want to stop them i just call

    String.to_atom("name_of_pipeline")
    |> Process.whereis()
    |> case do
      nil ->
        {:ok, :success}

      child_pid ->
        GenServer.stop(child_pid)
        {:ok, :success}
    end

[aravindanck]

@amacciola Broadway's stop API call that does the same thing I believe. I'm using it to stop the pipeline, but the pipeline gets restarted after a minute automatically.
Ref: https://hexdocs.pm/broadway/Broadway.html#stop/3

[josevalim]

@aravindanck this shouldn't happen because we stop the client on the producer shutdown:

broadway_kafka/lib/broadway_kafka/producer.ex

Lines 515 to 520 in 1eea41e

	def terminate(_reason, state) do
	%{client: client, group_coordinator: group_coordinator, client_id: client_id} = state
	client.stop_group_coordinator(group_coordinator)
	client.disconnect(client_id)
	:ok
	end

So you need to debug why that line is not being invoked and the client still sticks around. :)