protected synchronized void tryListen(final List<PipeSnapshotEvent> events) {
    if (!isClosed.get()) {
      clearSnapshots();
      queueTailIndex2SnapshotsCache.setLeft(queue.getTailIndex());
      events.forEach(
          event -> event.increaseReferenceCount(AbstractPipeListeningQueue.class.getName()));
      queueTailIndex2SnapshotsCache.setRight(events);
      LOGGER.info(
          "Pipe listening queue snapshot cache is updated: {}", queueTailIndex2SnapshotsCache);
    }
  }
{@link AbstractPipeListeningQueue} is the encapsulation of the {@link
AbstractSerializableListeningQueue} to enable using reference count to control opening and
closing. This class also enables a {@link PipeTask} to find the snapshots close enough to send if
existed.
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/queue/listening/AbstractPipeListeningQueue.java