refactor: remove persistence leakeage from emqx_cm level

apps/emqx/src/emqx_cm.erl

@@ -301,17 +301,9 @@ open_session(false, ClientInfo = #{clientid := ClientId}, ConnInfo) ->
     emqx_cm_locker:trans(ClientId, ResumeStart).
 
 create_session(ClientInfo, ConnInfo) ->
+    #{clientid := ClientId} = ClientInfo,
     Options = get_session_confs(ClientInfo, ConnInfo),
-    #{clientid := ClientID} = ClientInfo,
+    Session = emqx_session:init_and_open(ClientId, Options),
-    Session0 = emqx_session:init(Options),
-    IteratorIDs =
-        case emqx_persistent_session_ds:open_session(ClientID) of
-            {skipped, disabled} ->
-                [];
-            {_IsNew, _DSSessionID, Iterators0} ->
-                Iterators0
-        end,
-    Session = Session0#session{iterators = IteratorIDs},
     ok = emqx_metrics:inc('session.created'),
     ok = emqx_hooks:run('session.created', [ClientInfo, emqx_session:info(Session)]),
     Session.

apps/emqx/src/emqx_session.erl

@@ -60,7 +60,7 @@
     unpersist/1
 ]).
 
--export([init/1]).
+-export([init/1, init_and_open/2]).
 
 -export([
     info/1,
@@ -166,6 +166,18 @@
 %% Init a Session
 %%--------------------------------------------------------------------
 
+-spec init_and_open(emqx_types:clientid(), options()) -> session().
+init_and_open(ClientID, Options) ->
+    Session0 = emqx_session:init(Options),
+    IteratorIDs =
+        case emqx_persistent_session_ds:open_session(ClientID) of
+            {skipped, disabled} ->
+                [];
+            {_IsNew, _DSSessionID, Iterators0} ->
+                Iterators0
+        end,
+    Session0#session{iterators = IteratorIDs}.
+
 -spec init(options()) -> session().
 init(Opts) ->
     MaxInflight = maps:get(max_inflight, Opts),