@TestOnly
  public Pair<List<SubscriptionEvent>, List<SubscriptionEvent>> filter(
      final List<SubscriptionEvent> events, final long maxBytes) {
    final List<SubscriptionEvent> eventsToPoll = new ArrayList<>();
    final List<SubscriptionEvent> eventsToNack = new ArrayList<>();

    long totalSize = 0;
    final Map<String, Long> topicNameToIncrements = new HashMap<>();

    // Sort the events based on the current subscription state (event counts per topic)
    sort(events);
    for (final SubscriptionEvent event : events) {
      final long currentSize;
      try {
        currentSize = event.getCurrentResponseSize();
      } catch (final IOException e) {
        // If there's an issue with accessing the event's size, nack the event
        eventsToNack.add(event);
        continue;
      }
      // If adding this event's size would not exceed maxBytes, add it to the poll list
      if (totalSize + currentSize < maxBytes) {
        eventsToPoll.add(event);
        totalSize += currentSize;
        topicNameToIncrements.merge(event.getCommitContext().getTopicName(), 1L, Long::sum);
      } else {
        // Otherwise, nack the event
        eventsToNack.add(event);
      }
    }

    // Update the subscription state with the increments calculated during filtering
    updateStates(topicNameToIncrements);
    return new Pair<>(eventsToPoll, eventsToNack);
  }
Filters the given list of SubscriptionEvents into two categories: events to poll and events to
nack. Events to poll are selected based on the available size (maxBytes) and the number of
events already received per topic, with preference given to topics with fewer events.

@param events the list of SubscriptionEvents to filter
@param maxBytes the maximum size (in bytes) of events that can be polled
@return a Pair containing two lists: the first list contains the events to poll, and the second
    list contains the events to nack
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionStates.java