private void bindOrUpdateProgressIndexForRealtimeEvent(final PipeRealtimeEvent event) {
    if (PipeTsFileEpochProgressIndexKeeper.getInstance()
        .isProgressIndexAfterOrEquals(
            dataRegionId,
            event.getTsFileEpoch().getFilePath(),
            getProgressIndex4RealtimeEvent(event))) {
      event.bindProgressIndex(maxProgressIndexForRealtimeEvent.get());
      if (LOGGER.isDebugEnabled()) {
        LOGGER.debug(
            "Data region {} bind {} to event {} because it was flushed prematurely.",
            dataRegionId,
            maxProgressIndexForRealtimeEvent,
            event.coreReportMessage());
      }
    } else {
      maxProgressIndexForRealtimeEvent.updateAndGet(
          index ->
              index.updateToMinimumEqualOrIsAfterProgressIndex(
                  getProgressIndex4RealtimeEvent(event)));
    }
  }
The {@link DisruptorQueue} is used to assign the event to the extractor.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/assigner/PipeDataRegionAssigner.java