anchor.diff:  -spec(publish(queue_type(), any(), any()) ->
                ok | {error, any()}).
   publish(?QUEUE_TYPE_SYNC_BY_VNODE_ID = Id, VNodeId, Node) ->
  -    Clock = leo_date:clock(),
  -    KeyBin     = term_to_binary({VNodeId, Node, Clock}),
  +    KeyBin = term_to_binary({VNodeId, Node}),
       MessageBin = term_to_binary(
  -                   #sync_unit_of_vnode_message{id        = Clock,
  -                                               vnode_id  = VNodeId,
  -                                               node      = Node,
  +                   #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) ->
  -    Clock = leo_date:clock(),
  -    KeyBin     = term_to_binary({AddrId, Key, Clock}),
  +    KeyBin = term_to_binary({AddrId, Key}),
       MessageBin = term_to_binary(
  -                   #async_deletion_message{id        = Clock,
  -                                           addr_id   = AddrId,
  -                                           key       = Key,
  +                   #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);
   
  
positive.code: -spec(publish(queue_type(), any(), any(), any()) ->
               ok | {error, any()}).
  publish(?QUEUE_TYPE_PER_OBJECT = Id, AddrId, Key, ErrorType) ->
      Clock = leo_date:clock(),
      KeyBin = term_to_binary({ErrorType, Key, Clock}),
      MessageBin = term_to_binary(
                     #inconsistent_data_message{id        = Clock,
                                                type      = ErrorType,
                                                addr_id   = AddrId,
                                                key       = Key,
                                                timestamp = leo_date:now()}),
      leo_mq_api:publish(queue_id(Id), KeyBin, MessageBin);
  
  publish(?QUEUE_TYPE_SYNC_OBJ_WITH_DC = Id, ClusterId, AddrId, Key) ->
      Clock = leo_date:clock(),
      KeyBin = term_to_binary({ClusterId, AddrId, Key, Clock}),
      MessageBin  = term_to_binary(
                      #inconsistent_data_with_dc{id         = Clock,
                                                 cluster_id = ClusterId,
                                                 addr_id    = AddrId,
                                                 key        = Key,
                                                 del        = 0,
                                                 timestamp  = leo_date:now()}),
      leo_mq_api:publish(queue_id(Id), KeyBin, MessageBin);
  
  publish(_,_,_,_) ->
  
negative.code: -spec(publish(queue_type(), any(), any(), any(), any()) ->
               ok | {error, any()}).
  publish(?QUEUE_TYPE_REBALANCE = Id, Node, VNodeId, AddrId, Key) ->
      KeyBin = term_to_binary({Node, AddrId, Key}),
      MessageBin = term_to_binary(
                     #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
  