### Question:
IncrementRequestProcessor extends ListenableFutureRpcProcessor<IncrementRequest, IncrementResponse, RespT> { public static <T> IncrementRequestProcessor<T> of( IncrementRequest request, Function<IncrementResponse, T> responseFunc, StorageContainerChannel channel, ScheduledExecutorService executor, Policy backoffPolicy) { return new IncrementRequestProcessor<>(request, responseFunc, channel, executor, backoffPolicy); } private  IncrementRequestProcessor(IncrementRequest request,
                                      Function<IncrementResponse, RespT> respFunc,
                                      StorageContainerChannel channel,
                                      ScheduledExecutorService executor,
                                      Policy backoffPolicy); static IncrementRequestProcessor<T> of(
        IncrementRequest request,
        Function<IncrementResponse, T> responseFunc,
        StorageContainerChannel channel,
        ScheduledExecutorService executor,
        Policy backoffPolicy);  }### Answer:
@Test public void testProcess() throws Exception { @Cleanup("shutdown") ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(); StorageContainerChannel scChannel = mock(StorageContainerChannel.class); CompletableFuture<StorageServerChannel> serverChannelFuture = FutureUtils.createFuture(); when(scChannel.getStorageContainerChannelFuture()).thenReturn(serverChannelFuture); IncrementResponse response = newSuccessResponse(); AtomicReference<Object> receivedRequest = new AtomicReference<>(null); TableServiceImplBase tableService = new TableServiceImplBase() { @Override public void increment(IncrementRequest request, StreamObserver<IncrementResponse> responseObserver) { receivedRequest.set(request); complete(responseObserver); } private void complete(StreamObserver<IncrementResponse> responseStreamObserver) { responseStreamObserver.onNext(response); responseStreamObserver.onCompleted(); } }; serviceRegistry.addService(tableService.bindService()); StorageServerChannel ssChannel = new StorageServerChannel( InProcessChannelBuilder.forName(serverName).directExecutor().build(), Optional.empty()); serverChannelFuture.complete(ssChannel); IncrementRequest request = newRequest(); IncrementRequestProcessor<String> processor = IncrementRequestProcessor.of( request, resp -> "test", scChannel, scheduler, ClientConstants.DEFAULT_INFINIT_BACKOFF_POLICY); assertEquals("test", FutureUtils.result(processor.process())); assertSame(request, receivedRequest.get()); }