Merge pull request #9802 from rabbitmq/migration-live-insertions

Khepri migration: match expected records

Author: dumbbell

deps/rabbit/app.bzl

@@ -539,6 +539,7 @@ def all_srcs(name = "all_srcs"):
             "include/amqqueue.hrl",
             "include/amqqueue_v2.hrl",
             "include/gm_specs.hrl",
+            "include/internal_user.hrl",
             "include/mc.hrl",
             "include/rabbit_global_counters.hrl",
             "include/vhost.hrl",

deps/rabbit/include/internal_user.hrl

@@ -0,0 +1,4 @@
+-define(is_internal_user(U),
+        (?is_internal_user_v2(U))).
+
+-define(is_internal_user_v2(U), is_record(U, internal_user, 6)).

deps/rabbit/src/rabbit_db_exchange_m2k_converter.erl

@@ -45,7 +45,7 @@ init_copy_to_khepri(StoreId, _MigrationId, Tables) ->
 %% @private
 
 copy_to_khepri(
-  rabbit_exchange = Table, Record,
+  rabbit_exchange = Table, #exchange{} = Record,
   #?MODULE{store_id = StoreId} = State) ->
     Name = Record#exchange.name,
     ?LOG_DEBUG(

deps/rabbit/src/rabbit_db_maintenance_m2k_converter.erl

@@ -45,7 +45,7 @@ init_copy_to_khepri(StoreId, _MigrationId, Tables) ->
 %% @private
 
 copy_to_khepri(
-  rabbit_node_maintenance_states = Table, Record,
+  rabbit_node_maintenance_states = Table, #node_maintenance_state{} = Record,
   #?MODULE{store_id = StoreId} = State) ->
     Name = Record#node_maintenance_state.node,
     ?LOG_DEBUG(

deps/rabbit/src/rabbit_db_queue_m2k_converter.erl

@@ -13,6 +13,7 @@
 -include_lib("khepri/include/khepri.hrl").
 -include_lib("khepri_mnesia_migration/src/kmm_logging.hrl").
 -include_lib("rabbit_common/include/rabbit.hrl").
+-include("amqqueue.hrl").
 
 -export([init_copy_to_khepri/3,
          copy_to_khepri/3,
@@ -45,7 +46,7 @@ init_copy_to_khepri(StoreId, _MigrationId, Tables) ->
 %% @private
 
 copy_to_khepri(rabbit_queue = Table, Record,
-               #?MODULE{store_id = StoreId} = State) ->
+               #?MODULE{store_id = StoreId} = State) when ?is_amqqueue(Record) ->
     Name = amqqueue:get_name(Record),
     ?LOG_DEBUG(
        "Mnesia->Khepri data copy: [~0p] key: ~0p",

deps/rabbit/src/rabbit_db_user_m2k_converter.erl

@@ -13,6 +13,7 @@
 -include_lib("khepri/include/khepri.hrl").
 -include_lib("khepri_mnesia_migration/src/kmm_logging.hrl").
 -include_lib("rabbit_common/include/rabbit.hrl").
+-include("internal_user.hrl").
 
 -export([init_copy_to_khepri/3,
          copy_to_khepri/3,
@@ -46,7 +47,7 @@ init_copy_to_khepri(StoreId, _MigrationId, Tables) ->
 
 copy_to_khepri(
   rabbit_user = Table, Record,
-  #?MODULE{store_id = StoreId} = State) ->
+  #?MODULE{store_id = StoreId} = State) when ?is_internal_user(Record) ->
     Username = internal_user:get_username(Record),
     ?LOG_DEBUG(
        "Mnesia->Khepri data copy: [~0p] key: ~0p",
@@ -62,7 +63,7 @@ copy_to_khepri(
         Error -> Error
     end;
 copy_to_khepri(
-  rabbit_user_permission = Table, Record,
+  rabbit_user_permission = Table, #user_permission{} = Record,
   #?MODULE{store_id = StoreId} = State) ->
     #user_permission{
        user_vhost = #user_vhost{
@@ -89,7 +90,7 @@ copy_to_khepri(
         Error -> Error
     end;
 copy_to_khepri(
-  rabbit_topic_permission = Table, Record,
+  rabbit_topic_permission = Table, #topic_permission{} = Record,
   #?MODULE{store_id = StoreId} = State) ->
     #topic_permission{
        topic_permission_key =

deps/rabbit/src/rabbit_db_vhost_m2k_converter.erl

@@ -13,6 +13,7 @@
 -include_lib("khepri/include/khepri.hrl").
 -include_lib("khepri_mnesia_migration/src/kmm_logging.hrl").
 -include_lib("rabbit_common/include/rabbit.hrl").
+-include("vhost.hrl").
 
 -export([init_copy_to_khepri/3,
          copy_to_khepri/3,
@@ -46,7 +47,7 @@ init_copy_to_khepri(StoreId, _MigrationId, Tables) ->
 
 copy_to_khepri(
   rabbit_vhost = Table, Record,
-  #?MODULE{store_id = StoreId} = State) ->
+  #?MODULE{store_id = StoreId} = State) when ?is_vhost(Record) ->
     Name = vhost:get_name(Record),
     ?LOG_DEBUG(
        "Mnesia->Khepri data copy: [~0p] key: ~0p",