public boolean onEvent(final Consumer<SubscriptionEvent> consumer) {
    final AtomicBoolean hasNew = new AtomicBoolean(false);
    for (final int regionId : ImmutableList.copyOf(regionIdToBatch.keySet())) {
      regionIdToBatch.compute(
          regionId,
          (key, batch) -> {
            if (Objects.isNull(batch)) {
              return null;
            }

            try {
              if (batch.onEvent(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;
          });

      if (hasNew.get()) {
        break;
      }
    }

    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