IOManagerAsync extends IOManager implements UncaughtExceptionHandler { RequestQueue<ReadRequest> getReadRequestQueue(FileIOChannel.ID channelID) { return this.readers[channelID.getThreadNum()].requestQueue; } IOManagerAsync();  IOManagerAsync(String tempDir);  IOManagerAsync(String[] tempDirs); @Override void shutdown(); @Override boolean isProperlyShutDown(); @Override void uncaughtException(Thread t, Throwable e); @Override BlockChannelWriter<MemorySegment> createBlockChannelWriter(FileIOChannel.ID channelID,
								LinkedBlockingQueue<MemorySegment> returnQueue); @Override BlockChannelWriterWithCallback<MemorySegment> createBlockChannelWriter(FileIOChannel.ID channelID, RequestDoneCallback<MemorySegment> callback); @Override BlockChannelReader<MemorySegment> createBlockChannelReader(FileIOChannel.ID channelID,
										LinkedBlockingQueue<MemorySegment> returnQueue); @Override BufferFileWriter createBufferFileWriter(FileIOChannel.ID channelID); @Override BufferFileReader createBufferFileReader(FileIOChannel.ID channelID, RequestDoneCallback<Buffer> callback); @Override BufferFileSegmentReader createBufferFileSegmentReader(FileIOChannel.ID channelID, RequestDoneCallback<FileSegment> callback); @Override BulkBlockChannelReader createBulkBlockChannelReader(FileIOChannel.ID channelID,
			List<MemorySegment> targetSegments, int numBlocks);  }
@Test public void testExceptionPropagationReader() { try { final AtomicBoolean handlerCalled = new AtomicBoolean(); final AtomicBoolean exceptionForwarded = new AtomicBoolean(); ReadRequest req = new ReadRequest() { @Override public void requestDone(IOException ioex) { if (ioex instanceof TestIOException) { exceptionForwarded.set(true); } synchronized (handlerCalled) { handlerCalled.set(true); handlerCalled.notifyAll(); } } @Override public void read() throws IOException { throw new TestIOException(); } }; RequestQueue<ReadRequest> rq = ioManager.getReadRequestQueue(ioManager.createChannel()); rq.add(req); synchronized (handlerCalled) { while (!handlerCalled.get()) { handlerCalled.wait(); } } assertTrue(exceptionForwarded.get()); } catch (Exception e) { e.printStackTrace(); fail(e.getMessage()); } }
@Test public void testExceptionInCallbackRead() { try { final AtomicBoolean handlerCalled = new AtomicBoolean(); ReadRequest regularRequest = new ReadRequest() { @Override public void requestDone(IOException ioex) { synchronized (handlerCalled) { handlerCalled.set(true); handlerCalled.notifyAll(); } } @Override public void read() {} }; ReadRequest exceptionThrower = new ReadRequest() { @Override public void requestDone(IOException ioex) { throw new RuntimeException(); } @Override public void read() {} }; RequestQueue<ReadRequest> rq = ioManager.getReadRequestQueue(ioManager.createChannel()); rq.add(exceptionThrower); rq.add(regularRequest); synchronized (handlerCalled) { while (!handlerCalled.get()) { handlerCalled.wait(); } } } catch (Exception e) { e.printStackTrace(); fail(e.getMessage()); } }