EventQueueBackingStoreFactory { static EventQueueBackingStore get( File checkpointFile, int capacity, String name, FileChannelCounter counter ) throws Exception { return get(checkpointFile, capacity, name, counter, true); } private  EventQueueBackingStoreFactory();   }
@Test public void testWithNoFlag() throws Exception { verify( EventQueueBackingStoreFactory.get(checkpoint, 10, "test", new FileChannelCounter("test")), Serialization.VERSION_3, pointersInTestCheckpoint ); }
@Test public void testWithFlag() throws Exception { verify( EventQueueBackingStoreFactory.get( checkpoint, 10, "test", new FileChannelCounter("test"), true ), Serialization.VERSION_3, pointersInTestCheckpoint ); }
@Test public void testNoUprade() throws Exception { verify( EventQueueBackingStoreFactory.get( checkpoint, 10, "test", new FileChannelCounter("test"), false ), Serialization.VERSION_2, pointersInTestCheckpoint ); }
@Test(expected = BadCheckpointException.class) public void testDecreaseCapacity() throws Exception { Assert.assertTrue(checkpoint.delete()); EventQueueBackingStore backingStore = EventQueueBackingStoreFactory.get(checkpoint, 10, "test", new FileChannelCounter("test")); backingStore.close(); EventQueueBackingStoreFactory.get(checkpoint, 9, "test", new FileChannelCounter("test")); Assert.fail(); }
@Test(expected = BadCheckpointException.class) public void testIncreaseCapacity() throws Exception { Assert.assertTrue(checkpoint.delete()); EventQueueBackingStore backingStore = EventQueueBackingStoreFactory.get(checkpoint, 10, "test", new FileChannelCounter("test")); backingStore.close(); EventQueueBackingStoreFactory.get(checkpoint, 11, "test", new FileChannelCounter("test")); Assert.fail(); }
@Test public void testNewCheckpoint() throws Exception { Assert.assertTrue(checkpoint.delete()); verify( EventQueueBackingStoreFactory.get( checkpoint, 10, "test", new FileChannelCounter("test"), false ), Serialization.VERSION_3, Collections.<Long>emptyList() ); }
@Test(expected = BadCheckpointException.class) public void testCheckpointBadVersion() throws Exception { RandomAccessFile writer = new RandomAccessFile(checkpoint, "rw"); try { EventQueueBackingStore backingStore = EventQueueBackingStoreFactory.get(checkpoint, 10, "test", new FileChannelCounter("test")); backingStore.close(); writer.seek(EventQueueBackingStoreFile.INDEX_VERSION * Serialization.SIZE_OF_LONG); writer.writeLong(94L); writer.getFD().sync(); backingStore = EventQueueBackingStoreFactory.get( checkpoint, 10, "test", new FileChannelCounter("test") ); } finally { writer.close(); } }
@Test(expected = BadCheckpointException.class) public void testIncompleteCheckpoint() throws Exception { RandomAccessFile writer = new RandomAccessFile(checkpoint, "rw"); try { EventQueueBackingStore backingStore = EventQueueBackingStoreFactory.get(checkpoint, 10, "test", new FileChannelCounter("test")); backingStore.close(); writer.seek(EventQueueBackingStoreFile.INDEX_CHECKPOINT_MARKER * Serialization.SIZE_OF_LONG); writer.writeLong(EventQueueBackingStoreFile.CHECKPOINT_INCOMPLETE); writer.getFD().sync(); backingStore = EventQueueBackingStoreFactory.get( checkpoint, 10, "test", new FileChannelCounter("test") ); } finally { writer.close(); } }
@Test(expected = BadCheckpointException.class) public void testCheckpointVersionNotEqualToMeta() throws Exception { RandomAccessFile writer = new RandomAccessFile(checkpoint, "rw"); try { EventQueueBackingStore backingStore = EventQueueBackingStoreFactory.get(checkpoint, 10, "test", new FileChannelCounter("test")); backingStore.close(); writer.seek(EventQueueBackingStoreFile.INDEX_VERSION * Serialization.SIZE_OF_LONG); writer.writeLong(2L); writer.getFD().sync(); backingStore = EventQueueBackingStoreFactory.get( checkpoint, 10, "test", new FileChannelCounter("test") ); } finally { writer.close(); } }
@Test(expected = BadCheckpointException.class) public void testCheckpointVersionNotEqualToMeta2() throws Exception { FileOutputStream os = null; try { EventQueueBackingStore backingStore = EventQueueBackingStoreFactory.get(checkpoint, 10, "test", new FileChannelCounter("test")); backingStore.close(); Assert.assertTrue(checkpoint.exists()); Assert.assertTrue(Serialization.getMetaDataFile(checkpoint).length() != 0); FileInputStream is = new FileInputStream(Serialization.getMetaDataFile(checkpoint)); ProtosFactory.Checkpoint meta = ProtosFactory.Checkpoint.parseDelimitedFrom(is); Assert.assertNotNull(meta); is.close(); os = new FileOutputStream(Serialization.getMetaDataFile(checkpoint)); meta.toBuilder().setVersion(2).build().writeDelimitedTo(os); os.flush(); backingStore = EventQueueBackingStoreFactory.get( checkpoint, 10, "test", new FileChannelCounter("test") ); } finally { os.close(); } }
@Test(expected = BadCheckpointException.class) public void testCheckpointOrderIdNotEqualToMeta() throws Exception { RandomAccessFile writer = new RandomAccessFile(checkpoint, "rw"); try { EventQueueBackingStore backingStore = EventQueueBackingStoreFactory.get(checkpoint, 10, "test", new FileChannelCounter("test")); backingStore.close(); writer.seek(EventQueueBackingStoreFile.INDEX_WRITE_ORDER_ID * Serialization.SIZE_OF_LONG); writer.writeLong(2L); writer.getFD().sync(); backingStore = EventQueueBackingStoreFactory.get( checkpoint, 10, "test", new FileChannelCounter("test") ); } finally { writer.close(); } }
@Test(expected = BadCheckpointException.class) public void testCheckpointOrderIdNotEqualToMeta2() throws Exception { FileOutputStream os = null; try { EventQueueBackingStore backingStore = EventQueueBackingStoreFactory.get(checkpoint, 10, "test", new FileChannelCounter("test")); backingStore.close(); Assert.assertTrue(checkpoint.exists()); Assert.assertTrue(Serialization.getMetaDataFile(checkpoint).length() != 0); FileInputStream is = new FileInputStream(Serialization.getMetaDataFile(checkpoint)); ProtosFactory.Checkpoint meta = ProtosFactory.Checkpoint.parseDelimitedFrom(is); Assert.assertNotNull(meta); is.close(); os = new FileOutputStream( Serialization.getMetaDataFile(checkpoint)); meta.toBuilder().setWriteOrderID(1).build().writeDelimitedTo(os); os.flush(); backingStore = EventQueueBackingStoreFactory.get( checkpoint, 10, "test", new FileChannelCounter("test") ); } finally { os.close(); } }
@Test(expected = BadCheckpointException.class) public void testTruncateMeta() throws Exception { EventQueueBackingStore backingStore = EventQueueBackingStoreFactory.get(checkpoint, 10, "test", new FileChannelCounter("test")); backingStore.close(); Assert.assertTrue(checkpoint.exists()); File metaFile = Serialization.getMetaDataFile(checkpoint); Assert.assertTrue(metaFile.length() != 0); RandomAccessFile writer = new RandomAccessFile(metaFile, "rw"); writer.setLength(0); writer.getFD().sync(); writer.close(); backingStore = EventQueueBackingStoreFactory.get( checkpoint, 10, "test", new FileChannelCounter("test") ); }
@Test(expected = InvalidProtocolBufferException.class) public void testCorruptMeta() throws Throwable { EventQueueBackingStore backingStore = EventQueueBackingStoreFactory.get(checkpoint, 10, "test", new FileChannelCounter("test")); backingStore.close(); Assert.assertTrue(checkpoint.exists()); File metaFile = Serialization.getMetaDataFile(checkpoint); Assert.assertTrue(metaFile.length() != 0); RandomAccessFile writer = new RandomAccessFile(metaFile, "rw"); writer.seek(10); writer.writeLong(new Random().nextLong()); writer.getFD().sync(); writer.close(); try { backingStore = EventQueueBackingStoreFactory.get( checkpoint, 10, "test", new FileChannelCounter("test") ); } catch (BadCheckpointException ex) { throw ex.getCause(); } }