Khepri data migration using khepri_mnesia_migration

26 files changed, 1554 insertions, 844 deletions

diff --git a/deps/rabbit/src/mirrored_supervisor.hrl b/deps/rabbit/src/mirrored_supervisor.hrl
new file mode 100644
index 0000000000..5287d1508e
--- /dev/null
+++ b/deps/rabbit/src/mirrored_supervisor.hrl
@@ -0,0 +1 @@
+-record(mirrored_sup_childspec, {key, mirroring_pid, childspec}).
diff --git a/deps/rabbit/src/rabbit_core_ff.erl b/deps/rabbit/src/rabbit_core_ff.erl
index bb453f4411..afa2709fce 100644
--- a/deps/rabbit/src/rabbit_core_ff.erl
+++ b/deps/rabbit/src/rabbit_core_ff.erl
@@ -14,7 +14,7 @@
 
 -export([mds_phase1_migration_enable/1,
          mds_phase1_migration_post_enable/1]).
--export([mds_migration_enable/2,
+-export([mds_plugin_migration_enable/2,
          mds_migration_post_enable/2]).
 
 -rabbit_feature_flag(
@@ -156,55 +156,58 @@
 %% This table order is important. For instance, user permissions depend on
 %% both vhosts and users to exist in the metadata store.
 
-%% TODO should they be integrated on phase1?
--define(MDS_PHASE2_TABLES, [{rabbit_queue, rabbit_db_queue},
-                            {rabbit_durable_queue, rabbit_db_queue},
-                            {rabbit_exchange, rabbit_db_exchange},
-                            {rabbit_durable_exchange, rabbit_db_exchange},
-                            {rabbit_exchange_serial, rabbit_db_exchange},
-                            {rabbit_route, rabbit_db_binding},
-                            {rabbit_durable_route, rabbit_db_binding},
-                            {rabbit_semi_durable_route, rabbit_db_binding},
-                            {rabbit_reverse_route, rabbit_db_binding},
-                            {rabbit_node_maintenance_states, rabbit_db_maintenance},
-                            {mirrored_sup_childspec, rabbit_db_msup}]).
-
--define(MDS_PHASE1_TABLES, [{rabbit_vhost, rabbit_db_vhost},
-                            {rabbit_user, rabbit_db_user},
-                            {rabbit_user_permission, rabbit_db_user},
-                            {rabbit_topic_permission, rabbit_db_user},
-                            {rabbit_runtime_parameters, rabbit_db_rtparams}]
-        ++ ?MDS_PHASE2_TABLES).
+-define(MDS_TABLES,
+        [
+         {[rabbit_vhost], rabbit_db_vhost_m2k_converter},
+         {[rabbit_user, rabbit_user_permission, rabbit_topic_permission],
+          rabbit_db_user_m2k_converter},
+         {[rabbit_runtime_parameters], rabbit_db_rtparams_m2k_converter},
+         {[rabbit_queue], rabbit_db_queue_m2k_converter},
+         {[rabbit_exchange, rabbit_exchange_serial], rabbit_db_exchange_m2k_converter},
+         {[rabbit_route], rabbit_db_binding_m2k_converter},
+         {[rabbit_node_maintenance_states], rabbit_db_maintenance_m2k_converter},
+         {[mirrored_sup_childspec], rabbit_db_msup_m2k_converter}
+        ]
+       ).
+
+-define(MDS_CLEANUP_TABLES,
+        [
+         {[rabbit_vhost], rabbit_db_vhost_m2k_converter},
+         {[rabbit_user, rabbit_user_permission, rabbit_topic_permission],
+          rabbit_db_user_m2k_converter},
+         {[rabbit_runtime_parameters], rabbit_db_rtparams_m2k_converter},
+         {[rabbit_queue, rabbit_durable_queue], rabbit_db_queue_m2k_converter},
+         {[rabbit_exchange, rabbit_durable_exchange, rabbit_exchange_serial], rabbit_db_exchange_m2k_converter},
+         {[rabbit_route, rabbit_durable_route, rabbit_semi_durable_route, rabbit_reverse_route, rabbit_index_route], rabbit_db_binding_m2k_converter},
+         {[rabbit_node_maintenance_states], rabbit_db_maintenance_m2k_converter},
+         {[mirrored_sup_childspec], rabbit_db_msup_m2k_converter}
+        ]
+       ).
 
 mds_phase1_migration_enable(#{feature_name := FeatureName}) ->
     %% Channel and connection tracking are core features with difference:
     %% tables cannot be predeclared as they include the node name
-    Tables = ?MDS_PHASE1_TABLES,
     global:set_lock({FeatureName, self()}),
     Ret = case rabbit_khepri:is_ready() of
               true ->
                   ok;
               false ->
-                  mds_migration_enable(FeatureName, Tables)
+                  mds_migration_enable(FeatureName, ?MDS_TABLES)
           end,
     global:del_lock({FeatureName, self()}),
     Ret.
 
 mds_phase1_migration_post_enable(#{feature_name := FeatureName}) ->
-    %% Channel and connection tracking are core features with difference:
-    %% tables cannot be predeclared as they include the node name
-    Tables = ?MDS_PHASE1_TABLES,
-    mds_migration_post_enable(FeatureName, Tables),
-    %% Mark the migration as done. This flag is necessary to skip clearing
-    %% data from previous migrations when a node joins an existing cluster.
-    %% If the migration has never been completed, then it's fair to remove
-    %% the existing data in Khepri.
-    rabbit_db:set_migration_flag(FeatureName).
+    %% Some tables are not necessary for the migration, but they
+    %% must be emptied afterwards (i.e. rabbit_durable_queue,
+    %% rabbit_reverse_route...)
+    Tables = ?MDS_CLEANUP_TABLES,
+    mds_migration_post_enable(FeatureName, Tables).
 
 mds_migration_enable(FeatureName, TablesAndOwners) ->
     case ensure_khepri_cluster_matches_mnesia(FeatureName) of
         ok ->
-            migrate_tables_to_khepri(FeatureName, TablesAndOwners);
+            mds_migrate_tables_to_khepri(FeatureName, TablesAndOwners);
         Error ->
             {error, {migration_failure, Error}}
     end.
@@ -212,7 +215,7 @@ mds_migration_enable(FeatureName, TablesAndOwners) ->
 mds_migration_post_enable(FeatureName, TablesAndOwners) ->
     ?assert(rabbit_khepri:is_enabled(non_blocking)),
     {Tables, _} = lists:unzip(TablesAndOwners),
-    empty_unused_mnesia_tables(FeatureName, Tables).
+    empty_unused_mnesia_tables(FeatureName, lists:flatten(Tables)).
 
 ensure_khepri_cluster_matches_mnesia(FeatureName) ->
     %% This is the first time Khepri will be used for real. Therefore
@@ -224,8 +227,15 @@ ensure_khepri_cluster_matches_mnesia(FeatureName) ->
        [FeatureName]),
     rabbit_khepri:init_cluster().
 
-migrate_tables_to_khepri(FeatureName, TablesAndOwners) ->
-    {Tables, _} = lists:unzip(TablesAndOwners),
+mds_plugin_migration_enable(FeatureName, TablesAndOwners) ->
+    global:set_lock({FeatureName, self()}),
+    Ret = mds_migrate_tables_to_khepri(FeatureName, TablesAndOwners),
+    global:del_lock({FeatureName, self()}),
+    Ret.
+
+mds_migrate_tables_to_khepri(FeatureName, TablesAndOwners) ->
+    {Tables0, _} = lists:unzip(TablesAndOwners),
+    Tables = lists:flatten(Tables0),
     rabbit_table:wait(Tables, _Retry = true),
     ?LOG_NOTICE(
        "Feature flag `~s`:   starting migration from Mnesia "
@@ -234,7 +244,7 @@ migrate_tables_to_khepri(FeatureName, TablesAndOwners) ->
        [FeatureName]),
     Pid = spawn(
             fun() ->
-                    migrate_tables_to_khepri_run(FeatureName, TablesAndOwners)
+                    mds_migrate_tables_to_khepri_run(FeatureName, TablesAndOwners)
             end),
     MonitorRef = erlang:monitor(process, Pid),
     receive
@@ -243,6 +253,11 @@ migrate_tables_to_khepri(FeatureName, TablesAndOwners) ->
                "Feature flag `~s`:   migration from Mnesia to Khepri "
                "finished",
                [FeatureName]),
+            %% Mark the migration as done. This flag is necessary to skip clearing
+            %% data from previous migrations when a node joins an existing cluster.
+            %% If the migration has never been completed, then it's fair to remove
+            %% the existing data in Khepri.
+            rabbit_db:set_migration_flag(FeatureName),
             _ = rabbit_khepri:set_ready(),
             ok;
         {'DOWN', MonitorRef, process, Pid, Info} ->
@@ -253,7 +268,7 @@ migrate_tables_to_khepri(FeatureName, TablesAndOwners) ->
             {error, {migration_failure, Info}}
     end.
 
-migrate_tables_to_khepri_run(FeatureName, TablesAndOwners) ->
+mds_migrate_tables_to_khepri_run(FeatureName, TablesAndOwners) ->
     %% Clear data in Khepri which could come from a previously aborted copy
     %% attempt. The table list order is important so we need to reverse that
     %% order to clear the data.
@@ -269,189 +284,18 @@ migrate_tables_to_khepri_run(FeatureName, TablesAndOwners) ->
         true ->
             ok;
         false ->
-            ok = clear_data_from_previous_attempt(FeatureName, lists:reverse(TablesAndOwners))
-    end,
-
-    {Tables, _} = lists:unzip(TablesAndOwners),
-    %% Subscribe to Mnesia events: we want to know about all writes and
-    %% deletions happening in parallel to the copy we are about to start.
-    ?LOG_DEBUG(
-       "Feature flag `~s`:   subscribe to Mnesia writes",
-       [FeatureName]),
-    ok = subscribe_to_mnesia_changes(FeatureName, Tables),
-
-    %% Copy from Mnesia to Khepri. Tables are copied in a specific order to
-    %% make sure that if term A depends on term B, term B was copied before.
-    ?LOG_DEBUG(
-       "Feature flag `~s`:   copy records from Mnesia to Khepri",
-       [FeatureName]),
-    ok = copy_from_mnesia_to_khepri(FeatureName, TablesAndOwners),
-
-    %% Mnesia transaction to handle received Mnesia events and tables removal.
-    ?LOG_DEBUG(
-       "Feature flag `~s`:   final sync and Mnesia table removal",
-       [FeatureName]),
-    ok = final_sync_from_mnesia_to_khepri(FeatureName, TablesAndOwners),
-
-    %% Unsubscribe to Mnesia events. All Mnesia tables are synchronized and
-    %% read-only at this point.
-    ?LOG_DEBUG(
-       "Feature flag `~s`:   unsubscribe to Mnesia writes",
-       [FeatureName]),
-    ok = unsubscribe_to_mnesia_changes(FeatureName, Tables).
+            ok = clear_data_from_previous_attempt(FeatureName, lists:reverse(TablesAndOwners)),
+            [mnesia_to_khepri:copy_tables(metadata_store, Tables, Mod)
+             || {Tables, Mod} <- TablesAndOwners]
+    end.
 
 clear_data_from_previous_attempt(
-  FeatureName, [{Table, Mod} | Rest]) ->
-    ok = Mod:clear_data_in_khepri(Table),
+  FeatureName, [{Tables, Mod} | Rest]) ->
+    [ok = Mod:clear_data_in_khepri(Table) || Table <- Tables],
     clear_data_from_previous_attempt(FeatureName, Rest);
 clear_data_from_previous_attempt(_, []) ->
     ok.
 
-subscribe_to_mnesia_changes(FeatureName, [Table | Rest]) ->
-    ?LOG_DEBUG(
-       "Feature flag `~s`:     subscribe to writes to ~s",
-       [FeatureName, Table]),
-    case mnesia:subscribe({table, Table, detailed}) of
-        {ok, _} -> subscribe_to_mnesia_changes(FeatureName, Rest);
-        {error, {not_active_local, _}} ->
-            rabbit_log:warning("Feature flag `~s`: table ~s doesn't have a local copy. Until ff v2 is merged it won't be migrated", [FeatureName, Table]);
-        Error   -> Error
-    end;
-subscribe_to_mnesia_changes(_, []) ->
-    ok.
-
-unsubscribe_to_mnesia_changes(FeatureName, [Table | Rest]) ->
-    ?LOG_DEBUG(
-       "Feature flag `~s`:     unsubscribe to writes to ~s",
-       [FeatureName, Table]),
-    case mnesia:unsubscribe({table, Table, detailed}) of
-        {ok, _} -> unsubscribe_to_mnesia_changes(FeatureName, Rest);
-        Error   -> Error
-    end;
-unsubscribe_to_mnesia_changes(_, []) ->
-    ok.
-
-copy_from_mnesia_to_khepri(
-  FeatureName, [{Table, Mod} | Rest]) ->
-    Fun = fun(Entries) ->
-                  Mod:mnesia_write_to_khepri(Table, Entries)
-          end,
-    do_copy_from_mnesia_to_khepri(FeatureName, Table, Fun),
-    copy_from_mnesia_to_khepri(FeatureName, Rest);
-copy_from_mnesia_to_khepri(_, []) ->
-    ok.
-
-do_copy_from_mnesia_to_khepri(FeatureName, Table, Fun) ->
-    Count = mnesia:table_info(Table, size),
-    ?LOG_DEBUG(
-       "Feature flag `~s`:     table ~s: about ~b record(s) to copy",
-       [FeatureName, Table, Count]),
-    Batch = application:get_env(rabbit, khepri_migration_batch, 1000),
-    Cont = mnesia:async_dirty(
-             fun () ->
-                     mnesia:select(Table, [{'$1',[],['$1']}], Batch, read) end),
-    do_copy_from_mnesia_to_khepri(
-      FeatureName, Table, Cont, Fun, Count, 0).
-
-do_copy_from_mnesia_to_khepri(
-  FeatureName, Table, '$end_of_table', _, Count, Copied) ->
-    ?LOG_DEBUG(
-       "Feature flag `~s`:     table ~s: copy of ~b record(s) (out of ~b "
-       "initially) finished",
-       [FeatureName, Table, Copied, Count]),
-    ok;
-do_copy_from_mnesia_to_khepri(
-  FeatureName, Table, {Records, Cont}, Fun, Count, Copied) ->
-    %% TODO: Batch several records in a single Khepri insert.
-    %% TODO: Can/should we parallelize?
-    case Copied rem 100 of
-        0 ->
-            ?LOG_DEBUG(
-               "Feature flag `~s`:     table ~s: copying record ~b/~b",
-               [FeatureName, Table, Copied, Count]);
-        _ ->
-            ok
-    end,
-    %% rabbit_listener is a bag, so this query might return a list of records
-    Fun(Records),
-    Next = mnesia:async_dirty(fun() -> mnesia:select(Cont) end),
-    do_copy_from_mnesia_to_khepri(
-      FeatureName, Table, Next, Fun, Count, Copied + length(Records)).
-
-final_sync_from_mnesia_to_khepri(FeatureName, TablesAndOwners) ->
-    %% Switch all tables to read-only. All concurrent and future Mnesia
-    %% transaction involving a write to one of them will fail with the
-    %% `{no_exists, Table}` exception.
-    {Tables, _} = lists:unzip(TablesAndOwners),
-    lists:foreach(
-      fun(Table) ->
-              ?LOG_DEBUG(
-                 "Feature flag `~s`:     switch table ~s to read-only",
-                 [FeatureName, Table]),
-                 case mnesia:change_table_access_mode(Table, read_only) of
-                     {atomic, ok} -> ok;
-                     {aborted, {already_exists, _, read_only}} -> ok
-                 end
-      end, Tables),
-
-    %% During the first round of copy, we received all write events as
-    %% messages (parallel writes were authorized). Now, we want to consume
-    %% those messages to record the writes we probably missed.
-    ok = consume_mnesia_events(FeatureName, TablesAndOwners),
-
-    ok.
-
-consume_mnesia_events(FeatureName, TablesAndOwners) ->
-    {_, Count} = erlang:process_info(self(), message_queue_len),
-    ?LOG_DEBUG(
-       "Feature flag `~s`:     handling queued Mnesia events "
-       "(about ~b events)",
-       [FeatureName, Count]),
-    consume_mnesia_events(FeatureName, TablesAndOwners, Count, 0).
-
-consume_mnesia_events(FeatureName, TablesAndOwners, Count, Handled) ->
-    %% TODO: Batch several events in a single Khepri command.
-    Handled1 = Handled + 1,
-    receive
-        {mnesia_table_event, {write, Table, NewRecord, _, _}} ->
-            ?LOG_DEBUG(
-               "Feature flag `~s`:       handling event ~b/~b (write)",
-               [FeatureName, Handled1, Count]),
-            handle_mnesia_write(Table, NewRecord, TablesAndOwners),
-            consume_mnesia_events(FeatureName, TablesAndOwners, Count, Handled1);
-        {mnesia_table_event, {delete, Table, {Table, Key}, _, _}} ->
-            ?LOG_DEBUG(
-               "Feature flag `~s`:       handling event ~b/~b (delete)",
-               [FeatureName, Handled1, Count]),
-            handle_mnesia_delete(Table, Key, TablesAndOwners),
-            consume_mnesia_events(FeatureName, TablesAndOwners, Count, Handled1);
-        {mnesia_table_event, {delete, Table, Record, _, _}} ->
-            ?LOG_DEBUG(
-               "Feature flag `~s`:       handling event ~b/~b (delete)",
-               [FeatureName, Handled1, Count]),
-            handle_mnesia_delete(Table, Record, TablesAndOwners),
-            consume_mnesia_events(FeatureName, TablesAndOwners, Count, Handled1)
-    after 0 ->
-              {_, MsgCount} = erlang:process_info(self(), message_queue_len),
-              ?LOG_DEBUG(
-                 "Feature flag `~s`:     ~b messages remaining",
-                 [FeatureName, MsgCount]),
-              %% TODO: Wait for confirmation from Khepri.
-              ok
-    end.
-
-%% TODO handle mnesia_runtime_parameters, rabbit_amqqueue, rabbit_exchange, rabbit_binding,
-%% rabbit_exchange_type_topic
-handle_mnesia_write(Table, NewRecord, TablesAndOwners) ->
-    {_, Mod} = lists:keyfind(Table, 1, TablesAndOwners),
-    Mod:mnesia_write_to_khepri(Table, [NewRecord]).
-
-%% TODO handle mnesia_runtime_parameters, rabbit_amqqueue, rabbit_exchange, rabbit_binding,
-%% rabbit_exchange_type_topic
-handle_mnesia_delete(Table, Key, TablesAndOwners) ->
-    {Mod, _} = lists:keyfind(Table, 1, TablesAndOwners),
-    Mod:mnesia_delete_to_khepri(Table, Key).
-
 %% We can't remove unused tables at this point yet. The reason is that tables
 %% are synchronized before feature flags in `rabbit_mnesia`. So if a node is
 %% already using Khepri and another node wants to join him, but is using Mnesia
diff --git a/deps/rabbit/src/rabbit_db_binding.erl b/deps/rabbit/src/rabbit_db_binding.erl
index aab9d343b0..3966526eff 100644
--- a/deps/rabbit/src/rabbit_db_binding.erl
+++ b/deps/rabbit/src/rabbit_db_binding.erl
@@ -35,9 +35,11 @@
          match_source_and_destination_in_khepri_tx/2
         ]).
 
--export([mnesia_write_to_khepri/2,
-         mnesia_delete_to_khepri/2,
-         clear_data_in_khepri/1]).
+-export([
+         khepri_route_path/1,
+         khepri_routes_path/0,
+         khepri_route_exchange_path/1
+        ]).
 
 %% Recovery is only needed for transient entities. Once mnesia is removed, these
 %% functions can be deleted
@@ -901,91 +903,6 @@ match_source_and_destination_in_khepri_tx(#resource{virtual_host = VHost, name =
         _         -> []
     end.
 
-%% --------------------------------------------------------------
-%% Migration
-%% --------------------------------------------------------------
-
--spec mnesia_write_to_khepri(Table, [Route]) -> ok when
-      Table :: atom(),
-      Route :: #route{}.
-mnesia_write_to_khepri(rabbit_route, Routes)->
-    rabbit_khepri:transaction(
-      fun() ->
-              _ = lists:foldl(
-                    fun(Route, Xs0) ->
-                            #route{binding = #binding{source = XName} = Binding} = Route,
-                            Path = khepri_route_path(Binding),
-                            Xs =
-                            case sets:is_element(XName, Xs0) of
-                                true ->
-                                    Xs0;
-                                false ->
-                                    %% If the binding's source is a new exchange,
-                                    %% store the exchange's type in the exchange
-                                    %% name branch of the tree.
-                                    XPath = khepri_route_exchange_path(XName),
-                                    [#exchange{type = XType}] =
-                                    rabbit_db_exchange:get_in_khepri_tx(XName),
-                                    ok = khepri_tx:put(XPath, #{type => XType}),
-                                    sets:add_element(XName, Xs0)
-                            end,
-                            add_binding_tx(Path, Binding),
-                            Xs
-                    end, sets:new([{version, 2}]), Routes),
-              ok
-      end, rw);
-mnesia_write_to_khepri(rabbit_durable_route, _)->
-    ok;
-mnesia_write_to_khepri(rabbit_semi_durable_route, _)->
-    ok;
-mnesia_write_to_khepri(rabbit_reverse_route, _) ->
-    ok.
-
-add_binding_tx(Path, Binding) ->
-    Set = case khepri_tx:get(Path) of
-              {ok, Set0} ->
-                  Set0;
-              _ ->
-                  sets:new()
-          end,
-    ok = khepri_tx:put(Path, sets:add_element(Binding, Set)).
-
--spec mnesia_delete_to_khepri(Table, ToDelete) -> ok when
-      Table :: atom(),
-      ToDelete :: #route{} | rabbit_types:binding().
-
-mnesia_delete_to_khepri(rabbit_route, Route) when is_record(Route, route) ->
-    khepri_delete(khepri_route_path(Route#route.binding));
-mnesia_delete_to_khepri(rabbit_route, Name) ->
-    khepri_delete(khepri_route_path(Name));
-mnesia_delete_to_khepri(rabbit_durable_route, Route) when is_record(Route, route) ->
-    khepri_delete(khepri_route_path(Route#route.binding));
-mnesia_delete_to_khepri(rabbit_durable_route, Name) ->
-    khepri_delete(khepri_route_path(Name));
-mnesia_delete_to_khepri(rabbit_semi_durable_route, Route) when is_record(Route, route) ->
-    khepri_delete(khepri_route_path(Route#route.binding));
-mnesia_delete_to_khepri(rabbit_semi_durable_route, Name) ->
-    khepri_delete(khepri_route_path(Name)).
-
--spec clear_data_in_khepri(Table) -> ok when
-      Table :: atom().
-
-clear_data_in_khepri(rabbit_route) ->
-    khepri_delete(khepri_routes_path());
-%% There is a single khepri entry for routes and it should be already deleted
-clear_data_in_khepri(rabbit_durable_route) ->
-    ok;
-clear_data_in_khepri(rabbit_semi_durable_route) ->
-    ok;
-clear_data_in_khepri(rabbit_reverse_route) ->
-    ok.
-
-khepri_delete(Path) ->
-    case rabbit_khepri:delete(Path) of
-        ok -> ok;
-        Error -> throw(Error)
-    end.
-
 %% -------------------------------------------------------------------
 %% clear().
 %% -------------------------------------------------------------------
@@ -1009,8 +926,11 @@ clear_in_mnesia() ->
     ok.
 
 clear_in_khepri() ->
-    khepri_delete(khepri_routes_path()),
-    ok.
+    Path = rabbit_db_binding:khepri_routes_path(),
+    case rabbit_khepri:delete(Path) of
+        ok -> ok;
+        Error -> throw(Error)
+    end.
 
 %% --------------------------------------------------------------
 %% Paths
diff --git a/deps/rabbit/src/rabbit_db_binding_m2k_converter.erl b/deps/rabbit/src/rabbit_db_binding_m2k_converter.erl
new file mode 100644
index 0000000000..c7e80090af
--- /dev/null
+++ b/deps/rabbit/src/rabbit_db_binding_m2k_converter.erl
@@ -0,0 +1,122 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright © 2022-2023 VMware, Inc. or its affiliates.  All rights reserved.
+%%
+
+-module(rabbit_db_binding_m2k_converter).
+
+-behaviour(mnesia_to_khepri_converter).
+
+-include_lib("kernel/include/logger.hrl").
+-include_lib("khepri/include/khepri.hrl").
+-include_lib("khepri_mnesia_migration/src/kmm_logging.hrl").
+-include_lib("rabbit_common/include/rabbit.hrl").
+
+-export([init_copy_to_khepri/2,
+         copy_to_khepri/3,
+         delete_from_khepri/3,
+         clear_data_in_khepri/1]).
+
+-record(?MODULE, {store_id :: khepri:store_id(),
+                  exchanges :: sets:set()}).
+
+-spec init_copy_to_khepri(Tables, StoreId) -> Ret when
+      Tables :: [mnesia_to_khepri:mnesia_table()],
+      StoreId :: khepri:store_id(),
+      Ret :: {ok, Priv},
+      Priv :: #?MODULE{}.
+%% @private
+
+init_copy_to_khepri(_Tables, StoreId) ->
+    State = #?MODULE{store_id = StoreId,
+                     exchanges = sets:new([{version, 2}])},
+    {ok, State}.
+
+-spec copy_to_khepri(Table, Record, Priv) -> Ret when
+      Table :: mnesia_to_khepri:mnesia_table(),
+      Record :: tuple(),
+      Priv :: #?MODULE{},
+      Ret :: {ok, NewPriv} | {error, Reason},
+      NewPriv :: #?MODULE{},
+      Reason :: any().
+%% @private
+
+copy_to_khepri(rabbit_route = Table,
+               #route{binding = #binding{source = XName} = Binding},
+               #?MODULE{exchanges = Xs0} = State) ->
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data copy: [~0p] key: ~0p",
+       [Table, Binding],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    Path = rabbit_db_binding:khepri_route_path(Binding),
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data copy: [~0p] path: ~0p",
+       [Table, Path],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    rabbit_khepri:transaction(
+      fun() ->
+              Xs = case sets:is_element(XName, Xs0) of
+                       true ->
+                           Xs0;
+                       false ->
+                           %% If the binding's source is a new exchange,
+                           %% store the exchange's type in the exchange
+                           %% name branch of the tree.
+                           XPath = rabbit_db_binding:khepri_route_exchange_path(XName),
+                           [#exchange{type = XType}] =
+                               rabbit_db_exchange:get_in_khepri_tx(XName),
+                           ok = khepri_tx:put(XPath, #{type => XType}),
+                           sets:add_element(XName, Xs0)
+                   end,
+              Set = case khepri_tx:get(Path) of
+                        {ok, Set0} ->
+                            Set0;
+                        _ ->
+                            sets:new()
+                    end,
+              case khepri_tx:put(Path, sets:add_element(Binding, Set)) of
+                  ok    -> {ok, State#?MODULE{exchanges = Xs}};
+                 Error -> Error
+              end
+      end, rw);
+copy_to_khepri(Table, Record, State) ->
+    ?LOG_DEBUG("Mnesia->Khepri unexpected record table ~0p record ~0p state ~0p",
+               [Table, Record, State]),
+    {error, unexpected_record}.
+
+-spec delete_from_khepri(Table, Key, Priv) -> Ret when
+      Table :: mnesia_to_khepri:mnesia_table(),
+      Key :: any(),
+      Priv :: #?MODULE{},
+      Ret :: {ok, NewPriv} | {error, Reason},
+      NewPriv :: #?MODULE{},
+      Reason :: any().
+%% @private
+
+delete_from_khepri(rabbit_route = Table, Key,
+                   #?MODULE{store_id = StoreId} = State) ->
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data delete: [~0p] key: ~0p",
+       [Table, Key],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    Path = rabbit_db_binding:khepri_route_path(Key),
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data delete: [~0p] path: ~0p",
+       [Table, Path],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    case khepri:delete(StoreId, Path) of
+        ok    -> {ok, State};
+        Error -> Error
+    end.
+
+-spec clear_data_in_khepri(Table) -> ok when
+      Table :: atom().
+
+clear_data_in_khepri(rabbit_route) ->
+    Path = rabbit_db_binding:khepri_routes_path(),
+    case rabbit_khepri:delete(Path) of
+        ok -> ok;
+        Error -> throw(Error)
+    end.
diff --git a/deps/rabbit/src/rabbit_db_exchange.erl b/deps/rabbit/src/rabbit_db_exchange.erl
index a41c354f2b..8ce1fba886 100644
--- a/deps/rabbit/src/rabbit_db_exchange.erl
+++ b/deps/rabbit/src/rabbit_db_exchange.erl
@@ -44,13 +44,16 @@
          path/1
          ]).
 
--export([mnesia_write_to_khepri/2,
-         mnesia_delete_to_khepri/2,
-         clear_data_in_khepri/1]).
-
 %% For testing
 -export([clear/0]).
 
+-export([
+         khepri_exchange_path/1,
+         khepri_exchange_serial_path/1,
+         khepri_exchanges_path/0,
+         khepri_exchange_serials_path/0
+        ]).
+
 -define(MNESIA_TABLE, rabbit_exchange).
 -define(MNESIA_DURABLE_TABLE, rabbit_durable_exchange).
 -define(MNESIA_SERIAL_TABLE, rabbit_exchange_serial).
@@ -772,87 +775,6 @@ exists_in_mnesia(Name) ->
 exists_in_khepri(Name) ->
     rabbit_khepri:exists(khepri_exchange_path(Name)).
 
-%% --------------------------------------------------------------
-%% Migration
-%% --------------------------------------------------------------
-
--spec mnesia_write_to_khepri(Table, [Queue]) -> ok when
-      Table :: atom(),
-      Queue :: amqqueue:amqqueue().
-%% Mnesia contains two tables if an exchange has been recovered:
-%% rabbit_exchange (ram) and rabbit_durable_exchange (disc).
-%% As all data in Khepri is persistent, there is no point on
-%% having ram and data entries.
-%% How do we then transform data from mnesia to khepri when
-%% the feature flag is enabled?
-%% Let's create the Khepri entry from the ram table.
-
-mnesia_write_to_khepri(rabbit_exchange, Exchanges) ->
-    rabbit_khepri:transaction(
-      fun() ->
-              [khepri_create_tx(khepri_exchange_path(Exchange#exchange.name), Exchange)
-               || Exchange <- Exchanges]
-      end, rw);                       
-mnesia_write_to_khepri(rabbit_durable_exchange, _Exchange0) ->
-    ok;
-mnesia_write_to_khepri(rabbit_exchange_serial, Exchanges) ->
-    rabbit_khepri:transaction(
-      fun() ->
-              [begin
-                   #exchange_serial{name = Resource, next = Serial} = Exchange,
-                   Path = khepri_path:combine_with_conditions(
-                            khepri_exchange_serial_path(Resource),
-                            [#if_node_exists{exists = false}]),
-                   case khepri_tx:put(Path, Serial) of
-                       ok -> ok;
-                       Error -> throw(Error)
-                   end
-               end || Exchange <- Exchanges]
-      end, rw).
-
-khepri_create_tx(Path, Value) ->
-    case khepri_tx:create(Path, Value) of
-        ok -> ok;
-        {error, {khepri, mismatching_node, _}} -> ok;
-        Error -> throw(Error)
-    end.
-
--spec mnesia_delete_to_khepri(Table, ToDelete) -> ok when
-      Table :: atom(),
-      ToDelete :: amqqueue:amqqueue() | rabbit_amqqueue:name().
-
-mnesia_delete_to_khepri(rabbit_exchange, Exchange) when is_record(Exchange, exchange) ->
-    khepri_delete(khepri_exchange_path(Exchange#exchange.name));
-mnesia_delete_to_khepri(rabbit_exchange, Name) ->
-    khepri_delete(khepri_exchange_path(Name));
-mnesia_delete_to_khepri(rabbit_durable_exchange, Exchange)
-  when is_record(Exchange, exchange) ->
-    khepri_delete(khepri_exchange_path(Exchange#exchange.name));
-mnesia_delete_to_khepri(rabbit_durable_exchange, Name) ->
-    khepri_delete(khepri_exchange_path(Name));
-mnesia_delete_to_khepri(rabbit_exchange_serial, ExchangeSerial)
-  when is_record(ExchangeSerial, exchange_serial) ->
-    khepri_delete(khepri_exchange_serial_path(ExchangeSerial#exchange_serial.name));
-mnesia_delete_to_khepri(rabbit_exchange_serial, Name) ->
-    khepri_delete(khepri_exchange_serial_path(Name)).
-
--spec clear_data_in_khepri(Table) -> ok when
-      Table :: atom().
-
-clear_data_in_khepri(rabbit_exchange) ->
-    khepri_delete(khepri_exchanges_path());
-%% There is a single khepri entry for exchanges and it should be already deleted
-clear_data_in_khepri(rabbit_durable_exchange) ->
-    ok;
-clear_data_in_khepri(rabbit_exchange_serial) ->
-    khepri_delete(khepri_exchange_serials_path()).
-
-khepri_delete(Path) ->
-    case rabbit_khepri:delete(Path) of
-        ok -> ok;
-        Error -> throw(Error)
-    end.
-
 %% -------------------------------------------------------------------
 %% clear().
 %% -------------------------------------------------------------------
@@ -878,6 +800,12 @@ clear_in_khepri() ->
     khepri_delete(khepri_exchanges_path()),
     khepri_delete(khepri_exchange_serials_path()).    
 
+khepri_delete(Path) ->
+    case rabbit_khepri:delete(Path) of
+        ok -> ok;
+        Error -> throw(Error)
+    end.
+
 %% -------------------------------------------------------------------
 %% maybe_auto_delete_in_mnesia().
 %% -------------------------------------------------------------------
diff --git a/deps/rabbit/src/rabbit_db_exchange_m2k_converter.erl b/deps/rabbit/src/rabbit_db_exchange_m2k_converter.erl
new file mode 100644
index 0000000000..88cad76487
--- /dev/null
+++ b/deps/rabbit/src/rabbit_db_exchange_m2k_converter.erl
@@ -0,0 +1,136 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright © 2022-2023 VMware, Inc. or its affiliates.  All rights reserved.
+%%
+
+-module(rabbit_db_exchange_m2k_converter).
+
+-behaviour(mnesia_to_khepri_converter).
+
+-include_lib("kernel/include/logger.hrl").
+-include_lib("khepri/include/khepri.hrl").
+-include_lib("khepri_mnesia_migration/src/kmm_logging.hrl").
+-include_lib("rabbit_common/include/rabbit.hrl").
+
+-export([init_copy_to_khepri/2,
+         copy_to_khepri/3,
+         delete_from_khepri/3,
+         clear_data_in_khepri/1]).
+
+-record(?MODULE, {store_id :: khepri:store_id()}).
+
+-spec init_copy_to_khepri(Tables, StoreId) -> Ret when
+      Tables :: [mnesia_to_khepri:mnesia_table()],
+      StoreId :: khepri:store_id(),
+      Ret :: {ok, Priv},
+      Priv :: #?MODULE{}.
+%% @private
+
+init_copy_to_khepri(_Tables, StoreId) ->
+    State = #?MODULE{store_id = StoreId},
+    {ok, State}.
+
+-spec copy_to_khepri(Table, Record, Priv) -> Ret when
+      Table :: mnesia_to_khepri:mnesia_table(),
+      Record :: tuple(),
+      Priv :: #?MODULE{},
+      Ret :: {ok, NewPriv} | {error, Reason},
+      NewPriv :: #?MODULE{},
+      Reason :: any().
+%% @private
+
+copy_to_khepri(
+  rabbit_exchange = Table, Record,
+  #?MODULE{store_id = StoreId} = State) ->
+    Name = Record#exchange.name,
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data copy: [~0p] key: ~0p",
+       [Table, Name],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    Path = rabbit_db_exchange:khepri_exchange_path(Name),
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data copy: [~0p] path: ~0p",
+       [Table, Path],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    case khepri:put(StoreId, Path, Record) of
+        ok    -> {ok, State};
+        Error -> Error
+    end;
+copy_to_khepri(rabbit_exchange_serial = Table,
+               #exchange_serial{name = Resource, next = Serial},
+               #?MODULE{store_id = StoreId} = State) ->
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data copy: [~0p] key: ~0p",
+       [Table, Resource],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    Path = khepri_path:combine_with_conditions(
+             rabbit_db_exchange:khepri_exchange_serial_path(Resource),
+             [#if_node_exists{exists = false}]),
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data copy: [~0p] path: ~0p",
+       [Table, Path],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    case khepri:put(StoreId, Path, Serial) of
+        ok    -> {ok, State};
+        Error -> Error
+    end;
+copy_to_khepri(Table, Record, State) ->
+    ?LOG_DEBUG("Mnesia->Khepri unexpected record table ~0p record ~0p state ~0p",
+               [Table, Record, State]),
+    {error, unexpected_record}.
+
+-spec delete_from_khepri(Table, Key, Priv) -> Ret when
+      Table :: mnesia_to_khepri:mnesia_table(),
+      Key :: any(),
+      Priv :: #?MODULE{},
+      Ret :: {ok, NewPriv} | {error, Reason},
+      NewPriv :: #?MODULE{},
+      Reason :: any().
+%% @private
+
+delete_from_khepri(rabbit_exchange = Table, Key,
+                   #?MODULE{store_id = StoreId} = State) ->
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data delete: [~0p] key: ~0p",
+       [Table, Key],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    Path = rabbit_db_exchange:khepri_exchange_path(Key),
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data delete: [~0p] path: ~0p",
+       [Table, Path],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    case khepri:delete(StoreId, Path) of
+        ok    -> {ok, State};
+        Error -> Error
+    end;
+delete_from_khepri(rabbit_exchange_serial = Table, Key,
+                   #?MODULE{store_id = StoreId} = State) ->
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data delete: [~0p] key: ~0p",
+       [Table, Key],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    Path = rabbit_db_exchange:khepri_exchange_serial_path(Key),
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data delete: [~0p] path: ~0p",
+       [Table, Path],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    case khepri:delete(StoreId, Path) of
+        ok    -> {ok, State};
+        Error -> Error
+    end.
+
+-spec clear_data_in_khepri(Table) -> ok when
+      Table :: atom().
+
+clear_data_in_khepri(rabbit_exchange) ->
+    khepri_delete(rabbit_db_exchange:khepri_exchanges_path());
+clear_data_in_khepri(rabbit_exchange_serial) ->
+    khepri_delete(rabbit_db_exchange:khepri_exchange_serials_path()).
+
+khepri_delete(Path) ->
+    case rabbit_khepri:delete(Path) of
+        ok -> ok;
+        Error -> throw(Error)
+    end.
diff --git a/deps/rabbit/src/rabbit_db_maintenance.erl b/deps/rabbit/src/rabbit_db_maintenance.erl
index 7c6145d42b..dd336a8b8d 100644
--- a/deps/rabbit/src/rabbit_db_maintenance.erl
+++ b/deps/rabbit/src/rabbit_db_maintenance.erl
@@ -16,9 +16,10 @@
          get_consistent/1
         ]).
 
--export([mnesia_write_to_khepri/2,
-         mnesia_delete_to_khepri/2,
-         clear_data_in_khepri/1]).
+-export([
+         khepri_maintenance_path/1,
+         khepri_maintenance_path/0
+        ]).
 
 -define(TABLE, rabbit_node_maintenance_states).
 
@@ -162,39 +163,6 @@ get_consistent_in_khepri(Node) ->
     end.
 
 %% -------------------------------------------------------------------
-%% migration
-%% -------------------------------------------------------------------
-
-mnesia_write_to_khepri(?TABLE, States) ->
-    rabbit_khepri:transaction(
-      fun() ->
-              [khepri_create_tx(khepri_maintenance_path(State#node_maintenance_state.node),
-                                State)
-               || State <- States]
-      end, rw).
-
-mnesia_delete_to_khepri(?TABLE, State) when is_record(State, node_maintenance_state) ->
-    khepri_delete(khepri_maintenance_path(State#node_maintenance_state.node));
-mnesia_delete_to_khepri(?TABLE, Node) ->
-    khepri_delete(khepri_maintenance_path(Node)).
-
-clear_data_in_khepri(?TABLE) ->
-    khepri_delete(khepri_maintenance_path()).
-
-khepri_delete(Path) ->
-    case rabbit_khepri:delete(Path) of
-        ok -> ok;
-        Error -> throw(Error)
-    end.
-
-khepri_create_tx(Path, Value) ->
-    case khepri_tx:create(Path, Value) of
-        ok -> ok;
-        {error, {khepri, mismatching_node, _}} -> ok;
-        Error -> throw(Error)
-    end.
-
-%% -------------------------------------------------------------------
 %% Khepri paths
 %% -------------------------------------------------------------------
 
diff --git a/deps/rabbit/src/rabbit_db_maintenance_m2k_converter.erl b/deps/rabbit/src/rabbit_db_maintenance_m2k_converter.erl
new file mode 100644
index 0000000000..f5be468b73
--- /dev/null
+++ b/deps/rabbit/src/rabbit_db_maintenance_m2k_converter.erl
@@ -0,0 +1,102 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright © 2022-2023 VMware, Inc. or its affiliates.  All rights reserved.
+%%
+
+-module(rabbit_db_maintenance_m2k_converter).
+
+-behaviour(mnesia_to_khepri_converter).
+
+-include_lib("kernel/include/logger.hrl").
+-include_lib("khepri/include/khepri.hrl").
+-include_lib("khepri_mnesia_migration/src/kmm_logging.hrl").
+-include_lib("rabbit_common/include/rabbit.hrl").
+
+-export([
+         init_copy_to_khepri/2,
+         copy_to_khepri/3,
+         delete_from_khepri/3,
+         clear_data_in_khepri/1
+        ]).
+
+-record(?MODULE, {store_id :: khepri:store_id()}).
+
+-spec init_copy_to_khepri(Tables, StoreId) -> Ret when
+      Tables :: [mnesia_to_khepri:mnesia_table()],
+      StoreId :: khepri:store_id(),
+      Ret :: {ok, Priv},
+      Priv :: #?MODULE{}.
+%% @private
+
+init_copy_to_khepri(_Tables, StoreId) ->
+    State = #?MODULE{store_id = StoreId},
+    {ok, State}.
+
+-spec copy_to_khepri(Table, Record, Priv) -> Ret when
+      Table :: mnesia_to_khepri:mnesia_table(),
+      Record :: tuple(),
+      Priv :: #?MODULE{},
+      Ret :: {ok, NewPriv} | {error, Reason},
+      NewPriv :: #?MODULE{},
+      Reason :: any().
+%% @private
+
+copy_to_khepri(
+  rabbit_node_maintenance_states = Table, Record,
+  #?MODULE{store_id = StoreId} = State) ->
+    Name = Record#node_maintenance_state.node,
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data copy: [~0p] key: ~0p",
+       [Table, Name],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    Path = rabbit_db_maintenance:khepri_maintenance_path(Name),
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data copy: [~0p] path: ~0p",
+       [Table, Path],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    case khepri:put(StoreId, Path, Record) of
+        ok    -> {ok, State};
+        Error -> Error
+    end;
+copy_to_khepri(Table, Record, State) ->
+    ?LOG_DEBUG("Mnesia->Khepri unexpected record table ~0p record ~0p state ~0p",
+               [Table, Record, State]),
+    {error, unexpected_record}.
+
+-spec delete_from_khepri(Table, Key, Priv) -> Ret when
+      Table :: mnesia_to_khepri:mnesia_table(),
+      Key :: any(),
+      Priv :: #?MODULE{},
+      Ret :: {ok, NewPriv} | {error, Reason},
+      NewPriv :: #?MODULE{},
+      Reason :: any().
+%% @private
+
+delete_from_khepri(
+  rabbit_node_maintenance_states = Table, Key,
+  #?MODULE{store_id = StoreId} = State) ->
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data delete: [~0p] key: ~0p",
+       [Table, Key],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    Path = rabbit_db_maintenance:khepri_maintenance_path(Key),
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data delete: [~0p] path: ~0p",
+       [Table, Path],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    case khepri:delete(StoreId, Path) of
+        ok    -> {ok, State};
+        Error -> Error
+    end.
+
+-spec clear_data_in_khepri(Table) -> ok when
+      Table :: atom().
+
+clear_data_in_khepri(rabbit_node_maintenance_states) ->
+    Path = rabbit_db_maintenance:khepri_maintenance_path(),
+    case rabbit_khepri:delete(Path) of
+        ok -> ok;
+        Error -> throw(Error)
+    end.
diff --git a/deps/rabbit/src/rabbit_db_msup.erl b/deps/rabbit/src/rabbit_db_msup.erl
index 58fb70ab9e..24e1e951a7 100644
--- a/deps/rabbit/src/rabbit_db_msup.erl
+++ b/deps/rabbit/src/rabbit_db_msup.erl
@@ -8,6 +8,7 @@
 -module(rabbit_db_msup).
 
 -include_lib("khepri/include/khepri.hrl").
+-include("mirrored_supervisor.hrl").
 
 -export([
          create_tables/0,
@@ -21,9 +22,10 @@
 
 -export([clear/0]).
 
--export([mnesia_write_to_khepri/2,
-         mnesia_delete_to_khepri/2,
-         clear_data_in_khepri/1]).
+-export([
+         khepri_mirrored_supervisor_path/2,
+         khepri_mirrored_supervisor_path/0
+        ]).
 
 -define(TABLE, mirrored_sup_childspec).
 -define(TABLE_DEF,
@@ -33,8 +35,6 @@
           {attributes, record_info(fields, mirrored_sup_childspec)}]}).
 -define(TABLE_MATCH, {match, #mirrored_sup_childspec{ _ = '_' }}).
 
--record(mirrored_sup_childspec, {key, mirroring_pid, childspec}).
-
 %% -------------------------------------------------------------------
 %% create_tables().
 %% -------------------------------------------------------------------
@@ -311,42 +311,13 @@ clear_in_mnesia() ->
     ok.
 
 clear_in_khepri() ->
-    khepri_delete(khepri_mirrored_supervisor_path()).
-
-khepri_delete(Path) ->
+    Path = khepri_mirrored_supervisor_path(),
     case rabbit_khepri:delete(Path) of
         ok -> ok;
         Error -> throw(Error)
     end.
 
 %% -------------------------------------------------------------------
-%% migration
-%% -------------------------------------------------------------------
-
-mnesia_write_to_khepri(?TABLE, Specs) ->
-    rabbit_khepri:transaction(
-      fun() ->
-              [khepri_create_tx(khepri_mirrored_supervisor_path(Group, SimpleId),
-                                Spec)
-               || #mirrored_sup_childspec{key = {Group, {SimpleId, _}}} = Spec <- Specs]
-      end, rw).
-
-mnesia_delete_to_khepri(?TABLE, #mirrored_sup_childspec{key = {Group, Id}}) ->
-    khepri_delete(khepri_mirrored_supervisor_path(Group, Id));
-mnesia_delete_to_khepri(?TABLE, {Group, Id}) ->
-    khepri_delete(khepri_mirrored_supervisor_path(Group, Id)).
-
-clear_data_in_khepri(?TABLE) ->
-    khepri_delete(khepri_mirrored_supervisor_path()).
-
-khepri_create_tx(Path, Value) ->
-    case khepri_tx:create(Path, Value) of
-        ok -> ok;
-        {error, {khepri, mismatching_node, _}} -> ok;
-        Error -> throw(Error)
-    end.
-
-%% -------------------------------------------------------------------
 %% Khepri paths
 %% -------------------------------------------------------------------
 
diff --git a/deps/rabbit/src/rabbit_db_msup_m2k_converter.erl b/deps/rabbit/src/rabbit_db_msup_m2k_converter.erl
new file mode 100644
index 0000000000..0c91781450
--- /dev/null
+++ b/deps/rabbit/src/rabbit_db_msup_m2k_converter.erl
@@ -0,0 +1,100 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright © 2022-2023 VMware, Inc. or its affiliates.  All rights reserved.
+%%
+
+-module(rabbit_db_msup_m2k_converter).
+
+-behaviour(mnesia_to_khepri_converter).
+
+-include_lib("kernel/include/logger.hrl").
+-include_lib("khepri/include/khepri.hrl").
+-include_lib("khepri_mnesia_migration/src/kmm_logging.hrl").
+-include_lib("rabbit_common/include/rabbit.hrl").
+-include("mirrored_supervisor.hrl").
+
+-export([init_copy_to_khepri/2,
+         copy_to_khepri/3,
+         delete_from_khepri/3,
+         clear_data_in_khepri/1]).
+
+-record(?MODULE, {store_id :: khepri:store_id()}).
+
+-spec init_copy_to_khepri(Tables, StoreId) -> Ret when
+      Tables :: [mnesia_to_khepri:mnesia_table()],
+      StoreId :: khepri:store_id(),
+      Ret :: {ok, Priv},
+      Priv :: #?MODULE{}.
+%% @private
+
+init_copy_to_khepri(_Tables, StoreId) ->
+    State = #?MODULE{store_id = StoreId},
+    {ok, State}.
+
+-spec copy_to_khepri(Table, Record, Priv) -> Ret when
+      Table :: mnesia_to_khepri:mnesia_table(),
+      Record :: tuple(),
+      Priv :: #?MODULE{},
+      Ret :: {ok, NewPriv} | {error, Reason},
+      NewPriv :: #?MODULE{},
+      Reason :: any().
+%% @private
+
+copy_to_khepri(mirrored_sup_childspec = Table,
+               #mirrored_sup_childspec{key = {Group, {SimpleId, _}} = Key} = Record,
+               #?MODULE{store_id = StoreId} = State) ->
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data copy: [~0p] key: ~0p",
+       [Table, Key],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    Path = rabbit_db_msup:khepri_mirrored_supervisor_path(Group, SimpleId),
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data copy: [~0p] path: ~0p",
+       [Table, Path],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    case khepri:put(StoreId, Path, Record) of
+        ok    -> {ok, State};
+        Error -> Error
+    end;
+copy_to_khepri(Table, Record, State) ->
+    ?LOG_DEBUG("Mnesia->Khepri unexpected record table ~0p record ~0p state ~0p",
+               [Table, Record, State]),
+    {error, unexpected_record}.
+
+-spec delete_from_khepri(Table, Key, Priv) -> Ret when
+      Table :: mnesia_to_khepri:mnesia_table(),
+      Key :: any(),
+      Priv :: #?MODULE{},
+      Ret :: {ok, NewPriv} | {error, Reason},
+      NewPriv :: #?MODULE{},
+      Reason :: any().
+%% @private
+
+delete_from_khepri(
+  mirrored_sup_childspec = Table, {Group, Id} = Key,
+  #?MODULE{store_id = StoreId} = State) ->
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data delete: [~0p] key: ~0p",
+       [Table, Key],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    Path = rabbit_db_msup:khepri_mirrored_supervisor_path(Group, Id),
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data delete: [~0p] path: ~0p",
+       [Table, Path],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    case khepri:delete(StoreId, Path) of
+        ok    -> {ok, State};
+        Error -> Error
+    end.
+
+-spec clear_data_in_khepri(Table) -> ok when
+      Table :: atom().
+
+clear_data_in_khepri(mirrored_sup_childspec) ->
+    Path = rabbit_db_msup:khepri_mirrored_supervisor_path(),
+    case rabbit_khepri:delete(Path) of
+        ok -> ok;
+        Error -> throw(Error)
+    end.
diff --git a/deps/rabbit/src/rabbit_db_queue.erl b/deps/rabbit/src/rabbit_db_queue.erl
index f5f6438f03..c17efcf45a 100644
--- a/deps/rabbit/src/rabbit_db_queue.erl
+++ b/deps/rabbit/src/rabbit_db_queue.erl
@@ -67,13 +67,14 @@
          get_in_khepri_tx/1
         ]).
 
--export([mnesia_write_to_khepri/2,
-         mnesia_delete_to_khepri/2,
-         clear_data_in_khepri/1]).
-
 %% For testing
 -export([clear/0]).
 
+-export([
+         khepri_queue_path/1,
+         khepri_queues_path/0
+        ]).
+
 -define(MNESIA_TABLE, rabbit_queue).
 -define(MNESIA_DURABLE_TABLE, rabbit_durable_queue).
 
@@ -1103,60 +1104,13 @@ clear_in_mnesia() ->
     ok.
 
 clear_in_khepri() ->
-    khepri_delete(khepri_queues_path()).
-
-khepri_delete(Path) ->
+    Path = khepri_queues_path(),
     case rabbit_khepri:delete(Path) of
         ok -> ok;
         Error -> throw(Error)
     end.
 
 %% --------------------------------------------------------------
-%% Feature flags
-%% --------------------------------------------------------------
-
--spec mnesia_write_to_khepri(Table, [Queue]) -> ok when
-      Table :: atom(),
-      Queue :: amqqueue:amqqueue().
-
-mnesia_write_to_khepri(rabbit_queue, Qs) ->
-    rabbit_khepri:transaction(
-      fun() ->
-              [begin
-                   Path = khepri_queue_path(amqqueue:get_name(Q)),
-                   case khepri_tx:create(Path, Q) of
-                       ok -> ok;
-                       {error, {khepri, mismatching_node, _}} -> ok;
-                       Error -> throw(Error)
-                   end
-               end || Q <- Qs]
-      end, rw);
-mnesia_write_to_khepri(rabbit_durable_queue, _Qs) ->
-    %% All durable queues are on the `rabbit_queue` table too
-    ok.
-
--spec mnesia_delete_to_khepri(Table, ToDelete) -> ok when
-      Table :: atom(),
-      ToDelete :: amqqueue:amqqueue() | rabbit_amqqueue:name().
-
-mnesia_delete_to_khepri(rabbit_queue, Q) when ?is_amqqueue(Q) ->
-    khepri_delete(khepri_queue_path(amqqueue:get_name(Q)));
-mnesia_delete_to_khepri(rabbit_queue, Name) when is_record(Name, resource) ->
-    khepri_delete(khepri_queue_path(Name));
-mnesia_delete_to_khepri(rabbit_durable_queue, Q) when ?is_amqqueue(Q) ->
-    khepri_delete(khepri_queue_path(amqqueue:get_name(Q)));
-mnesia_delete_to_khepri(rabbit_durable_queue, Name) when is_record(Name, resource) ->
-    khepri_delete(khepri_queue_path(Name)).
-
--spec clear_data_in_khepri(Table) -> ok when
-      Table :: atom().
-
-clear_data_in_khepri(rabbit_queue) ->
-    khepri_delete(khepri_queues_path());
-clear_data_in_khepri(rabbit_durable_queue) ->
-    khepri_delete(khepri_queues_path()).
-
-%% --------------------------------------------------------------
 %% Internal
 %% --------------------------------------------------------------
 
diff --git a/deps/rabbit/src/rabbit_db_queue_m2k_converter.erl b/deps/rabbit/src/rabbit_db_queue_m2k_converter.erl
new file mode 100644
index 0000000000..bac29d25ad
--- /dev/null
+++ b/deps/rabbit/src/rabbit_db_queue_m2k_converter.erl
@@ -0,0 +1,102 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright © 2022-2023 VMware, Inc. or its affiliates.  All rights reserved.
+%%
+
+-module(rabbit_db_queue_m2k_converter).
+
+-behaviour(mnesia_to_khepri_converter).
+
+-include_lib("kernel/include/logger.hrl").
+-include_lib("khepri/include/khepri.hrl").
+-include_lib("khepri_mnesia_migration/src/kmm_logging.hrl").
+-include_lib("rabbit_common/include/rabbit.hrl").
+
+-export([init_copy_to_khepri/2,
+         copy_to_khepri/3,
+         delete_from_khepri/3,
+         clear_data_in_khepri/1]).
+
+-record(?MODULE, {store_id :: khepri:store_id()}).
+
+-spec init_copy_to_khepri(Tables, StoreId) -> Ret when
+      Tables :: [mnesia_to_khepri:mnesia_table()],
+      StoreId :: khepri:store_id(),
+      Ret :: {ok, Priv},
+      Priv :: #?MODULE{}.
+%% @private
+
+init_copy_to_khepri(_Tables, StoreId) ->
+    State = #?MODULE{store_id = StoreId},
+    {ok, State}.
+
+-spec copy_to_khepri(Table, Record, Priv) -> Ret when
+      Table :: mnesia_to_khepri:mnesia_table(),
+      Record :: tuple(),
+      Priv :: #?MODULE{},
+      Ret :: {ok, NewPriv} | {error, Reason},
+      NewPriv :: #?MODULE{},
+      Reason :: any().
+%% @private
+
+copy_to_khepri(rabbit_queue = Table, Record,
+               #?MODULE{store_id = StoreId} = State) ->
+    Name = amqqueue:get_name(Record),
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data copy: [~0p] key: ~0p",
+       [Table, Name],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    Path = rabbit_db_queue:khepri_queue_path(Name),
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data copy: [~0p] path: ~0p",
+       [Table, Path],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    case khepri:put(StoreId, Path, Record) of
+        ok    -> {ok, State};
+        Error -> Error
+    end;
+copy_to_khepri(Table, Record, State) ->
+    ?LOG_DEBUG("Mnesia->Khepri unexpected record table ~0p record ~0p state ~0p",
+               [Table, Record, State]),
+    {error, unexpected_record}.
+
+-spec delete_from_khepri(Table, Key, Priv) -> Ret when
+      Table :: mnesia_to_khepri:mnesia_table(),
+      Key :: any(),
+      Priv :: #?MODULE{},
+      Ret :: {ok, NewPriv} | {error, Reason},
+      NewPriv :: #?MODULE{},
+      Reason :: any().
+%% @private
+
+delete_from_khepri(rabbit_queue = Table, Key,
+                   #?MODULE{store_id = StoreId} = State) ->
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data delete: [~0p] key: ~0p",
+       [Table, Key],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    Path = rabbit_db_queue:khepri_queue_path(Key),
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data delete: [~0p] path: ~0p",
+       [Table, Path],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    case khepri:delete(StoreId, Path) of
+        ok    -> {ok, State};
+        Error -> Error
+    end.
+
+-spec clear_data_in_khepri(Table) -> ok when
+      Table :: atom().
+
+clear_data_in_khepri(rabbit_queue) ->
+    khepri_delete(rabbit_db_queue:khepri_queues_path());
+clear_data_in_khepri(rabbit_durable_queue) ->
+    khepri_delete(rabbit_db_queue:khepri_queues_path()).
+
+khepri_delete(Path) ->
+    case rabbit_khepri:delete(Path) of
+        ok -> ok;
+        Error -> throw(Error)
+    end.
diff --git a/deps/rabbit/src/rabbit_db_rtparams.erl b/deps/rabbit/src/rabbit_db_rtparams.erl
index 9b8951892b..5479059be5 100644
--- a/deps/rabbit/src/rabbit_db_rtparams.erl
+++ b/deps/rabbit/src/rabbit_db_rtparams.erl
@@ -16,9 +16,10 @@
          get_all/0, get_all/2,
          delete/1, delete/3]).
 
--export([clear_data_in_khepri/1,
-         mnesia_write_to_khepri/2,
-         mnesia_delete_to_khepri/2]).
+-export([khepri_vhost_rp_path/3,
+         khepri_global_rp_path/1,
+         khepri_rp_path/0
+        ]).
 
 -define(MNESIA_TABLE, rabbit_runtime_parameters).
 -define(any(Value), case Value of
@@ -360,46 +361,6 @@ delete_matching_in_khepri(VHostName, Comp, Name) ->
     Key = {?any(VHostName), ?any(Comp), ?any(Name)},
     delete_in_khepri(Key).
 
-clear_data_in_khepri(rabbit_runtime_parameters) ->
-    Path = khepri_rp_path(),
-    case rabbit_khepri:delete(Path) of
-        ok -> ok;
-        Error -> throw(Error)
-    end.
-
-mnesia_write_to_khepri(rabbit_runtime_parameters, RuntimeParams) ->
-    rabbit_khepri:transaction(
-      fun() ->
-              lists:foreach(fun(#runtime_parameters{key = {VHost, Comp, Name}} = RuntimeParam) ->
-                                    Path = khepri_vhost_rp_path(VHost, Comp, Name),
-                                    case khepri_tx:put(Path, RuntimeParam) of
-                                        ok -> ok;
-                                        Error -> throw(Error)
-                                    end;
-                               (#runtime_parameters{key = Key} = RuntimeParam) ->
-                                    Path = khepri_global_rp_path(Key),
-                                    case khepri_tx:put(Path, RuntimeParam) of
-                                        ok -> ok;
-                                        Error -> throw(Error)
-                                    end
-                            end, RuntimeParams)
-      end).
-
-mnesia_delete_to_khepri(rabbit_runtime_parameters,
-  #runtime_parameters{key = {VHost, Comp, Name}}) ->
-    Path = khepri_vhost_rp_path(VHost, Comp, Name),
-    case rabbit_khepri:delete(Path) of
-        ok    -> ok;
-        Error -> throw(Error)
-    end;
-mnesia_delete_to_khepri(rabbit_runtime_parameters,
-  #runtime_parameters{key = Key}) ->
-    Path = khepri_global_rp_path(Key),
-    case rabbit_khepri:delete(Path) of
-        ok    -> ok;
-        Error -> throw(Error)
-    end.
-
 khepri_rp_path() ->
     [?MODULE].
 
diff --git a/deps/rabbit/src/rabbit_db_rtparams_m2k_converter.erl b/deps/rabbit/src/rabbit_db_rtparams_m2k_converter.erl
new file mode 100644
index 0000000000..caec2e7d97
--- /dev/null
+++ b/deps/rabbit/src/rabbit_db_rtparams_m2k_converter.erl
@@ -0,0 +1,104 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright © 2022-2023 VMware, Inc. or its affiliates.  All rights reserved.
+%%
+
+-module(rabbit_db_rtparams_m2k_converter).
+
+-behaviour(mnesia_to_khepri_converter).
+
+-include_lib("kernel/include/logger.hrl").
+-include_lib("khepri/include/khepri.hrl").
+-include_lib("khepri_mnesia_migration/src/kmm_logging.hrl").
+-include_lib("rabbit_common/include/rabbit.hrl").
+
+-export([init_copy_to_khepri/2,
+         copy_to_khepri/3,
+         delete_from_khepri/3,
+         clear_data_in_khepri/1]).
+
+-record(?MODULE, {store_id :: khepri:store_id()}).
+
+-spec init_copy_to_khepri(Tables, StoreId) -> Ret when
+      Tables :: [mnesia_to_khepri:mnesia_table()],
+      StoreId :: khepri:store_id(),
+      Ret :: {ok, Priv},
+      Priv :: #?MODULE{}.
+%% @private
+
+init_copy_to_khepri(_Tables, StoreId) ->
+    State = #?MODULE{store_id = StoreId},
+    {ok, State}.
+
+-spec copy_to_khepri(Table, Record, Priv) -> Ret when
+      Table :: mnesia_to_khepri:mnesia_table(),
+      Record :: tuple(),
+      Priv :: #?MODULE{},
+      Ret :: {ok, NewPriv} | {error, Reason},
+      NewPriv :: #?MODULE{},
+      Reason :: any().
+%% @private
+
+copy_to_khepri(
+  rabbit_runtime_parameters = Table, #runtime_parameters{key = Key} = Record,
+  #?MODULE{store_id = StoreId} = State) ->
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data copy: [~0p] key: ~0p",
+       [Table, Key],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    Path = rtparams_path(Key),
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data copy: [~0p] path: ~0p",
+       [Table, Path],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    case khepri:put(StoreId, Path, Record) of
+        ok    -> {ok, State};
+        Error -> Error
+    end;
+copy_to_khepri(Table, Record, State) ->
+    ?LOG_DEBUG("Mnesia->Khepri unexpected record table ~0p record ~0p state ~0p",
+               [Table, Record, State]),
+    {error, unexpected_record}.
+
+-spec delete_from_khepri(Table, Key, Priv) -> Ret when
+      Table :: mnesia_to_khepri:mnesia_table(),
+      Key :: any(),
+      Priv :: #?MODULE{},
+      Ret :: {ok, NewPriv} | {error, Reason},
+      NewPriv :: #?MODULE{},
+      Reason :: any().
+%% @private
+
+delete_from_khepri(
+  rabbit_runtime_parameters = Table, Key,
+  #?MODULE{store_id = StoreId} = State) ->
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data delete: [~0p] key: ~0p",
+       [Table, Key],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    Path = rtparams_path(Key),
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data delete: [~0p] path: ~0p",
+       [Table, Path],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    case khepri:delete(StoreId, Path) of
+        ok    -> {ok, State};
+        Error -> Error
+    end.
+
+rtparams_path({VHost, Comp, Name})->
+    rabbit_db_rtparams:khepri_vhost_rp_path(VHost, Comp, Name);
+rtparams_path(Key) ->
+    rabbit_db_rtparams:khepri_global_rp_path(Key).
+
+-spec clear_data_in_khepri(Table) -> ok when
+      Table :: atom().
+
+clear_data_in_khepri(rabbit_runtime_parameters) ->
+    Path = rabbit_db_rtparams:khepri_rp_path(),
+    case rabbit_khepri:delete(Path) of
+        ok -> ok;
+        Error -> throw(Error)
+    end.
diff --git a/deps/rabbit/src/rabbit_db_topic_exchange.erl b/deps/rabbit/src/rabbit_db_topic_exchange.erl
index 09fb41ba99..e783623334 100644
--- a/deps/rabbit/src/rabbit_db_topic_exchange.erl
+++ b/deps/rabbit/src/rabbit_db_topic_exchange.erl
@@ -15,12 +15,6 @@
          delete/1,
          match/2]).
 
--export([
-         mnesia_write_to_khepri/2,
-         mnesia_delete_to_khepri/2,
-         clear_data_in_khepri/1
-        ]).
-
 %% These functions are used to process mnesia deletion events generated during the
 %% migration from mnesia to khepri
 -export([
@@ -152,22 +146,6 @@ match_in_khepri(XName, RoutingKey) ->
     Words = split_topic_key(RoutingKey),
     trie_match_in_khepri(XName, Words).
 
-%% --------------------------------------------------------------
-%% Migration
-%% --------------------------------------------------------------
-
-%% Nothing to do, the `rabbit_db_binding' migration creates the topic
-%% graph projection in Khepri which replaces this data.
-
-mnesia_write_to_khepri(_Table, _Records) ->
-    ok.
-
-mnesia_delete_to_khepri(_Table, _Record) ->
-    ok.
-
-clear_data_in_khepri(_Table) ->
-    ok.
-
 %% -------------------------------------------------------------------
 %% clear().
 %% -------------------------------------------------------------------
diff --git a/deps/rabbit/src/rabbit_db_user.erl b/deps/rabbit/src/rabbit_db_user.erl
index 06c110be95..1b1cb75b6d 100644
--- a/deps/rabbit/src/rabbit_db_user.erl
+++ b/deps/rabbit/src/rabbit_db_user.erl
@@ -32,11 +32,10 @@
          clear_matching_topic_permissions/3,
          delete/1]).
 
--export([clear_data_in_khepri/1,
-         mnesia_write_to_khepri/2,
-         mnesia_delete_to_khepri/2]).
-
--export([khepri_users_path/0]).
+-export([khepri_users_path/0,
+         khepri_user_path/1,
+         khepri_user_permission_path/2,
+         khepri_topic_permission_path/3]).
 
 %% for testing
 -export([clear/0]).
@@ -994,113 +993,7 @@ clear_in_mnesia() ->
     ok.
 
 clear_in_khepri() ->
-    khepri_delete(khepri_users_path()).
-
-%% -------------------------------------------------------------------
-%% Migration
-%% -------------------------------------------------------------------
-
--spec mnesia_write_to_khepri(Table, [Queue]) -> ok when
-      Table :: atom(),
-      Queue :: amqqueue:amqqueue().
-
-mnesia_write_to_khepri(rabbit_user, Users) ->
-    rabbit_khepri:transaction(
-      fun() ->
-              [begin
-                   Username = internal_user:get_username(User),
-                   Path = khepri_user_path(Username),
-                   case khepri_tx:put(Path, User) of
-                       ok    -> ok;
-                       Error -> throw(Error)
-                   end
-               end || User <- Users, ?is_internal_user(User)]
-      end, rw);
-mnesia_write_to_khepri(rabbit_user_permission, UserPermissions) ->
-    rabbit_khepri:transaction(
-      fun() ->
-              [begin
-                   Path = khepri_user_permission_path(
-                            #if_all{conditions =
-                                        [Username,
-                                         #if_node_exists{exists = true}]},
-                            VHost),
-                   Extra = #{keep_while =>
-                                 #{rabbit_db_vhost:khepri_vhost_path(VHost) =>
-                                       #if_node_exists{exists = true}}},
-                   case khepri_tx:put(Path, UserPermission, Extra) of
-                       ok      -> ok;
-                       Error   -> throw(Error)
-                   end
-               end || #user_permission{
-                         user_vhost = #user_vhost{
-                                         username = Username,
-                                         virtual_host = VHost}} = UserPermission <- UserPermissions]
-      end, rw);
-mnesia_write_to_khepri(rabbit_topic_permission, TopicPermissions) ->
-    rabbit_khepri:transaction(
-      fun() ->
-              [begin
-                   Path = khepri_topic_permission_path(
-                            #if_all{conditions =
-                                        [Username,
-                                         #if_node_exists{exists = true}]},
-                            VHost,
-                            Exchange),
-                   Extra = #{keep_while =>
-                                 #{rabbit_db_vhost:khepri_vhost_path(VHost) =>
-                                       #if_node_exists{exists = true}}},
-                   case khepri_tx:put(Path, TopicPermission, Extra) of
-                       ok      -> ok;
-                       Error   -> throw(Error)
-                   end
-               end || #topic_permission{
-                         topic_permission_key =
-                             #topic_permission_key{
-                                user_vhost = #user_vhost{
-                                                username = Username,
-                                                virtual_host = VHost},
-                                exchange = Exchange}} = TopicPermission <- TopicPermissions]
-      end, rw).
-
--spec mnesia_delete_to_khepri(Table, ToDelete) -> ok when
-      Table :: atom(),
-      ToDelete :: amqqueue:amqqueue() | rabbit_amqqueue:name().
-
-mnesia_delete_to_khepri(rabbit_user, Record) ->
-    %% The record and the Mnesia table have different names.
-    User = erlang:setelement(1, Record, internal_user),
-    true = ?is_internal_user(User),
-    Username = internal_user:get_username(User),
-    Path = khepri_user_path(Username),
-    khepri_delete(Path);
-mnesia_delete_to_khepri(rabbit_user_permission,
-                        #user_permission{
-                           user_vhost = #user_vhost{
-                                           username = Username,
-                                           virtual_host = VHost}}) ->
-    Path = khepri_user_permission_path(Username, VHost),
-    khepri_delete(Path);
-mnesia_delete_to_khepri(rabbit_topic_permission,
-                        #topic_permission{
-                           topic_permission_key =
-                               #topic_permission_key{
-                                  user_vhost = #user_vhost{
-                                                  username = Username,
-                                                  virtual_host = VHost},
-                                  exchange = Exchange}}) ->
-    Path = khepri_topic_permission_path(Username, VHost, Exchange),
-    khepri_delete(Path).
-
--spec clear_data_in_khepri(Table) -> ok when
-      Table :: atom().
-
-clear_data_in_khepri(rabbit_user) ->
-    khepri_delete(khepri_users_path());
-clear_data_in_khepri(_) ->
-    ok.
-
-khepri_delete(Path) ->
+    Path = khepri_users_path(),
     case rabbit_khepri:delete(Path) of
         ok    -> ok;
         Error -> throw(Error)
diff --git a/deps/rabbit/src/rabbit_db_user_m2k_converter.erl b/deps/rabbit/src/rabbit_db_user_m2k_converter.erl
new file mode 100644
index 0000000000..95eb9c7c49
--- /dev/null
+++ b/deps/rabbit/src/rabbit_db_user_m2k_converter.erl
@@ -0,0 +1,200 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright © 2022-2023 VMware, Inc. or its affiliates.  All rights reserved.
+%%
+
+-module(rabbit_db_user_m2k_converter).
+
+-behaviour(mnesia_to_khepri_converter).
+
+-include_lib("kernel/include/logger.hrl").
+-include_lib("khepri/include/khepri.hrl").
+-include_lib("khepri_mnesia_migration/src/kmm_logging.hrl").
+-include_lib("rabbit_common/include/rabbit.hrl").
+
+-export([init_copy_to_khepri/2,
+         copy_to_khepri/3,
+         delete_from_khepri/3,
+         clear_data_in_khepri/1]).
+
+-record(?MODULE, {store_id :: khepri:store_id()}).
+
+-spec init_copy_to_khepri(Tables, StoreId) -> Ret when
+      Tables :: [mnesia_to_khepri:mnesia_table()],
+      StoreId :: khepri:store_id(),
+      Ret :: {ok, Priv},
+      Priv :: #?MODULE{}.
+%% @private
+
+init_copy_to_khepri(_Tables, StoreId) ->
+    State = #?MODULE{store_id = StoreId},
+    {ok, State}.
+
+-spec copy_to_khepri(Table, Record, Priv) -> Ret when
+      Table :: mnesia_to_khepri:mnesia_table(),
+      Record :: tuple(),
+      Priv :: #?MODULE{},
+      Ret :: {ok, NewPriv} | {error, Reason},
+      NewPriv :: #?MODULE{},
+      Reason :: any().
+%% @private
+
+copy_to_khepri(
+  rabbit_user = Table, Record,
+  #?MODULE{store_id = StoreId} = State) ->
+    Username = internal_user:get_username(Record),
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data copy: [~0p] key: ~0p",
+       [Table, Username],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    Path = rabbit_db_user:khepri_user_path(Username),
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data copy: [~0p] path: ~0p",
+       [Table, Path],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    case khepri:put(StoreId, Path, Record) of
+        ok    -> {ok, State};
+        Error -> Error
+    end;
+copy_to_khepri(
+  rabbit_user_permission = Table, Record,
+  #?MODULE{store_id = StoreId} = State) ->
+    #user_permission{
+       user_vhost = #user_vhost{
+                       username = Username,
+                       virtual_host = VHost}} = Record,
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data copy: [~0p] user: ~0p vhost: ~0p",
+       [Table, Username, VHost],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    Path = rabbit_db_user:khepri_user_permission_path(
+             #if_all{conditions =
+                         [Username,
+                          #if_node_exists{exists = true}]},
+             VHost),
+    Extra = #{keep_while =>
+                  #{rabbit_db_vhost:khepri_vhost_path(VHost) =>
+                        #if_node_exists{exists = true}}},
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data copy: [~0p] path: ~0p",
+       [Table, Path],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    case khepri:put(StoreId, Path, Record, Extra) of
+        ok    -> {ok, State};
+        Error -> Error
+    end;
+copy_to_khepri(
+  rabbit_topic_permission = Table, Record,
+  #?MODULE{store_id = StoreId} = State) ->
+    #topic_permission{
+       topic_permission_key =
+           #topic_permission_key{
+              user_vhost = #user_vhost{
+                              username = Username,
+                              virtual_host = VHost},
+              exchange = Exchange}} = Record,
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data copy: [~0p] user: ~0p vhost: ~0p",
+       [Table, Username, VHost],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    Path = rabbit_db_user:khepri_topic_permission_path(
+             #if_all{conditions =
+                         [Username,
+                          #if_node_exists{exists = true}]},
+             VHost,
+             Exchange),
+    Extra = #{keep_while =>
+                  #{rabbit_db_vhost:khepri_vhost_path(VHost) =>
+                        #if_node_exists{exists = true}}},
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data copy: [~0p] path: ~0p",
+       [Table, Path],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    case khepri:put(StoreId, Path, Record, Extra) of
+        ok    -> {ok, State};
+        Error -> Error
+    end;
+copy_to_khepri(Table, Record, State) ->
+    ?LOG_DEBUG("Mnesia->Khepri unexpected record table ~0p record ~0p state ~0p",
+               [Table, Record, State]),
+    {error, unexpected_record}.
+
+-spec delete_from_khepri(Table, Key, Priv) -> Ret when
+      Table :: mnesia_to_khepri:mnesia_table(),
+      Key :: any(),
+      Priv :: #?MODULE{},
+      Ret :: {ok, NewPriv} | {error, Reason},
+      NewPriv :: #?MODULE{},
+      Reason :: any().
+%% @private
+
+delete_from_khepri(
+  rabbit_user = Table, Key,
+  #?MODULE{store_id = StoreId} = State) ->
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data delete: [~0p] key: ~0p",
+       [Table, Key],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    Path = rabbit_db_user:khepri_user_path(Key),
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data delete: [~0p] path: ~0p",
+       [Table, Path],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    case khepri:delete(StoreId, Path) of
+        ok    -> {ok, State};
+        Error -> Error
+    end;
+delete_from_khepri(
+  rabbit_user_permission = Table, Key,
+  #?MODULE{store_id = StoreId} = State) ->
+    #user_vhost{
+       username = Username,
+       virtual_host = VHost} = Key,
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data delete: [~0p] key: ~0p",
+       [Table, Key],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    Path = rabbit_db_user:khepri_user_permission_path(Username, VHost),
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data delete: [~0p] path: ~0p",
+       [Table, Path],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    case khepri:delete(StoreId, Path) of
+        ok    -> {ok, State};
+        Error -> Error
+    end;
+delete_from_khepri(
+  rabbit_topic_permission = Table, Key,
+  #?MODULE{store_id = StoreId} = State) ->
+    #topic_permission_key{
+       user_vhost = #user_vhost{
+                       username = Username,
+                       virtual_host = VHost},
+       exchange = Exchange} = Key,
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data delete: [~0p] key: ~0p",
+       [Table, Key],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    Path = rabbit_db_user:khepri_topic_permission_path(Username, VHost, Exchange),
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data delete: [~0p] path: ~0p",
+       [Table, Path],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    case khepri:delete(StoreId, Path) of
+        ok    -> {ok, State};
+        Error -> Error
+    end.
+
+-spec clear_data_in_khepri(Table) -> ok when
+      Table :: atom().
+
+clear_data_in_khepri(rabbit_user) ->
+    Path = rabbit_db_user:khepri_users_path(),
+    case rabbit_khepri:delete(Path) of
+        ok    -> ok;
+        Error -> throw(Error)
+    end;
+clear_data_in_khepri(_) ->
+    ok.
diff --git a/deps/rabbit/src/rabbit_db_vhost.erl b/deps/rabbit/src/rabbit_db_vhost.erl
index f9dc9b2692..b45fb6e548 100644
--- a/deps/rabbit/src/rabbit_db_vhost.erl
+++ b/deps/rabbit/src/rabbit_db_vhost.erl
@@ -29,10 +29,6 @@
 -export([khepri_vhost_path/1,
          khepri_vhosts_path/0]).
 
--export([clear_data_in_khepri/1,
-         mnesia_write_to_khepri/2,
-         mnesia_delete_to_khepri/2]).
-
 %% For testing
 -export([clear/0]).
 
@@ -475,46 +471,7 @@ clear_in_mnesia() ->
     ok.
 
 clear_in_khepri() ->
-    khepri_delete(khepri_vhosts_path()).
-
-%% -------------------------------------------------------------------
-%% migration().
-%% -------------------------------------------------------------------
-
--spec mnesia_write_to_khepri(Table, [Queue]) -> ok when
-      Table :: atom(),
-      Queue :: amqqueue:amqqueue().
-
-mnesia_write_to_khepri(rabbit_vhost, VHosts) ->
-    rabbit_khepri:transaction(
-      fun() ->
-              [begin
-                   Name = vhost:get_name(VHost),
-                   Path = khepri_vhost_path(Name),
-                   case khepri_tx:put(Path, VHost) of
-                       ok    -> ok;
-                       Error -> throw(Error)
-                   end
-               end || VHost <- VHosts]
-      end).
-
--spec mnesia_delete_to_khepri(Table, ToDelete) -> ok when
-      Table :: atom(),
-      ToDelete :: vhost:vhost().
-
-mnesia_delete_to_khepri(rabbit_vhost, VHost) when ?is_vhost(VHost) ->
-    Name = vhost:get_name(VHost),
-    Path = khepri_vhost_path(Name),
-    khepri_delete(Path).
-
--spec clear_data_in_khepri(Table) -> ok when
-      Table :: atom().
-
-clear_data_in_khepri(rabbit_vhost) ->
     Path = khepri_vhosts_path(),
-    khepri_delete(Path).
-
-khepri_delete(Path) ->
     case rabbit_khepri:delete(Path) of
         ok    -> ok;
         Error -> throw(Error)
diff --git a/deps/rabbit/src/rabbit_db_vhost_m2k_converter.erl b/deps/rabbit/src/rabbit_db_vhost_m2k_converter.erl
new file mode 100644
index 0000000000..126a6e3070
--- /dev/null
+++ b/deps/rabbit/src/rabbit_db_vhost_m2k_converter.erl
@@ -0,0 +1,100 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright © 2022-2023 VMware, Inc. or its affiliates.  All rights reserved.
+%%
+
+-module(rabbit_db_vhost_m2k_converter).
+
+-behaviour(mnesia_to_khepri_converter).
+
+-include_lib("kernel/include/logger.hrl").
+-include_lib("khepri/include/khepri.hrl").
+-include_lib("khepri_mnesia_migration/src/kmm_logging.hrl").
+-include_lib("rabbit_common/include/rabbit.hrl").
+
+-export([init_copy_to_khepri/2,
+         copy_to_khepri/3,
+         delete_from_khepri/3,
+         clear_data_in_khepri/1]).
+
+-record(?MODULE, {store_id :: khepri:store_id()}).
+
+-spec init_copy_to_khepri(Tables, StoreId) -> Ret when
+      Tables :: [mnesia_to_khepri:mnesia_table()],
+      StoreId :: khepri:store_id(),
+      Ret :: {ok, Priv},
+      Priv :: #?MODULE{}.
+%% @private
+
+init_copy_to_khepri(_Tables, StoreId) ->
+    State = #?MODULE{store_id = StoreId},
+    {ok, State}.
+
+-spec copy_to_khepri(Table, Record, Priv) -> Ret when
+      Table :: mnesia_to_khepri:mnesia_table(),
+      Record :: tuple(),
+      Priv :: #?MODULE{},
+      Ret :: {ok, NewPriv} | {error, Reason},
+      NewPriv :: #?MODULE{},
+      Reason :: any().
+%% @private
+
+copy_to_khepri(
+  rabbit_vhost = Table, Record,
+  #?MODULE{store_id = StoreId} = State) ->
+    Name = vhost:get_name(Record),
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data copy: [~0p] key: ~0p",
+       [Table, Name],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    Path = rabbit_db_vhost:khepri_vhost_path(Name),
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data copy: [~0p] path: ~0p",
+       [Table, Path],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    case khepri:put(StoreId, Path, Record) of
+        ok    -> {ok, State};
+        Error -> Error
+    end;
+copy_to_khepri(Table, Record, State) ->
+    ?LOG_DEBUG("Mnesia->Khepri unexpected record table ~0p record ~0p state ~0p",
+               [Table, Record, State]),
+    {error, unexpected_record}.
+
+-spec delete_from_khepri(Table, Key, Priv) -> Ret when
+      Table :: mnesia_to_khepri:mnesia_table(),
+      Key :: any(),
+      Priv :: #?MODULE{},
+      Ret :: {ok, NewPriv} | {error, Reason},
+      NewPriv :: #?MODULE{},
+      Reason :: any().
+%% @private
+
+delete_from_khepri(
+  rabbit_vhost = Table, Key,
+  #?MODULE{store_id = StoreId} = State) ->
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data delete: [~0p] key: ~0p",
+       [Table, Key],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    Path = rabbit_db_vhost:khepri_vhost_path(Key),
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data delete: [~0p] path: ~0p",
+       [Table, Path],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    case khepri:delete(StoreId, Path) of
+        ok    -> {ok, State};
+        Error -> Error
+    end.
+
+-spec clear_data_in_khepri(Table) -> ok when
+      Table :: atom().
+
+clear_data_in_khepri(rabbit_vhost) ->
+    Path = rabbit_db_vhost:khepri_vhosts_path(),
+    case rabbit_khepri:delete(Path) of
+        ok    -> ok;
+        Error -> throw(Error)
+    end.
diff --git a/deps/rabbit/test/metadata_store_migration_SUITE.erl b/deps/rabbit/test/metadata_store_migration_SUITE.erl
new file mode 100644
index 0000000000..d2ff0b9c5f
--- /dev/null
+++ b/deps/rabbit/test/metadata_store_migration_SUITE.erl
@@ -0,0 +1,161 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright (c) 2018-2022 VMware, Inc. or its affiliates.  All rights reserved.
+%%
+
+-module(metadata_store_migration_SUITE).
+
+-include_lib("common_test/include/ct.hrl").
+-include_lib("eunit/include/eunit.hrl").
+-include_lib("amqp_client/include/amqp_client.hrl").
+
+-compile([nowarn_export_all, export_all]).
+-compile(export_all).
+
+suite() ->
+    [{timetrap, 5 * 60000}].
+
+all() ->
+    [
+     {group, khepri_migration}
+    ].
+
+groups() ->
+    [
+     {khepri_migration, [], [
+                             from_mnesia_to_khepri
+                            ]}
+    ].
+
+%% -------------------------------------------------------------------
+%% Testsuite setup/teardown.
+%% -------------------------------------------------------------------
+
+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(khepri_migration = Group, Config0) ->
+    rabbit_ct_helpers:set_config(Config0, [{metadata_store, mnesia},
+                                            {rmq_nodes_count, 1},
+                                            {rmq_nodename_suffix, Group},
+                                            {tcp_ports_base}]).
+
+end_per_group(_, Config) ->
+    Config.
+
+init_per_testcase(_Testcase, Config) ->
+    rabbit_ct_helpers:run_steps(Config, rabbit_ct_broker_helpers:setup_steps() ++ rabbit_ct_client_helpers:setup_steps()).
+
+end_per_testcase(Testcase, Config) ->
+    Config1 = rabbit_ct_helpers:run_steps(
+                Config,
+                rabbit_ct_client_helpers:teardown_steps()),
+    Config2 = rabbit_ct_helpers:testcase_finished(Config1, Testcase),
+    rabbit_ct_helpers:run_steps(Config2,
+                                rabbit_ct_broker_helpers:teardown_steps()).
+
+
+%% -------------------------------------------------------------------
+%% Testcases.
+%% -------------------------------------------------------------------
+from_mnesia_to_khepri(Config) ->
+    %% 1) Ensure there is at least one entry on each Mnesia table
+    %% 2) Enable the Khepri feature flag
+    %% 3) Check that all listings return the same values than before the migration
+
+    %% 1)
+    ok = rabbit_ct_broker_helpers:rpc(
+           Config, 0, rabbit_vhost, add, [<<"test">>, none]),
+    ok = rabbit_ct_broker_helpers:rpc(
+           Config, 0, rabbit_auth_backend_internal, set_topic_permissions,
+           [<<"guest">>, <<"/">>, <<"amq.topic">>, "^t", "^t", <<"acting-user">>]),
+    ok = rabbit_ct_broker_helpers:rpc(
+           Config, 0, rabbit_policy, set,
+           [<<"/">>, <<"policy">>, <<".*">>, [{<<"max-length">>, 100}], 0, <<"queues">>, none]),
+    ok = rabbit_ct_broker_helpers:rpc(
+           Config, 0, rabbit_runtime_parameters, set_global,
+           [<<"test-global-rt">>, <<"good">>, none]),
+    
+    Ch = rabbit_ct_client_helpers:open_channel(Config, 0),
+    amqp_channel:call(Ch, #'queue.declare'{queue = <<"test">>,
+                                           durable = true}),
+    amqp_channel:call(Ch, #'queue.declare'{queue = <<"test-transient">>,
+                                           durable = false}),
+    #'queue.bind_ok'{} = amqp_channel:call(Ch, #'queue.bind'{exchange = <<"amq.direct">>,
+                                                             queue = <<"test">>,
+                                                             routing_key = <<"test">>}),
+    #'queue.bind_ok'{} = amqp_channel:call(Ch, #'queue.bind'{exchange = <<"amq.topic">>,
+                                                             queue = <<"test">>,
+                                                             routing_key = <<"test">>}),
+    rabbit_ct_client_helpers:close_channel(Ch),
+
+    VHosts = lists:sort(rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_vhost, list, [])),
+    ?assertMatch(VHosts, lists:sort([<<"/">>, <<"test">>])),
+    Users = rabbit_ct_broker_helpers:rpc(
+              Config, 0, rabbit_auth_backend_internal, list_users, []),
+    ?assertMatch([_], Users),
+    UserPermissions = rabbit_ct_broker_helpers:rpc(
+                        Config, 0, rabbit_auth_backend_internal,
+                        list_user_permissions, [<<"guest">>]),
+    ?assertMatch([_], UserPermissions),
+    TopicPermissions = rabbit_ct_broker_helpers:rpc(
+                         Config, 0, rabbit_auth_backend_internal,
+                         list_user_topic_permissions, [<<"guest">>]),
+    ?assertMatch([_], TopicPermissions),
+    Policies = rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_policy, list, []),
+    ?assertMatch([_], Policies),
+    GlobalRuntimeParameters = lists:sort(rabbit_ct_broker_helpers:rpc(
+                                           Config, 0, rabbit_runtime_parameters, list_global, [])),
+    GRPNames = [proplists:get_value(name, RT) || RT <- GlobalRuntimeParameters],
+    ?assert(lists:member('test-global-rt', GRPNames)),
+    ?assert(lists:member('internal_cluster_id', GRPNames)),
+    Queues = lists:sort(rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_amqqueue, list, [])),
+    ?assertMatch([_, _], Queues),
+    Exchanges = lists:sort(rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_exchange, list, [])),
+    ?assertEqual(14, length(Exchanges)),
+    Bindings = lists:sort(rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_binding, list, [<<"/">>])),
+    ?assertEqual(4, length(Bindings)),
+    Server = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
+    Maintenance = rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_db_maintenance, get, [Server]),
+    ?assertNot(undefined == Maintenance),
+
+    %% 2)
+    Servers = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+    ok = rabbit_ct_broker_helpers:enable_feature_flag(Config, Servers, raft_based_metadata_store_phase1),
+    
+    %% 3)
+    VHostsK = lists:sort(rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_vhost, list, [])),
+    ?assertEqual(VHosts, VHostsK),
+    UsersK = rabbit_ct_broker_helpers:rpc(
+              Config, 0, rabbit_auth_backend_internal, list_users, []),
+    ?assertEqual(Users, UsersK),
+    UserPermissionsK = rabbit_ct_broker_helpers:rpc(
+                        Config, 0, rabbit_auth_backend_internal,
+                        list_user_permissions, [<<"guest">>]),
+    ?assertEqual(UserPermissions, UserPermissionsK),
+    TopicPermissionsK = rabbit_ct_broker_helpers:rpc(
+                          Config, 0, rabbit_auth_backend_internal,
+                          list_user_topic_permissions, [<<"guest">>]),
+    ?assertEqual(TopicPermissions, TopicPermissionsK),
+    PoliciesK = rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_policy, list, []),
+    ?assertEqual(Policies, PoliciesK),
+    GlobalRuntimeParametersK = lists:sort(rabbit_ct_broker_helpers:rpc(
+                                            Config, 0, rabbit_runtime_parameters, list_global, [])),
+    ?assertMatch(GlobalRuntimeParametersK, GlobalRuntimeParameters),
+    QueuesK = lists:sort(rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_amqqueue, list, [])),
+    ?assertEqual(Queues, QueuesK),
+    ExchangesK = lists:sort(rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_exchange, list, [])),
+    ?assertEqual(Exchanges, ExchangesK),
+    BindingsK = lists:sort(rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_binding, list, [<<"/">>])),
+    ?assertEqual(Bindings, BindingsK),
+    MaintenanceK = rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_db_maintenance, get, [Server]),
+    ?assertEqual(MaintenanceK, Maintenance),
+
+    ok.
+
diff --git a/deps/rabbitmq_consistent_hash_exchange/src/rabbit_db_ch_exchange.erl b/deps/rabbitmq_consistent_hash_exchange/src/rabbit_db_ch_exchange.erl
index 8bbe5b5f62..735d22e897 100644
--- a/deps/rabbitmq_consistent_hash_exchange/src/rabbit_db_ch_exchange.erl
+++ b/deps/rabbitmq_consistent_hash_exchange/src/rabbit_db_ch_exchange.erl
@@ -20,11 +20,12 @@
         ]).
 
 -export([mds_migration_enable/1,
-         mds_migration_post_enable/1,
-         mnesia_write_to_khepri/2,
-         mnesia_delete_to_khepri/2,
-         clear_data_in_khepri/1]).
+         mds_migration_post_enable/1]).
 
+-export([
+         khepri_consistent_hash_path/0,
+         khepri_consistent_hash_path/1
+        ]).
 
 -rabbit_feature_flag(
    {rabbit_consistent_hash_exchange_raft_based_metadata_store,
@@ -229,43 +230,13 @@ delete_binding_in_khepri(#binding{source = S, destination = D}, DeleteFun) ->
     end.
 
 mds_migration_enable(#{feature_name := FeatureName}) ->
-    TablesAndOwners = [{?HASH_RING_STATE_TABLE, ?MODULE}],
-    rabbit_core_ff:mds_migration_enable(FeatureName, TablesAndOwners).
+    TablesAndOwners = [{[?HASH_RING_STATE_TABLE], rabbit_db_ch_exchange_m2k_converter}],
+    rabbit_core_ff:mds_plugin_migration_enable(FeatureName, TablesAndOwners).
 
 mds_migration_post_enable(#{feature_name := FeatureName}) ->
-    TablesAndOwners = [{?HASH_RING_STATE_TABLE, ?MODULE}],
+    TablesAndOwners = [{[?HASH_RING_STATE_TABLE], rabbit_db_ch_exchange_m2k_converter}],
     rabbit_core_ff:mds_migration_post_enable(FeatureName, TablesAndOwners).
 
-clear_data_in_khepri(?HASH_RING_STATE_TABLE) ->
-    case rabbit_khepri:delete(khepri_consistent_hash_path()) of
-        ok ->
-            ok;
-        Error ->
-            throw(Error)
-    end.
-
-mnesia_write_to_khepri(?HASH_RING_STATE_TABLE, #chx_hash_ring{exchange = XName} = Record) ->
-    case rabbit_khepri:create(khepri_consistent_hash_path(XName), Record) of
-        ok -> ok;
-        {error, {khepri, mismatching_node, _}} -> ok;
-        Error -> throw(Error)
-    end.
-
-mnesia_delete_to_khepri(?HASH_RING_STATE_TABLE, #chx_hash_ring{exchange = XName}) ->
-    case rabbit_khepri:delete(khepri_consistent_hash_path(XName)) of
-        ok ->
-            ok;
-        Error ->
-            throw(Error)
-    end;
-mnesia_delete_to_khepri(?HASH_RING_STATE_TABLE, Key) ->
-    case rabbit_khepri:delete(khepri_consistent_hash_path(Key)) of
-        ok ->
-            ok;
-        Error ->
-            throw(Error)
-    end.
-
 khepri_consistent_hash_path(#exchange{name = Name}) ->
     khepri_consistent_hash_path(Name);
 khepri_consistent_hash_path(#resource{virtual_host = VHost, name = Name}) ->
diff --git a/deps/rabbitmq_consistent_hash_exchange/src/rabbit_db_ch_exchange_m2k_converter.erl b/deps/rabbitmq_consistent_hash_exchange/src/rabbit_db_ch_exchange_m2k_converter.erl
new file mode 100644
index 0000000000..c155f582b8
--- /dev/null
+++ b/deps/rabbitmq_consistent_hash_exchange/src/rabbit_db_ch_exchange_m2k_converter.erl
@@ -0,0 +1,101 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright © 2022-2023 VMware, Inc. or its affiliates.  All rights reserved.
+%%
+
+-module(rabbit_db_ch_exchange_m2k_converter).
+
+-behaviour(mnesia_to_khepri_converter).
+
+-include_lib("kernel/include/logger.hrl").
+-include_lib("khepri/include/khepri.hrl").
+-include_lib("khepri_mnesia_migration/src/kmm_logging.hrl").
+-include_lib("rabbit_common/include/rabbit.hrl").
+
+-include("rabbitmq_consistent_hash_exchange.hrl").
+
+-export([init_copy_to_khepri/2,
+         copy_to_khepri/3,
+         delete_from_khepri/3,
+         clear_data_in_khepri/1]).
+
+-record(?MODULE, {store_id :: khepri:store_id()}).
+
+-define(HASH_RING_STATE_TABLE, rabbit_exchange_type_consistent_hash_ring_state).
+
+-spec init_copy_to_khepri(Tables, StoreId) -> Ret when
+      Tables :: [mnesia_to_khepri:mnesia_table()],
+      StoreId :: khepri:store_id(),
+      Ret :: {ok, Priv},
+      Priv :: #?MODULE{}.
+%% @private
+
+init_copy_to_khepri(_Tables, StoreId) ->
+    State = #?MODULE{store_id = StoreId},
+    {ok, State}.
+
+-spec copy_to_khepri(Table, Record, Priv) -> Ret when
+      Table :: mnesia_to_khepri:mnesia_table(),
+      Record :: tuple(),
+      Priv :: #?MODULE{},
+      Ret :: {ok, NewPriv} | {error, Reason},
+      NewPriv :: #?MODULE{},
+      Reason :: any().
+%% @private
+
+copy_to_khepri(?HASH_RING_STATE_TABLE = Table,
+               #chx_hash_ring{exchange = XName} = Record,
+               #?MODULE{store_id = StoreId} = State) ->
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data copy: [~0p] key: ~0p",
+       [Table, XName],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    Path = rabbit_db_ch_exchange:khepri_consistent_hash_path(XName),
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data copy: [~0p] path: ~0p",
+       [Table, Path],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    case khepri:put(StoreId, Path, Record) of
+        ok    -> {ok, State};
+        Error -> Error
+    end;
+copy_to_khepri(Table, Record, State) ->
+    ?LOG_DEBUG("Mnesia->Khepri unexpected record table ~0p record ~0p state ~0p",
+               [Table, Record, State]),
+    {error, unexpected_record}.
+
+-spec delete_from_khepri(Table, Key, Priv) -> Ret when
+      Table :: mnesia_to_khepri:mnesia_table(),
+      Key :: any(),
+      Priv :: #?MODULE{},
+      Ret :: {ok, NewPriv} | {error, Reason},
+      NewPriv :: #?MODULE{},
+      Reason :: any().
+%% @private
+
+delete_from_khepri(?HASH_RING_STATE_TABLE = Table, Key,
+                   #?MODULE{store_id = StoreId} = State) ->
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data delete: [~0p] key: ~0p",
+       [Table, Key],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    Path = rabbit_db_vhost:khepri_consistent_hash_path(Key),
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data delete: [~0p] path: ~0p",
+       [Table, Path],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    case khepri:delete(StoreId, Path) of
+        ok    -> {ok, State};
+        Error -> Error
+    end.
+
+clear_data_in_khepri(?HASH_RING_STATE_TABLE) ->
+    Path = rabbit_db_ch_exchange:khepri_consistent_hash_path(),
+    case rabbit_khepri:delete(Path) of
+        ok ->
+            ok;
+        Error ->
+            throw(Error)
+    end.
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 8a5086dc45..61bfa9205a 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
@@ -20,10 +20,12 @@
         ]).
 
 -export([mds_migration_enable/1,
-         mds_migration_post_enable/1,
-         mnesia_write_to_khepri/2,
-         mnesia_delete_to_khepri/2,
-         clear_data_in_khepri/1]).
+         mds_migration_post_enable/1]).
+
+-export([
+         khepri_jms_topic_exchange_path/0,
+         khepri_jms_topic_exchange_path/1
+        ]).
 
 -rabbit_feature_flag(
    {rabbit_jms_topic_exchange_raft_based_metadata_store,
@@ -193,43 +195,13 @@ write_state_fun_in_mnesia(XName, BFuns) ->
 %% -------------------------------------------------------------------
 
 mds_migration_enable(#{feature_name := FeatureName}) ->
-    TablesAndOwners = [{?JMS_TOPIC_TABLE, ?MODULE}],
-    rabbit_core_ff:mds_migration_enable(FeatureName, TablesAndOwners).
+    TablesAndOwners = [{[?JMS_TOPIC_TABLE], rabbit_db_jms_exchange_m2k_converter}],
+    rabbit_core_ff:mds_plugin_migration_enable(FeatureName, TablesAndOwners).
 
 mds_migration_post_enable(#{feature_name := FeatureName}) ->
-    TablesAndOwners = [{?JMS_TOPIC_TABLE, ?MODULE}],
+    TablesAndOwners = [{[?JMS_TOPIC_TABLE], rabbit_db_jms_exchange_m2k_converter}],
     rabbit_core_ff:mds_migration_post_enable(FeatureName, TablesAndOwners).
 
-clear_data_in_khepri(?JMS_TOPIC_TABLE) ->
-    case rabbit_khepri:delete(khepri_jms_topic_exchange_path()) of
-        ok ->
-            ok;
-        Error ->
-            throw(Error)
-    end.
-
-mnesia_write_to_khepri(?JMS_TOPIC_TABLE, #?JMS_TOPIC_RECORD{x_name = XName, x_selector_funs = BFuns}) ->
-    case rabbit_khepri:create(khepri_jms_topic_exchange_path(XName), BFuns) of
-        ok -> ok;
-        {error, {khepri, mismatching_node, _}} -> ok;
-        Error -> throw(Error)
-    end.
-
-mnesia_delete_to_khepri(?JMS_TOPIC_TABLE, #?JMS_TOPIC_RECORD{x_name = XName}) ->
-    case rabbit_khepri:delete(khepri_jms_topic_exchange_path(XName)) of
-        ok ->
-            ok;
-        Error ->
-            throw(Error)
-    end;
-mnesia_delete_to_khepri(?JMS_TOPIC_TABLE, Key) ->
-    case rabbit_khepri:delete(khepri_jms_topic_exchange_path(Key)) of
-        ok ->
-            ok;
-        Error ->
-            throw(Error)
-    end.
-
 %% -------------------------------------------------------------------
 %% dictionary handling
 %% -------------------------------------------------------------------
diff --git a/deps/rabbitmq_jms_topic_exchange/src/rabbit_db_jms_exchange_m2k_converter.erl b/deps/rabbitmq_jms_topic_exchange/src/rabbit_db_jms_exchange_m2k_converter.erl
new file mode 100644
index 0000000000..d8f6a798ca
--- /dev/null
+++ b/deps/rabbitmq_jms_topic_exchange/src/rabbit_db_jms_exchange_m2k_converter.erl
@@ -0,0 +1,99 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright © 2022-2023 VMware, Inc. or its affiliates.  All rights reserved.
+%%
+
+-module(rabbit_db_jms_exchange_m2k_converter).
+
+-behaviour(mnesia_to_khepri_converter).
+
+-include_lib("kernel/include/logger.hrl").
+-include_lib("khepri/include/khepri.hrl").
+-include_lib("khepri_mnesia_migration/src/kmm_logging.hrl").
+-include_lib("rabbit_common/include/rabbit.hrl").
+
+-include("rabbit_jms_topic_exchange.hrl").
+
+-export([init_copy_to_khepri/2,
+         copy_to_khepri/3,
+         delete_from_khepri/3,
+         clear_data_in_khepri/1]).
+
+-record(?MODULE, {store_id :: khepri:store_id()}).
+
+-spec init_copy_to_khepri(Tables, StoreId) -> Ret when
+      Tables :: [mnesia_to_khepri:mnesia_table()],
+      StoreId :: khepri:store_id(),
+      Ret :: {ok, Priv},
+      Priv :: #?MODULE{}.
+%% @private
+
+init_copy_to_khepri(_Tables, StoreId) ->
+    State = #?MODULE{store_id = StoreId},
+    {ok, State}.
+
+-spec copy_to_khepri(Table, Record, Priv) -> Ret when
+      Table :: mnesia_to_khepri:mnesia_table(),
+      Record :: tuple(),
+      Priv :: #?MODULE{},
+      Ret :: {ok, NewPriv} | {error, Reason},
+      NewPriv :: #?MODULE{},
+      Reason :: any().
+%% @private
+
+copy_to_khepri(
+  ?JMS_TOPIC_TABLE = Table, #?JMS_TOPIC_RECORD{x_name = XName, x_selector_funs = BFuns},
+  #?MODULE{store_id = StoreId} = State) ->
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data copy: [~0p] key: ~0p",
+       [Table, XName],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    Path = rabbit_db_jms_exchange:khepri_jms_topic_exchange_path(XName),
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data copy: [~0p] path: ~0p",
+       [Table, Path],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    case khepri:put(StoreId, Path, BFuns) of
+        ok    -> {ok, State};
+        Error -> Error
+    end;
+copy_to_khepri(Table, Record, State) ->
+    ?LOG_DEBUG("Mnesia->Khepri unexpected record table ~0p record ~0p state ~0p",
+               [Table, Record, State]),
+    {error, unexpected_record}.
+
+-spec delete_from_khepri(Table, Key, Priv) -> Ret when
+      Table :: mnesia_to_khepri:mnesia_table(),
+      Key :: any(),
+      Priv :: #?MODULE{},
+      Ret :: {ok, NewPriv} | {error, Reason},
+      NewPriv :: #?MODULE{},
+      Reason :: any().
+%% @private
+
+delete_from_khepri(?JMS_TOPIC_TABLE = Table, Key,
+                   #?MODULE{store_id = StoreId} = State) ->
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data delete: [~0p] key: ~0p",
+       [Table, Key],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    Path = rabbit_db_jms_exchange:khepri_jms_topic_exchange_path(Key),
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data delete: [~0p] path: ~0p",
+       [Table, Path],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    case khepri:delete(StoreId, Path) of
+        ok    -> {ok, State};
+        Error -> Error
+    end.
+
+clear_data_in_khepri(?JMS_TOPIC_TABLE) ->
+    Path = rabbit_db_jms_exchange:khepri_jms_topic_exchange_path(),
+    case rabbit_khepri:delete(Path) of
+        ok ->
+            ok;
+        Error ->
+            throw(Error)
+    end.
diff --git a/deps/rabbitmq_recent_history_exchange/src/rabbit_db_rh_exchange.erl b/deps/rabbitmq_recent_history_exchange/src/rabbit_db_rh_exchange.erl
index 28e3813202..c63f73f7ca 100644
--- a/deps/rabbitmq_recent_history_exchange/src/rabbit_db_rh_exchange.erl
+++ b/deps/rabbitmq_recent_history_exchange/src/rabbit_db_rh_exchange.erl
@@ -22,11 +22,8 @@
 -export([mds_migration_enable/1,
          mds_migration_post_enable/1]).
 
--export([
-         mnesia_write_to_khepri/2,
-         mnesia_delete_to_khepri/2,
-         clear_data_in_khepri/1
-        ]).
+-export([khepri_recent_history_path/1,
+         khepri_recent_history_path/0]).
 
 -rabbit_feature_flag(
    {rabbit_recent_history_exchange_raft_based_metadata_store,
@@ -179,43 +176,13 @@ delete_in_khepri(XName) ->
 %% migration
 %% -------------------------------------------------------------------
 mds_migration_enable(#{feature_name := FeatureName}) ->
-    TablesAndOwners = [{?RH_TABLE, ?MODULE}],
-    rabbit_core_ff:mds_migration_enable(FeatureName, TablesAndOwners).
+    TablesAndOwners = [{[?RH_TABLE], rabbit_db_rh_exchange_m2k_converter}],
+    rabbit_core_ff:mds_plugin_migration_enable(FeatureName, TablesAndOwners).
 
 mds_migration_post_enable(#{feature_name := FeatureName}) ->
-    TablesAndOwners = [{?RH_TABLE, ?MODULE}],
+    TablesAndOwners = [{[?RH_TABLE], rabbit_db_rh_exchange_m2k_converter}],
     rabbit_core_ff:mds_migration_post_enable(FeatureName, TablesAndOwners).
 
-clear_data_in_khepri(?RH_TABLE) ->
-    case rabbit_khepri:delete(khepri_recent_history_path()) of
-        ok ->
-            ok;
-        Error ->
-            throw(Error)
-    end.
-
-mnesia_write_to_khepri(?RH_TABLE, #cached{key = Key, content = Content}) ->
-    case rabbit_khepri:create(khepri_recent_history_path(Key), Content) of
-        ok -> ok;
-        {error, {khepri, mismatching_node, _}} -> ok;
-        Error -> throw(Error)
-    end.
-
-mnesia_delete_to_khepri(?RH_TABLE, #cached{key = Key}) ->
-    case rabbit_khepri:delete(khepri_recent_history_path(Key)) of
-        ok ->
-            ok;
-        Error ->
-            throw(Error)
-    end;
-mnesia_delete_to_khepri(?RH_TABLE, Key) ->
-    case rabbit_khepri:delete(khepri_recent_history_path(Key)) of
-        ok ->
-            ok;
-        Error ->
-            throw(Error)
-    end.
-
 %% -------------------------------------------------------------------
 %% paths
 %% -------------------------------------------------------------------
diff --git a/deps/rabbitmq_recent_history_exchange/src/rabbit_db_rh_exchange_m2k_converter.erl b/deps/rabbitmq_recent_history_exchange/src/rabbit_db_rh_exchange_m2k_converter.erl
new file mode 100644
index 0000000000..ac62a97359
--- /dev/null
+++ b/deps/rabbitmq_recent_history_exchange/src/rabbit_db_rh_exchange_m2k_converter.erl
@@ -0,0 +1,98 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright © 2022-2023 VMware, Inc. or its affiliates.  All rights reserved.
+%%
+
+-module(rabbit_db_rh_exchange_m2k_converter).
+
+-behaviour(mnesia_to_khepri_converter).
+
+-include_lib("kernel/include/logger.hrl").
+-include_lib("khepri/include/khepri.hrl").
+-include_lib("khepri_mnesia_migration/src/kmm_logging.hrl").
+-include_lib("rabbit_common/include/rabbit.hrl").
+
+-include("rabbit_recent_history.hrl").
+
+-export([init_copy_to_khepri/2,
+         copy_to_khepri/3,
+         delete_from_khepri/3,
+         clear_data_in_khepri/1]).
+
+-record(?MODULE, {store_id :: khepri:store_id()}).
+
+-spec init_copy_to_khepri(Tables, StoreId) -> Ret when
+      Tables :: [mnesia_to_khepri:mnesia_table()],
+      StoreId :: khepri:store_id(),
+      Ret :: {ok, Priv},
+      Priv :: #?MODULE{}.
+%% @private
+
+init_copy_to_khepri(Tables, StoreId) ->
+    State = #?MODULE{store_id = StoreId},
+    {ok, State}.
+
+-spec copy_to_khepri(Table, Record, Priv) -> Ret when
+      Table :: mnesia_to_khepri:mnesia_table(),
+      Record :: tuple(),
+      Priv :: #?MODULE{},
+      Ret :: {ok, NewPriv} | {error, Reason},
+      NewPriv :: #?MODULE{},
+      Reason :: any().
+%% @private
+
+copy_to_khepri(?RH_TABLE = Table, #cached{key = Key, content = Content},
+               #?MODULE{store_id = StoreId} = State) ->
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data copy: [~0p] key: ~0p",
+       [Table, Key],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    Path = rabbit_db_rh_exchange:khepri_recent_history_path(Key),
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data copy: [~0p] path: ~0p",
+       [Table, Path],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    case khepri:put(StoreId, Path, Content) of
+        ok    -> {ok, State};
+        Error -> Error
+    end;
+copy_to_khepri(Table, Record, State) ->
+    ?LOG_DEBUG("Mnesia->Khepri unexpected record table ~0p record ~0p state ~0p",
+               [Table, Record, State]),
+    {error, unexpected_record}.
+
+-spec delete_from_khepri(Table, Key, Priv) -> Ret when
+      Table :: mnesia_to_khepri:mnesia_table(),
+      Key :: any(),
+      Priv :: #?MODULE{},
+      Ret :: {ok, NewPriv} | {error, Reason},
+      NewPriv :: #?MODULE{},
+      Reason :: any().
+%% @private
+
+delete_from_khepri(?RH_TABLE = Table, Key,
+                   #?MODULE{store_id = StoreId} = State) ->
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data delete: [~0p] key: ~0p",
+       [Table, Key],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    Path = rabbit_db_rh_exchange:khepri_recent_history_path(Key),
+    ?LOG_DEBUG(
+       "Mnesia->Khepri data delete: [~0p] path: ~0p",
+       [Table, Path],
+       #{domain => ?KMM_M2K_TABLE_COPY_LOG_DOMAIN}),
+    case khepri:delete(StoreId, Path) of
+        ok    -> {ok, State};
+        Error -> Error
+    end.
+
+clear_data_in_khepri(?RH_TABLE) ->
+    Path = rabbit_db_rh_exchange:khepri_recent_history_path(),
+    case rabbit_khepri:delete(Path) of
+        ok ->
+            ok;
+        Error ->
+            throw(Error)
+    end.