@Override
  public synchronized void send(TsBlock tsBlock) {
    long startTime = System.nanoTime();
    try {
      Validate.notNull(tsBlock, "tsBlocks is null");
      if (closed) {
        // SinkChannel may have been closed by its downstream SourceHandle
        return;
      }
      checkState();
      if (!blocked.isDone()) {
        throw new IllegalStateException("Sink handle is blocked.");
      }
      if (noMoreTsBlocks) {
        return;
      }
      long sizeInBytes = tsBlock.getSizeInBytes();
      int startSequenceId;
      startSequenceId = nextSequenceId;
      blocked =
          localMemoryManager
              .getQueryPool()
              .reserve(
                  localFragmentInstanceId.getQueryId(),
                  fullFragmentInstanceId,
                  localPlanNodeId,
                  sizeInBytes,
                  maxBytesCanReserve)
              .left;
      bufferRetainedSizeInBytes += sizeInBytes;

      sequenceIdToTsBlock.put(nextSequenceId, new Pair<>(tsBlock, currentTsBlockSize));
      nextSequenceId += 1;
      currentTsBlockSize = sizeInBytes;

      submitSendNewDataBlockEventTask(startSequenceId, ImmutableList.of(sizeInBytes));
    } finally {
      DATA_EXCHANGE_COST_METRIC_SET.recordDataExchangeCost(
          SINK_HANDLE_SEND_TSBLOCK_REMOTE, System.nanoTime() - startTime);
    }
  }
max bytes this SinkChannel can reserve.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/sink/SinkChannel.java