From dc1deff3f3d9aa834c208112727f8d755690df9c Mon Sep 17 00:00:00 2001
From: k32 <10274441+k32@users.noreply.github.com>
Date: Fri, 25 Jun 2021 19:02:55 +0200
Subject: [PATCH] refactor(rlog): Fix initialization of emqx_cm_registry table

---
 apps/emqx/src/emqx_cm_registry.erl | 7 +++++--
 1 file changed, 5 insertions(+), 2 deletions(-)

diff --git a/apps/emqx/src/emqx_cm_registry.erl b/apps/emqx/src/emqx_cm_registry.erl
index d8095b445..30035eca5 100644
--- a/apps/emqx/src/emqx_cm_registry.erl
+++ b/apps/emqx/src/emqx_cm_registry.erl
@@ -48,7 +48,9 @@
 -define(TAB, emqx_channel_registry).
 -define(LOCK, {?MODULE, cleanup_down}).
 
--rlog_shard({?ROUTE_SHARD, ?TAB}).
+-define(CM_SHARD, emqx_cm_shard).
+
+-rlog_shard({?CM_SHARD, ?TAB}).
 
 -record(channel, {chid, pid}).
 
@@ -111,6 +113,7 @@ init([]) ->
                 {storage_properties, [{ets, [{read_concurrency, true},
                                              {write_concurrency, true}]}]}]),
     ok = ekka_mnesia:copy_table(?TAB, ram_copies),
+    ok = ekka_rlog:wait_for_shards([?CM_SHARD], infinity),
     ok = ekka:monitor(membership),
     {ok, #{}}.
 
@@ -125,7 +128,7 @@ handle_cast(Msg, State) ->
 handle_info({membership, {mnesia, down, Node}}, State) ->
     global:trans({?LOCK, self()},
                  fun() ->
-                     ekka_mnesia:transaction(?ROUTE_SHARD, fun cleanup_channels/1, [Node])
+                     ekka_mnesia:transaction(?CM_SHARD, fun cleanup_channels/1, [Node])
                  end),
     {noreply, State};