code: @Override
      public void expire(K key, V val) {
          if (val == null) {
              return;
          }
          try {
              TupleInfo tupleInfo = (TupleInfo) val;
              FailSpoutMsg fail = new FailSpoutMsg(key, spout, tupleInfo, taskStats);
  
              disruptorEventQueue.publish(fail);
          } catch (Exception e) {
              LOG.error("expire error", e);
          }
      }
docs: pending.put(root_id, JStormUtils.mk_list(message_id, TupleInfo, ms));