@Override
    public void setNumBuffers(int numBuffers) {
        CompletableFuture<?> toNotify;
        synchronized (availableMemorySegments) {
            checkArgument(
                    numBuffers >= numberOfRequiredMemorySegments,
                    "Buffer pool needs at least %s buffers, but tried to set to %s",
                    numberOfRequiredMemorySegments,
                    numBuffers);

            currentPoolSize = Math.min(numBuffers, maxNumberOfMemorySegments);

            returnExcessMemorySegments();

            if (isDestroyed) {
                // FLINK-19964: when two local buffer pools are released concurrently, one of them
                // gets buffers assigned
                // make sure that checkAndUpdateAvailability is not called as it would proactively
                // acquire one buffer from NetworkBufferPool.
                return;
            }

            toNotify = checkAndUpdateAvailability();
        }

        mayNotifyAvailable(toNotify);
    }
@Test
    @SuppressWarnings("unchecked")
    void testConcurrentRequestRecycle() throws ExecutionException, InterruptedException {
        int numConcurrentTasks = 128;
        int numBuffersToRequestPerTask = 1024;

        localBufferPool.setNumBuffers(numConcurrentTasks);

        Future<Boolean>[] taskResults = new Future[numConcurrentTasks];
        for (int i = 0; i < numConcurrentTasks; i++) {
            taskResults[i] =
                    EXECUTOR_EXTENSION
                            .getExecutor()
                            .submit(
                                    new BufferRequesterTask(
                                            localBufferPool, numBuffersToRequestPerTask));
        }

        for (int i = 0; i < numConcurrentTasks; i++) {
            assertThat(taskResults[i].get()).isTrue();
        }
    }