Still need help with this. When I try running Broadway.Mesage.failed(messages, "fail"), I get this error message:
[error] ** (FunctionClauseError) no function clause matching in Broadway.Message.failed/2
(broadway 1.0.1) lib/broadway/message.ex:139: Broadway.Message.failed([%Broadway.Message{acknowledger: {BroadwayRabbitMQ.Producer, %AMQP.Channel{conn: %AMQP.Connection{pid: #PID<0.568.0>}, custom_consumer: {AMQP.SelectiveConsumer, #PID<0.554.0>}, pid: #PID<0.577.0>}, %{client: BroadwayRabbitMQ.AmqpClient, delivery_tag: 1, on_failure: :reject_and_requeue, on_success: :ack, redelivered: true}}, batch_key: :default, batch_mode: :bulk, batcher: :default, data: <<131, 116, 0, 0, 0, 4, 100, 0, 7, 109, 101, 115, 115, 97, 103, 101, 109, 0, 0, 0, 4, 97, 115, 100, 102, 100, 0, 6, 114, 111, 111, 109, 105, 100, 109, 0, 0, 0, 10, 114, 111, 111, 109, 58, ...>>, metadata: %{amqp_channel: %AMQP.Channel{conn: %AMQP.Connection{pid: #PID<0.568.0>}, custom_consumer: {AMQP.SelectiveConsumer, #PID<0.554.0>}, pid: #PID<0.577.0>}}, status: :ok}], "fail")
(exchat 0.1.0) lib/exchat_web/broadway/savetest.ex:45: ExchatWeb.TestSave.handle_batch/4
(broadway 1.0.1) lib/broadway/topology/batch_processor_stage.ex:100: Broadway.Topology.BatchProcessorStage.handle_batch/4
(broadway 1.0.1) lib/broadway/topology/batch_processor_stage.ex:56: anonymous fn/5 in Broadway.Topology.BatchProcessorStage.handle_events/3
(telemetry 0.4.3) /home/exchat/exchat/deps/telemetry/src/telemetry.erl:272: :telemetry.span/3
(gen_stage 1.1.2) lib/gen_stage.ex:2471: GenStage.consumer_dispatch/6
(stdlib 3.16.1) gen_server.erl:695: :gen_server.try_dispatch/4
(stdlib 3.16.1) gen_server.erl:771: :gen_server.handle_msg/6
(stdlib 3.16.1) proc_lib.erl:226: :proc_lib.init_p_do_apply/3
I’m trying to mark the entire batch as “failed”. I’m not sure why this isn’t working. I even added a handle_failed function.
def handle_failed(messages, "fail") do
for message <- messages do
Broadway.Message.configure_ack(message, retry: true)
end
end
But I get an error for that as well:
[error] ** (FunctionClauseError) no function clause matching in ExchatWeb.TestSave.handle_failed/2
(exchat 0.1.0) lib/exchat_web/broadway/savetest.ex:55: ExchatWeb.TestSave.handle_failed([%Broadway.Message{acknowledger: {BroadwayRabbitMQ.Producer, %AMQP.Channel{conn: %AMQP.Connection{pid: #PID<0.568.0>}, custom_consumer: {AMQP.SelectiveConsumer, #PID<0.554.0>}, pid: #PID<0.577.0>}, %{client: BroadwayRabbitMQ.AmqpClient, delivery_tag: 1, on_failure: :reject_and_requeue, on_success: :ack, redelivered: true}}, batch_key: :default, batch_mode: :bulk, batcher: :default, data: <<131, 116, 0, 0, 0, 4, 100, 0, 7, 109, 101, 115, 115, 97, 103, 101, 109, 0, 0, 0, 4, 97, 115, 100, 102, 100, 0, 6, 114, 111, 111, 109, 105, 100, 109, 0, 0, 0, 10, 114, 111, 111, 109, 58, ...>>, metadata: %{amqp_channel: %AMQP.Channel{conn: %AMQP.Connection{pid: #PID<0.568.0>}, custom_consumer: {AMQP.SelectiveConsumer, #PID<0.554.0>}, pid: #PID<0.577.0>}}, status: {:error, %FunctionClauseError{args: nil, arity: 2, clauses: nil, function: :failed, kind: nil, module: Broadway.Message}, [{Broadway.Message, :failed, [[%Broadway.Message{acknowledger: {BroadwayRabbitMQ.Producer, %AMQP.Channel{conn: %AMQP.Connection{pid: #PID<0.568.0>}, custom_consumer: {AMQP.SelectiveConsumer, #PID<0.554.0>}, pid: #PID<0.577.0>}, %{client: BroadwayRabbitMQ.AmqpClient, delivery_tag: 1, on_failure: :reject_and_requeue, on_success: :ack, redelivered: true}}, batch_key: :default, batch_mode: :bulk, batcher: :default, data: <<131, 116, 0, 0, 0, 4, 100, 0, 7, 109, 101, 115, 115, 97, 103, 101, 109, 0, 0, 0, 4, 97, 115, 100, 102, 100, 0, 6, ...>>, metadata: %{amqp_channel: %AMQP.Channel{conn: %AMQP.Connection{pid: #PID<0.568.0>}, custom_consumer: {AMQP.SelectiveConsumer, #PID<0.554.0>}, pid: #PID<0.577.0>}}, status: :ok}], "fail"], [file: 'lib/broadway/message.ex', line: 139]}, {ExchatWeb.TestSave, :handle_batch, 4, [file: 'lib/exchat_web/broadway/savetest.ex', line: 45]}, {Broadway.Topology.BatchProcessorStage, :handle_batch, 4, [file: 'lib/broadway/topology/batch_processor_stage.ex', line: 100]}, {Broadway.Topology.BatchProcessorStage, :"-handle_events/3-fun-0-", 5, [file: 'lib/broadway/topology/batch_processor_stage.ex', line: 56]}, {:telemetry, :span, 3, [file: '/home/exchat/exchat/deps/telemetry/src/telemetry.erl', line: 272]}, {GenStage, :consumer_dispatch, 6, [file: 'lib/gen_stage.ex', line: 2471]}, {:gen_server, :try_dispatch, 4, [file: 'gen_server.erl', line: 695]}, {:gen_server, :handle_msg, 6, [file: 'gen_server.erl', line: 771]}, {:proc_lib, :init_p_do_apply, 3, [file: 'proc_lib.erl', line: 226]}]}}], :context_not_set)
(broadway 1.0.1) lib/broadway/acknowledger.ex:102: Broadway.Acknowledger.handle_failed_messages/3
(broadway 1.0.1) lib/broadway/topology/batch_processor_stage.ex:59: anonymous fn/5 in Broadway.Topology.BatchProcessorStage.handle_events/3
(telemetry 0.4.3) /home/exchat/exchat/deps/telemetry/src/telemetry.erl:272: :telemetry.span/3
(gen_stage 1.1.2) lib/gen_stage.ex:2471: GenStage.consumer_dispatch/6
(stdlib 3.16.1) gen_server.erl:695: :gen_server.try_dispatch/4
(stdlib 3.16.1) gen_server.erl:771: :gen_server.handle_msg/6
(stdlib 3.16.1) proc_lib.erl:226: :proc_lib.init_p_do_apply/3
All I really need to do is mark the batch of messages as failed so they can be retried later. I actually don’t need to modify the failed messages or anything, so the handle_failed call might not even be necessary. Unfortunately, in both cases I don’t know how to do this in elixir. The script is almost finished except for this last part.
