private synchronized CachedSubscriptionPollResponse generateNextTabletResponse()
      throws InterruptedException, PipeRuntimeOutOfMemoryCriticalException {
    if (availableForNext) {
      // generate next subscription event with the same batch
      queue.prefetchEvent(new SubscriptionEvent(batch, queue, rootCommitContext));
      // frozen iterated enriched events
      transportIterationSnapshot();
      // return last response of this subscription event
      return new CachedSubscriptionPollResponse(
          SubscriptionPollResponseType.TABLETS.getType(),
          new TabletsPayload(Collections.emptyList(), -totalTablets),
          commitContext);
    }

    CachedSubscriptionPollResponse response = null;
    final Map<String, List<Tablet>> currentTablets = new HashMap<>();
    long currentBufferSize = 0;

    // TODO: TBD.
    // waitForResourceEnough4Parsing(SubscriptionAgent.receiver().remainingMs());

    while (batch.hasNext()) {
      final Pair<String, List<Tablet>> tablets = batch.next();
      if (Objects.isNull(tablets)) {
        continue;
      }

      currentTablets
          .computeIfAbsent(tablets.left, databaseName -> new ArrayList<>())
          .addAll(tablets.right);
      final long bufferSize =
          tablets.right.stream()
              .map(PipeMemoryWeightUtil::calculateTabletSizeInBytes)
              .reduce(Long::sum)
              .orElse(0L);
      totalTablets += tablets.right.size();
      totalBufferSize += bufferSize;
      currentBufferSize += bufferSize;

      if (bufferSize > READ_TABLET_BUFFER_SIZE) {
        // TODO: split tablets
        LOGGER.warn(
            "Detect large tablets with {} byte(s), current tablets size {} byte(s)",
            bufferSize,
            currentTablets);
        response =
            new CachedSubscriptionPollResponse(
                SubscriptionPollResponseType.TABLETS.getType(),
                new TabletsPayload(new HashMap<>(currentTablets), nextOffset.incrementAndGet()),
                commitContext);
        break;
      }

      if (currentBufferSize > READ_TABLET_BUFFER_SIZE) {
        // TODO: split tablets
        response =
            new CachedSubscriptionPollResponse(
                SubscriptionPollResponseType.TABLETS.getType(),
                new TabletsPayload(new HashMap<>(currentTablets), nextOffset.incrementAndGet()),
                commitContext);
        break;
      }

      // limit control for large message
      if (totalBufferSize > PREFETCH_TABLET_BUFFER_SIZE && batch.hasNext()) {
        // we generate seal signal at next round
        availableForNext = true;
        break;
      }
    }

    if (Objects.isNull(response)) {
      if (currentTablets.isEmpty()) {
        // frozen iterated enriched events
        transportIterationSnapshot();
        // return last response of this subscription event
        response =
            new CachedSubscriptionPollResponse(
                SubscriptionPollResponseType.TABLETS.getType(),
                new TabletsPayload(Collections.emptyList(), -totalTablets),
                commitContext);
        hasNoMore = true;
      } else {
        response =
            new CachedSubscriptionPollResponse(
                SubscriptionPollResponseType.TABLETS.getType(),
                new TabletsPayload(new HashMap<>(currentTablets), nextOffset.incrementAndGet()),
                commitContext);
      }
    }

    // set fixed memory block for response
    final List<Tablet> tablets = ((TabletsPayload) response.getPayload()).getTablets();
    if (Objects.nonNull(tablets) && !tablets.isEmpty()) {
      final PipeTabletMemoryBlock memoryBlock =
          PipeDataNodeResourceManager.memory().forceAllocateForTabletWithRetry(currentBufferSize);
      response.setMemoryBlock(memoryBlock);
    }

    return response;
  }
The {@code SubscriptionEventTabletResponse} class extends {@link
SubscriptionEventExtendableResponse} to handle subscription responses specifically for tablet
data. The actual payload of the response includes a {@link TabletsPayload}, which contains the
tablet information being processed.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/response/SubscriptionEventTabletResponse.java