@Test
    public void testCallbackCallerHandleException() {

        serverConfig = new ServerConfig()
            .setPort(22223)
            .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_HANDLE_EXCEPTION.name())
            .setDirectUrl("bolt://127.0.0.1:22223");
        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 sofaExceptionCount = 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) {
                }
                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();
                sofaExceptionCount.addAndGet(1);
            }
        };

        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) {
        }

        // Exception callbacks are triggered by IO threads after the thread pool is full, so the total number of responses received is equal to the total number of calls
        assertEquals(0, latch.getCount());
        // The number of exception callbacks triggered by IO threads must exist 
        assertTrue(sofaExceptionCount.get() > 0);
        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