protected synchronized boolean onEvent(final Consumer<SubscriptionEvent> consumer)
      throws Exception {
    if (shouldEmit() && !enrichedEvents.isEmpty()) {
      if (Objects.isNull(events)) {
        events = generateSubscriptionEvents();
      }
      if (Objects.nonNull(events)) {
        events.forEach(consumer);
        return true;
      }
      return false;
    }
    return false;
  }
@return {@code true} if there are subscription events consumed.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeEventBatch.java