public synchronized void stopListenAndAssign(
      final String dataRegionId, final PipeRealtimeDataRegionExtractor extractor) {
    final PipeDataRegionAssigner assigner = dataRegionId2Assigner.get(dataRegionId);
    if (assigner == null) {
      return;
    }

    assigner.stopAssignTo(extractor);

    if (extractor.isNeedListenToTsFile()) {
      listenToTsFileExtractorCount.decrementAndGet();
    }
    if (extractor.isNeedListenToInsertNode()) {
      listenToInsertNodeExtractorCount.decrementAndGet();
    }

    if (assigner.notMoreExtractorNeededToBeAssigned()) {
      // The removed assigner will is the same as the one referenced by the variable `assigner`
      dataRegionId2Assigner.remove(dataRegionId);
      // This will help to release the memory occupied by the assigner
      assigner.close();
    }
  }
PipeInsertionEventListener is a singleton in each data node.

<p>It is used to listen to events from storage engine and publish them to pipe engine.

<p>2 kinds of events are extracted: 1. level-0 tsfile sealed event 2. insertion operation event

<p>All events extracted by this listener will be first published to different
PipeEventDataRegionAssigners (identified by data region id), and then PipeEventDataRegionAssigner
will filter events and assign them to different PipeRealtimeEventDataRegionExtractors.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/listener/PipeInsertionDataNodeListener.java