FlumeEventQueue { synchronized void close() throws IOException { try { if (db != null) { db.close(); } } catch (Exception ex) { LOG.warn("Error closing db", ex); } try { backingStore.close(); inflightPuts.close(); inflightTakes.close(); } catch (IOException e) { LOG.warn("Error closing backing store", e); } } FlumeEventQueue(EventQueueBackingStore backingStore, File inflightTakesFile,
                  File inflightPutsFile, File queueSetDBDir); int getCapacity();  }
@Test(expected = IllegalArgumentException.class) public void testInvalidCapacityZero() throws Exception { backingStore.close(); File checkpoint = backingStoreSupplier.getCheckpoint(); Assert.assertTrue(checkpoint.delete()); backingStore = new EventQueueBackingStoreFileV2( checkpoint, 0, "test", new FileChannelCounter("test") ); queue = new FlumeEventQueue(backingStore, backingStoreSupplier.getInflightTakes(), backingStoreSupplier.getInflightPuts(), backingStoreSupplier.getQueueSetDir()); }
@Test(expected = IllegalArgumentException.class) public void testInvalidCapacityNegative() throws Exception { backingStore.close(); File checkpoint = backingStoreSupplier.getCheckpoint(); Assert.assertTrue(checkpoint.delete()); backingStore = new EventQueueBackingStoreFileV2( checkpoint, -1, "test", new FileChannelCounter("test") ); queue = new FlumeEventQueue(backingStore, backingStoreSupplier.getInflightTakes(), backingStoreSupplier.getInflightPuts(), backingStoreSupplier.getQueueSetDir()); }