public boolean onEvent(
      final @NonNull EnrichedEvent event, final Consumer<SubscriptionEvent> consumer) {
    final int regionId = event.getCommitterKey().getRegionId();

    final AtomicBoolean hasNew = new AtomicBoolean(false);
    regionIdToBatch.compute(
        regionId,
        (key, batch) -> {
          if (Objects.isNull(batch)) {
            batch =
                prefetchingQueue instanceof SubscriptionPrefetchingTabletQueue
                    ? new SubscriptionPipeTabletEventBatch(
                        key,
                        (SubscriptionPrefetchingTabletQueue) prefetchingQueue,
                        maxDelayInMs,
                        maxBatchSizeInBytes)
                    : new SubscriptionPipeTsFileEventBatch(
                        key,
                        (SubscriptionPrefetchingTsFileQueue) prefetchingQueue,
                        maxDelayInMs,
                        maxBatchSizeInBytes);
          }

          try {
            if (batch.onEvent(event, consumer)) {
              hasNew.set(true);
              return null; // remove this entry
            }
            // Seal this batch next time.
          } catch (final Exception e) {
            LOGGER.warn("Exception occurred when sealing events from batch {}", batch, e);
            // Seal this batch next time.
          }

          return batch;
        });

    return hasNew.get();
  }
@return {@code true} if there are subscription events consumed.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeEventBatches.java