LargeMessageSafeStore implements KeyValueStore<byte[], byte[]> { @Override public void put(byte[] key, byte[] value) { validateMessageSize(value); if (!isLargeMessage(value)) { store.put(key, value); } else { LOG.info("Ignoring a large message with size " + value.length + " since it is greater than " + "the maximum allowed value of " + maxMessageSize); largeMessageSafeStoreMetrics.ignoredLargeMessages().inc(); } } LargeMessageSafeStore(KeyValueStore<byte[], byte[]> store, String storeName, boolean dropLargeMessages, int maxMessageSize); @Override byte[] get(byte[] key); @Override void put(byte[] key, byte[] value); @Override void putAll(List<Entry<byte[], byte[]>> entries); @Override void delete(byte[] key); @Override void deleteAll(List<byte[]> keys); @Override KeyValueIterator<byte[], byte[]> range(byte[] from, byte[] to); @Override KeyValueSnapshot<byte[], byte[]> snapshot(byte[] from, byte[] to); @Override KeyValueIterator<byte[], byte[]> all(); @Override void close(); @Override void flush(); @Override Optional<Path> checkpoint(CheckpointId id);  }
@Test public void testSmallMessagePut() { LargeMessageSafeStore largeMessageSafeKeyValueStore = new LargeMessageSafeStore(store, storeName, false, maxMessageSize); byte[] key = new byte[16]; byte[] smallMessage = new byte[32]; largeMessageSafeKeyValueStore.put(key, smallMessage); Mockito.verify(store).put(Matchers.eq(key), Matchers.eq(smallMessage)); }
@Test public void testLargeMessagePutWithDropLargeMessageDisabled() { LargeMessageSafeStore largeMessageSafeKeyValueStore = new LargeMessageSafeStore(store, storeName, false, maxMessageSize); byte[] key = new byte[16]; byte[] largeMessage = new byte[maxMessageSize + 1]; try { largeMessageSafeKeyValueStore.put(key, largeMessage); Assert.fail("The test case should have failed due to a large message being passed to the changelog, but it didn't."); } catch (RecordTooLargeException e) { Mockito.verifyZeroInteractions(store); } }
@Test public void testSmallMessagePutSuccessWithDropLargeMessageEnabled() { LargeMessageSafeStore largeMessageSafeKeyValueStore = new LargeMessageSafeStore(store, storeName, true, maxMessageSize); byte[] key = new byte[16]; byte[] smallMessage = new byte[32]; largeMessageSafeKeyValueStore.put(key, smallMessage); Mockito.verify(store).put(Matchers.eq(key), Matchers.eq(smallMessage)); }
@Test public void testLargeMessagePutSuccessWithDropLargeMessageEnabled() { LargeMessageSafeStore largeMessageSafeKeyValueStore = new LargeMessageSafeStore(store, storeName, true, maxMessageSize); byte[] key = new byte[16]; byte[] largeMessage = new byte[maxMessageSize + 1]; largeMessageSafeKeyValueStore.put(key, largeMessage); Mockito.verifyZeroInteractions(store); }