anchor.diff:                                                key = Key,
                                                 node = Node,
                                                 timestamp = leo_date:now()}),
  -            Table = ?TBL_REBALANCE_COUNTER,
  -            case ets_lookup(Table, VNodeId) of
  -                {ok, 0} ->
  -                    ets:insert(Table, {VNodeId, 0});
  -                _Other ->
  -                    void
  -            end,
  -            ok = increment_counter(Table, VNodeId),
               leo_mq_api:publish(Id, KeyBin, MessageBin)
       end;
   
  
positive.code:             end;
          _ ->
              Redundancies
      end.
  
  
  %% @doc Notify a rebalance-progress messages to manager.
  %%      - Retrieve # of published messages for rebalance,
  %%        after notify a message to manager.
  %%
  -spec(notify_rebalance_message_to_manager(integer()) ->
               ok | {error, any()}).
  notify_rebalance_message_to_manager(VNodeId) ->
      case ets_lookup(?TBL_REBALANCE_COUNTER, VNodeId) of
          {ok, NumOfMessages} ->
              Fun = fun(_Manager, true) ->
                            void;
                       (Manager0, false = Res) ->
                            Manager1 = case is_atom(Manager0) of
                                           true  -> Manager0;
                                           false -> list_to_atom(Manager0)
                                       end,
                            case catch rpc:call(Manager1, leo_manager_api, notify,
                                                [rebalance, VNodeId,
                                                 erlang:node(), NumOfMessages],
                                                ?DEF_REQ_TIMEOUT) of
                                ok ->
                                    true;
                                {_, Cause} ->
                                    ?error("notify_rebalance_message_to_manager/1",
                                           [{manager, Manager1},
                                            {vnode_id, VNodeId}, {cause, Cause}]),
                                    Res;
                                timeout ->
                                    Res
                            end
                    end,
              lists:foldl(Fun, false, ?env_manager_nodes(leo_storage)),
              ok;
          Error ->
              Error
      end.
  
  
  
negative.code:             false -> RootPath ++ ?SLASH
          end,
  
      DelBucketQueues = ?del_dir_queue_list(),
      start_1([{?QUEUE_ID_PER_OBJECT,        ?MSG_PATH_PER_OBJECT},
               {?QUEUE_ID_SYNC_BY_VNODE_ID,  ?MSG_PATH_SYNC_VNODE_ID},
  