@Override
    public Response request(Request request, long timeoutMills) throws NacosException {
        DefaultRequestFuture pushFuture = sendRequestInner(request, null);
        try {
            return pushFuture.get(timeoutMills);
        } catch (Exception e) {
            throw new NacosException(NacosException.SERVER_ERROR, e);
        } finally {
            RpcAckCallbackSynchronizer.clearFuture(getMetaInfo().getConnectionId(), pushFuture.getRequestId());
        }
    }
@Test
    void testBusy() {
        controlManagerCenterMockedStatic = Mockito.mockStatic(ControlManagerCenter.class);
        Mockito.when(ControlManagerCenter.getInstance()).thenReturn(controlManagerCenter);
        Mockito.when(ControlManagerCenter.getInstance().getTpsControlManager()).thenReturn(tpsControlManager);
        Mockito.when(tpsControlManager.check(Mockito.any())).thenReturn(new TpsCheckResponse(true, 200, ""));
        Mockito.doReturn(false).when(streamObserver).isReady();
        
        try {
            connection.request(new NotifySubscriberRequest(), 3000L);
            assertTrue(false);
        } catch (Exception e) {
            assertTrue(e instanceof ConnectionBusyException);
        }
        
        try {
            Thread.sleep(3001);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        try {
            connection.request(new NotifySubscriberRequest(), 3000L);
            assertTrue(false);
        } catch (Exception e) {
            assertTrue(e instanceof ConnectionBusyException);
        }
        
        assertTrue(connection.getMetaInfo().pushQueueBlockTimesLastOver(3000));
        
    }