diff --git a/deps/rabbitmq_jms_topic_exchange/src/rabbit_db_jms_exchange.erl b/deps/rabbitmq_jms_topic_exchange/src/rabbit_db_jms_exchange.erl index 107f5ddf13a6..999003be7285 100644 --- a/deps/rabbitmq_jms_topic_exchange/src/rabbit_db_jms_exchange.erl +++ b/deps/rabbitmq_jms_topic_exchange/src/rabbit_db_jms_exchange.erl @@ -38,14 +38,51 @@ setup_schema() -> }). setup_schema_in_mnesia() -> - case mnesia:create_table( ?JMS_TOPIC_TABLE - , [ {attributes, record_info(fields, ?JMS_TOPIC_RECORD)} - , {record_name, ?JMS_TOPIC_RECORD} - , {type, set} ] - ) of - {atomic, ok} -> ok; - {aborted, {already_exists, ?JMS_TOPIC_TABLE}} -> ok - end, + TableName = ?JMS_TOPIC_TABLE, + rabbit_log:info( + "Creating table ~ts for JMS topic exchange", + [TableName]), + _ = try + rabbit_table:create( + TableName, + [{attributes, record_info(fields, ?JMS_TOPIC_RECORD)}, + {record_name, ?JMS_TOPIC_RECORD}, + {type, set}]), + %% The JMS topic exchange table must be available on all nodes. + %% If it existed on only one node, messages could not be published + %% to JMS topic exchanges and routed to topic subscribers if the node + %% was unavailable. + %% The call below makes sure this node has a copy of the table. + case rabbit_table:ensure_table_copy(TableName, node(), ram_copies) of + ok -> + %% Next, we try to fix other nodes in the cluster if they are + %% running a version of RabbitMQ which does not replicate the + %% table. All nodes must have a replica for Mnesia operations + %% to work properly. Therefore the code below is to make older + %% compatible with newer nodes. + Replicas = mnesia:table_info(TableName, all_nodes), + Members = rabbit_nodes:list_running(), + MissingOn = Members -- Replicas, + lists:foreach( + fun(Node) -> + %% Errors from adding a replica on those older nodes + %% are ignored however. They should not be fatal. The + %% problem will solve by itself once all nodes are + %% upgraded. + _ = rpc:call( + Node, + rabbit_table, ensure_table_copy, + [TableName, Node, ram_copies]) + end, MissingOn), + ok; + Error -> + Error + end + catch throw:Reason -> + rabbit_log:error( + "Failed to create JMS topic exchange table: ~tp", + [Reason]) + end, ok. %% -------------------------------------------------------------------