EndpointResultQueue { private synchronized void timeout(String operationId) { InflightOperation operation = inflightOperations.remove(operationId); if (operation == null) { log.finer("Timeout of operation '" + operationId + "', but operation " + "not found in map. Result was probably received just-in-time from server, while timeout " + "task could not be cancelled."); return; } EndpointResult endpointResult = EndPointResultFactory.createTransientError( endpoint, operationId, new RuntimeException("Timed out waiting for reply from server.")); operationProcessor.resultReceived(endpointResult, clusterId); } EndpointResultQueue(OperationProcessor operationProcessor,
                        Endpoint endpoint,
                        int clusterId,
                        ScheduledThreadPoolExecutor timer,
                        long totalTimeoutMs); synchronized void operationSent(String operationId, GatewayConnection connection); synchronized void failOperation(EndpointResult result, int clusterId); synchronized void resultReceived(EndpointResult result, int clusterId); synchronized int getPendingSize(); synchronized void failPending(Exception exception); synchronized boolean hasInflightOperations(GatewayConnection connection);  }
@Test public void testTimeout() throws InterruptedException { Endpoint endpoint = Endpoint.create("a"); OperationProcessor mockAggregator = mock(OperationProcessor.class); CountDownLatch latch = new CountDownLatch(1); doAnswer(invocationOnMock -> { latch.countDown(); return null; }).when(mockAggregator).resultReceived(any(), eq(0)); EndpointResultQueue q = new EndpointResultQueue( mockAggregator, endpoint, 0, new ScheduledThreadPoolExecutor(1), 100L); q.operationSent("1234", new DryRunGatewayConnection(endpoint, Clock.systemUTC())); assert(latch.await(120, TimeUnit.SECONDS)); }