FileChannel extends BasicChannelSemantics { @Override public synchronized void start() { LOG.info("Starting FileChannel with dataDir " + Arrays.toString(dataDirs)); try { log = new Log(checkpointInterval, maxFileSize, capacity, checkpointDir, dataDirs); log.replay(); } catch (IOException e) { Throwables.propagate(e); } open = true; boolean error = true; try { int depth = getDepth(); Preconditions.checkState(queueRemaining.tryAcquire(depth), "Unable to acquire " + depth + " permits"); LOG.info("Queue Size after replay: " + depth); if(!shutdownHookAdded) { shutdownHookAdded = true; final FileChannel fileChannel = this; LOG.info("Adding shutdownhook for " + fileChannel); shutdownHook = new Thread() { @Override public void run() { String desc = Arrays.toString(fileChannel.dataDirs); LOG.info("Closing FileChannel " + desc); try { fileChannel.close(); } catch (Exception e) { LOG.error("Error closing fileChannel " + desc, e); } } }; Runtime.getRuntime().addShutdownHook(shutdownHook); } error = false; } finally { if(error) { open = false; } } super.start(); }  @Override void configure(Context context); @Override synchronized void start(); @Override synchronized void stop();  }
@Test public void testThreaded() throws IOException, InterruptedException { int numThreads = 10; final CountDownLatch producerStopLatch = new CountDownLatch(numThreads); final CountDownLatch consumerStopLatch = new CountDownLatch(numThreads); final List<Exception> errors = Collections .synchronizedList(new ArrayList<Exception>()); final List<String> expected = Collections .synchronizedList(new ArrayList<String>()); final List<String> actual = Collections .synchronizedList(new ArrayList<String>()); for (int i = 0; i < numThreads; i++) { final int id = i; Thread t = new Thread() { @Override public void run() { try { if (id % 2 == 0) { expected.addAll(putEvents(channel, Integer.toString(id), 1, 5)); } else { expected.addAll(putEvents(channel, Integer.toString(id), 5, 5)); } LOG.info("Completed some puts " + expected.size()); } catch (Exception e) { LOG.error("Error doing puts", e); errors.add(e); } finally { producerStopLatch.countDown(); } } }; t.setDaemon(true); t.start(); } for (int i = 0; i < numThreads; i++) { final int id = i; Thread t = new Thread() { @Override public void run() { try { while(!producerStopLatch.await(1, TimeUnit.SECONDS) || expected.size() > actual.size()) { if (id % 2 == 0) { actual.addAll(takeEvents(channel, 1, Integer.MAX_VALUE)); } else { actual.addAll(takeEvents(channel, 5, Integer.MAX_VALUE)); } } if(actual.isEmpty()) { LOG.error("Found nothing!"); } else { LOG.info("Completed some takes " + actual.size()); } } catch (Exception e) { LOG.error("Error doing takes", e); errors.add(e); } finally { consumerStopLatch.countDown(); } } }; t.setDaemon(true); t.start(); } Assert.assertTrue("Timed out waiting for producers", producerStopLatch.await(30, TimeUnit.SECONDS)); Assert.assertTrue("Timed out waiting for consumer", consumerStopLatch.await(30, TimeUnit.SECONDS)); Assert.assertEquals(Collections.EMPTY_LIST, errors); Collections.sort(expected); Collections.sort(actual); Assert.assertEquals(expected, actual); }