anchor.diff:  -spec(publish(queue_type(), any(), any(), any(), any()) ->
                ok | {error, any()}).
   publish(?QUEUE_TYPE_REBALANCE = Id, Node, VNodeId, AddrId, Key) ->
  -    Clock = leo_date:clock(),
  -    KeyBin = term_to_binary({Node, AddrId, Key, Clock}),
  +    KeyBin = term_to_binary({Node, AddrId, Key}),
       MessageBin = term_to_binary(
  -                   #rebalance_message{id        = Clock,
  -                                      vnode_id  = VNodeId,
  -                                      addr_id   = AddrId,
  -                                      key       = Key,
  -                                      node      = Node,
  +                   #rebalance_message{id = leo_date:clock(),
  +                                      vnode_id = VNodeId,
  +                                      addr_id = AddrId,
  +                                      key = Key,
  +                                      node = Node,
                                         timestamp = leo_date:now()}),
       Table = ?TBL_REBALANCE_COUNTER,
       case ets_lookup(Table, VNodeId) of
  
positive.code:     publish(?QUEUE_TYPE_SYNC_OBJ_WITH_DC, undefined, AddrId, Key);
  
  publish(?QUEUE_TYPE_COMP_META_WITH_DC = Id, ClusterId, AddrAndKeyList) ->
      Clock = leo_date:clock(),
      KeyBin     = term_to_binary({ClusterId, Clock}),
      MessageBin = term_to_binary(
                     #comparison_metadata_with_dc{id         = Clock,
                                                  cluster_id = ClusterId,
                                                  list_of_addrid_and_key = AddrAndKeyList,
                                                  timestamp = leo_date:now()}),
      leo_mq_api:publish(queue_id(Id), KeyBin, MessageBin);
  
  publish(?QUEUE_TYPE_DEL_DIR = Id, Node, Keys) ->
      Clock = leo_date:clock(),
      KeyBin = term_to_binary({Node, Keys, Clock}),
      MsgBin = term_to_binary(
                 #delete_dir{id   = Clock,
                             node = Node,
                             keys = Keys,
                             timestamp = leo_date:now()}),
  
negative.code: -spec(publish(queue_type(), any(), any()) ->
               ok | {error, any()}).
  publish(?QUEUE_TYPE_SYNC_BY_VNODE_ID = Id, VNodeId, Node) ->
      KeyBin = term_to_binary({VNodeId, Node}),
      MessageBin = term_to_binary(
                     #sync_unit_of_vnode_message{id = leo_date:clock(),
                                                 vnode_id = VNodeId,
                                                 node = Node,
                                                 timestamp = leo_date:now()}),
      leo_mq_api:publish(queue_id(Id), KeyBin, MessageBin);
  
  publish(?QUEUE_TYPE_ASYNC_DELETION = Id, AddrId, Key) ->
      KeyBin = term_to_binary({AddrId, Key}),
      MessageBin = term_to_binary(
                     #async_deletion_message{id = leo_date:clock(),
                                             addr_id = AddrId,
                                             key = Key,
                                             timestamp = leo_date:now()}),
      leo_mq_api:publish(queue_id(Id), KeyBin, MessageBin);
  
  