### Question:
PutRequestProcessor extends ListenableFutureRpcProcessor<PutRequest, PutResponse, RespT> { public static <T> PutRequestProcessor<T> of( PutRequest request, Function<PutResponse, T> responseFunc, StorageContainerChannel channel, ScheduledExecutorService executor, Policy backoffPolicy) { return new PutRequestProcessor<>(request, responseFunc, channel, executor, backoffPolicy); } private  PutRequestProcessor(PutRequest request,
                                Function<PutResponse, RespT> respFunc,
                                StorageContainerChannel channel,
                                ScheduledExecutorService executor,
                                Policy backoffPolicy); static PutRequestProcessor<T> of(
        PutRequest request,
        Function<PutResponse, 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); PutResponse response = newSuccessResponse(); AtomicReference<Object> receivedRequest = new AtomicReference<>(null); TableServiceImplBase tableService = new TableServiceImplBase() { @Override public void put(PutRequest request, StreamObserver<PutResponse> responseObserver) { receivedRequest.set(request); complete(responseObserver); } private void complete(StreamObserver<PutResponse> responseStreamObserver) { responseStreamObserver.onNext(response); responseStreamObserver.onCompleted(); } }; serviceRegistry.addService(tableService.bindService()); StorageServerChannel ssChannel = new StorageServerChannel( InProcessChannelBuilder.forName(serverName).directExecutor().build(), Optional.empty()); serverChannelFuture.complete(ssChannel); PutRequest request = newRequest(); PutRequestProcessor<String> processor = PutRequestProcessor.of( request, resp -> "test", scChannel, scheduler, ClientConstants.DEFAULT_INFINIT_BACKOFF_POLICY); assertEquals("test", FutureUtils.result(processor.process())); assertSame(request, receivedRequest.get()); }