@Test
    public void testCallbackCallerRuns() {

        serverConfig = new ServerConfig()
            .setPort(22224)
            .setDaemon(false);

        // RpcServer for C
        CProvider = new ProviderConfig<HelloService>()
            .setInterfaceId(HelloService.class.getName())
            .setRef(new HelloServiceImpl(0))
            .setServer(serverConfig);
        CProvider.export();

        // RpcClient For B invoke C
        Filter filter = new TestAsyncFilter();
        BConsumer = new ConsumerConfig<HelloService>()
            .setInterfaceId(HelloService.class.getName())
            .setInvokeType(RpcConstants.INVOKER_TYPE_CALLBACK)
            .setTimeout(3000)
            .setFilterRef(Arrays.asList(filter))
            .setRejectedExecutionPolicy(RejectedExecutionPolicy.CALLER_RUNS.name())
            .setDirectUrl("bolt://127.0.0.1:22224");
        HelloService helloService = BConsumer.refer();

        int maxsize = RpcConfigs.getIntValue(RpcOptions.ASYNC_POOL_MAX);
        int queuesize = RpcConfigs.getIntValue(RpcOptions.ASYNC_POOL_QUEUE);
        int invokeCount = (maxsize + queuesize) * 2;
        final CountDownLatch latch = new CountDownLatch(invokeCount);
        AtomicInteger runCount = new AtomicInteger(0);

        SofaResponseCallback callback = new SofaResponseCallback() {
            @Override
            public void onAppResponse(Object appResponse, String methodName, RequestBase request) {
                LOGGER.info("B get result: {}", appResponse);
                try {
                    Thread.sleep(100);
                } catch (Exception e) {
                }
                runCount.addAndGet(1);
                latch.countDown();
            }

            @Override
            public void onAppException(Throwable throwable, String methodName, RequestBase request) {
                LOGGER.info("B get app exception: {}", throwable);
                latch.countDown();
            }

            @Override
            public void onSofaException(SofaRpcException sofaException, String methodName,
                                        RequestBase request) {
                LOGGER.info("B get sofa exception: {}", sofaException);
                latch.countDown();
            }
        };

        for (int i = 0; i < invokeCount; i++) {
            RpcInvokeContext.getContext().setResponseCallback(callback);
            helloService.sayHello("" + i, 33);
        }

        try {
            latch.await(100L * invokeCount, TimeUnit.MILLISECONDS);
        } catch (InterruptedException ignore) {
        }

        assertEquals(0, latch.getCount());
        // Invoke callbacks are triggered by IO threads after the thread pool is full, so the total number of normal responses received is equal to the total number of calls
        assertEquals(invokeCount, runCount.get());
        RpcInvokeContext.removeContext();
    }
@author <a href="mailto:zhanggeng.zg@antfin.com">GengZhang</a>
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/invoke/AsyncCallbackTest.java