anchor.diff:              {ok, {K0, V0}} ->
                   case leo_mq_backend_db:get(BackendMessage, V0) of
                       {ok, V1} ->
  -                        error_logger:info_msg("~w,~w,~p~n", [?MODULE, consume, V0]),
  -
                           {_, MsgBin} = binary_to_term(V1),
   
                           catch erlang:apply(Mod, Fun, [Id, MsgBin]),
  
positive.code:                     ok ->
                          case leo_mq_backend_db:put(BackendMessage, MsgKeyBin, MessageBin) of
                              ok ->
                                  error_logger:info_msg("~w,~w,~p~n", [?MODULE, put_message, {MsgId, binary_to_list(MsgKeyBin)}]),
                                  ok;
                              Error ->
                                  leo_mq_backend_db:delete(BackendIndex, MsgIdBin),
  
negative.code: %% @doc Publish - Msg:"REPLICATE DATA".
  %%
  handle_cast({publish, KeyBin, MessageBin}, State) ->
      catch put_message(KeyBin, {leo_utils:clock(), MessageBin}, State),
  
      NewState = maybe_consume(State),
  