### Question:
ListingConnector implements IndexingConnector, ItemRetriever,
    IncrementalChangeHandler { @Subscribe public void handleAsyncOperation(AsyncApiOperation asyncOp) { ListenableFuture<List<GenericJson>> result = listeningExecutorService.submit( () -> { logger.log(Level.INFO, "Processing an asynchronous repository operation."); try { return execute( Collections.singleton(asyncOp.getOperation()), "asynchronous repository operation"); } catch (IOException ex) { logger.log( Level.WARNING, "Exception occured while processing an asynchronous repository operation: ", ex); throw ex; } }); asyncOp.setResult(result); } ListingConnector(Repository repository);  ListingConnector(Repository repository, CheckpointHandler checkpointHandler); @Override String getDefaultId(); @Override void init(IndexingConnectorContext context); @Override void traverse(); @Subscribe void handleAsyncOperation(AsyncApiOperation asyncOp); @Override void saveCheckpoint(boolean isShutdown); @Override void destroy(); @Override void process(Item item); @Override void handleIncrementalChanges(); static final String CONFIG_TRAVERSER; static final String CHECKPOINT_FULL; static final String CHECKPOINT_INCREMENTAL; }### Answer:
@Test public void testHandleAsyncOperation() throws Exception { setDefaultConfig(); ApiOperation mockPush = errorOperation; AsyncApiOperation pushAsyncOperation = new AsyncApiOperation(mockPush); ApiOperation mockDelete = spy(new ApiOperation() { @Override public List<GenericJson> execute(IndexingService service) throws IOException { return Lists.newArrayList(); } }); AsyncApiOperation deleteAsyncOperation = new AsyncApiOperation(mockDelete); ListingConnector connector = new ListingConnector(mockRepository); connector.init(mockConnectorContext); connector.handleAsyncOperation(pushAsyncOperation); connector.handleAsyncOperation(deleteAsyncOperation); assertNotNull(deleteAsyncOperation.getResult().get()); connector.destroy(); verify(mockPush).execute(eq(mockIndexingService), any()); verify(mockPush).execute(mockIndexingService); verify(mockDelete).execute(eq(mockIndexingService), any()); verify(mockDelete).execute(mockIndexingService); verifyNoMoreInteractions(mockIndexingService); thrown.expect(ExecutionException.class); pushAsyncOperation.getResult().get(); }