private @NonNull CachedSubscriptionPollResponse generateResponseWithPieceOrSealPayload(
      final long writingOffset)
      throws IOException, InterruptedException, PipeRuntimeOutOfMemoryCriticalException {
    final long tsFileLength = tsFile.length();
    if (writingOffset >= tsFileLength) {
      // generate subscription poll response with seal payload
      hasNoMore = true;
      return new CachedSubscriptionPollResponse(
          SubscriptionPollResponseType.FILE_SEAL.getType(),
          new FileSealPayload(tsFile.getName(), tsFile.length(), databaseName),
          commitContext);
    }

    final long bufferSize;
    if (writingOffset + READ_FILE_BUFFER_SIZE >= tsFileLength) {
      // last piece
      bufferSize = tsFileLength - writingOffset;
    } else {
      // not last piece
      bufferSize = READ_FILE_BUFFER_SIZE;
    }

    waitForResourceEnough4Slicing(SubscriptionAgent.receiver().remainingMs());
    try (final RandomAccessFile reader = new RandomAccessFile(tsFile, "r")) {
      reader.seek(writingOffset);

      final PipeTsFileMemoryBlock memoryBlock =
          PipeDataNodeResourceManager.memory().forceAllocateForTsFileWithRetry(bufferSize);
      final byte[] readBuffer = new byte[(int) bufferSize];

      final int readLength = reader.read(readBuffer);
      if (readLength != bufferSize) {
        memoryBlock.close();
        throw new SubscriptionException(
            String.format(
                "inconsistent read length (broken invariant), expected: %s, actual: %s",
                bufferSize, readLength));
      }

      // generate subscription poll response with piece payload
      final CachedSubscriptionPollResponse response =
          new CachedSubscriptionPollResponse(
              SubscriptionPollResponseType.FILE_PIECE.getType(),
              new FilePiecePayload(tsFile.getName(), writingOffset + readLength, readBuffer),
              commitContext);

      // set fixed memory block for response
      response.setMemoryBlock(memoryBlock);
      return response;
    }
  }
The {@code SubscriptionEventTsFileResponse} class extends {@link
SubscriptionEventExtendableResponse} to manage subscription responses related to time series
files. The actual payload can include {@link FileInitPayload}, {@link FilePiecePayload}, and
{@link FileSealPayload}, allowing for detailed control over file data streaming.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/response/SubscriptionEventTsFileResponse.java