public synchronized void tryExpandAllAndCheckConsistency() {
    allocatedBlocks.forEach(PipeMemoryBlock::expand);

    long blockSum =
        allocatedBlocks.stream().mapToLong(PipeMemoryBlock::getMemoryUsageInBytes).sum();
    if (blockSum != memoryBlock.getUsedMemoryInBytes()) {
      LOGGER.warn(
          "tryExpandAllAndCheckConsistency: memory usage is not consistent with allocated blocks,"
              + " usedMemorySizeInBytes is {} but sum of all blocks is {}",
          memoryBlock.getUsedMemoryInBytes(),
          blockSum);
    }

    long tabletBlockSum =
        allocatedBlocks.stream()
            .filter(PipeTabletMemoryBlock.class::isInstance)
            .mapToLong(PipeMemoryBlock::getMemoryUsageInBytes)
            .sum();
    if (tabletBlockSum != usedMemorySizeInBytesOfTablets) {
      LOGGER.warn(
          "tryExpandAllAndCheckConsistency: memory usage of tablets is not consistent with allocated blocks,"
              + " usedMemorySizeInBytesOfTablets is {} but sum of all tablet blocks is {}",
          usedMemorySizeInBytesOfTablets,
          tabletBlockSum);
    }

    long tsFileBlockSum =
        allocatedBlocks.stream()
            .filter(PipeTsFileMemoryBlock.class::isInstance)
            .mapToLong(PipeMemoryBlock::getMemoryUsageInBytes)
            .sum();
    if (tsFileBlockSum != usedMemorySizeInBytesOfTsFiles) {
      LOGGER.warn(
          "tryExpandAllAndCheckConsistency: memory usage of tsfiles is not consistent with allocated blocks,"
              + " usedMemorySizeInBytesOfTsFiles is {} but sum of all tsfile blocks is {}",
          usedMemorySizeInBytesOfTsFiles,
          tsFileBlockSum);
    }
  }
Allocate a {@link PipeMemoryBlock} for pipe only if memory used after allocation is less than
the specified threshold.

@param sizeInBytes size of memory needed to allocate
@param usedThreshold proportion of memory used, ranged from 0.0 to 1.0
@return {@code null} if the proportion of memory used after allocation exceeds {@code
    usedThreshold}. Will return a memory block otherwise.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java