Merge pull request #11107 from rabbitmq/mergify/bp/v3.13.x/pr-11101

Cleanup shovels deleted from another node (backport #11101)

Author: michaelklishin

deps/rabbitmq_shovel/src/rabbit_shovel_dyn_worker_sup_sup.erl

@@ -82,8 +82,7 @@ stop_child({VHost, ShovelName} = Name) ->
                     TmpExpId = temp_experimental_id(Name),
                     _ = stop_and_delete_child(TmpExpId),
                     ok
-            end,
-            rabbit_shovel_status:remove(Name)
+            end
     end,
     rabbit_shovel_locks:unlock(LockId),
     ok.

deps/rabbitmq_shovel/src/rabbit_shovel_worker.erl

@@ -150,10 +150,14 @@ terminate({shutdown, autodelete}, State = #state{name = Name,
     _ = rabbit_runtime_parameters:clear(VHost, <<"shovel">>, ShovelName, ?SHOVEL_USER),
     rabbit_shovel_status:remove(Name),
     ok;
-terminate(shutdown, State) ->
+terminate(shutdown, State = #state{name = Name}) ->
     close_connections(State),
+    rabbit_shovel_status:remove(Name),
     ok;
-terminate(socket_closed_unexpectedly, State) ->
+terminate(socket_closed_unexpectedly, State = #state{name = Name}) ->
+    rabbit_log_shovel:error("Shovel ~ts is stopping because of the socket closed unexpectedly", [human_readable_name(Name)]),
+    rabbit_shovel_status:report(State#state.name, State#state.type,
+                                {terminated, "socket closed"}),
     close_connections(State),
     ok;
 terminate({'EXIT', heartbeat_timeout}, State = #state{name = Name}) ->

deps/rabbitmq_shovel/test/delete_shovel_command_SUITE.erl

@@ -9,21 +9,26 @@
 
 -include_lib("common_test/include/ct.hrl").
 -include_lib("amqp_client/include/amqp_client.hrl").
+-include_lib("stdlib/include/assert.hrl").
 
 -compile(export_all).
 
 -define(CMD, 'Elixir.RabbitMQ.CLI.Ctl.Commands.DeleteShovelCommand').
 
 all() ->
     [
-      {group, non_parallel_tests}
+      {group, non_parallel_tests},
+      {group, cluster_size_2}
     ].
 
 groups() ->
     [
      {non_parallel_tests, [], [
                                delete_not_found,
                                delete
+                              ]},
+     {cluster_size_2, [], [
+                               clear_param_on_different_node
                               ]}
     ].
 
@@ -33,25 +38,31 @@ groups() ->
 
 init_per_suite(Config) ->
     rabbit_ct_helpers:log_environment(),
+    rabbit_ct_helpers:run_setup_steps(Config).
+
+end_per_suite(Config) ->
+    rabbit_ct_helpers:run_teardown_steps(Config).
+
+init_per_group(cluster_size_2, Config) ->
+    init_per_multinode_group(cluster_size_2, Config, 2);
+init_per_group(Group, Config) ->
+    init_per_multinode_group(Group, Config, 1).
+
+init_per_multinode_group(_Group, Config, NodeCount) ->
+    Suffix = rabbit_ct_helpers:testcase_absname(Config, "", "-"),
     Config1 = rabbit_ct_helpers:set_config(Config, [
-        {rmq_nodename_suffix, ?MODULE}
+                                                    {rmq_nodes_count, NodeCount},
+                                                    {rmq_nodename_suffix, Suffix}
       ]),
-    Config2 = rabbit_ct_helpers:run_setup_steps(Config1,
-      rabbit_ct_broker_helpers:setup_steps() ++
-      rabbit_ct_client_helpers:setup_steps()),
-    Config2.
+    rabbit_ct_helpers:run_steps(Config1,
+    rabbit_ct_broker_helpers:setup_steps() ++
+    rabbit_ct_client_helpers:setup_steps()).
 
-end_per_suite(Config) ->
-    rabbit_ct_helpers:run_teardown_steps(Config,
+end_per_group(_Group, Config) ->
+    rabbit_ct_helpers:run_steps(Config,
       rabbit_ct_client_helpers:teardown_steps() ++
       rabbit_ct_broker_helpers:teardown_steps()).
 
-init_per_group(_, Config) ->
-    Config.
-
-end_per_group(_, Config) ->
-    Config.
-
 init_per_testcase(Testcase, Config) ->
     rabbit_ct_helpers:testcase_started(Config, Testcase).
 
@@ -76,3 +87,18 @@ delete(Config) ->
     ok = ?CMD:run([<<"myshovel">>], Opts),
     [] = rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_shovel_status,
                                       status, []).
+clear_param_on_different_node(Config) ->
+    shovel_test_utils:set_param(
+      Config,
+      <<"myshovel">>, [{<<"src-queue">>,  <<"src">>},
+                       {<<"dest-queue">>, <<"dest">>}]),
+    [A, B] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+    [_] = rabbit_ct_broker_helpers:rpc(Config, A, rabbit_shovel_status,
+                                       status, []),
+    [] = rabbit_ct_broker_helpers:rpc(Config, B, rabbit_shovel_status,
+                                      status, []),
+    shovel_test_utils:clear_param(Config, B, <<"myshovel">>),
+    ?assertEqual([], rabbit_ct_broker_helpers:rpc(Config, A, rabbit_shovel_status,
+                                                  status, []), "Deleted shovel still reported on node A"),
+    ?assertEqual([], rabbit_ct_broker_helpers:rpc(Config, B, rabbit_shovel_status,
+                                                  status, []), "Deleted shovel still reported on node B").