LargeMessageSafeStore implements KeyValueStore<byte[], byte[]> { @Override public void putAll(List<Entry<byte[], byte[]>> entries) { entries.forEach(entry -> { validateMessageSize(entry.getValue()); }); List<Entry<byte[], byte[]>> largeMessageSafeEntries = removeLargeMessages(entries); store.putAll(largeMessageSafeEntries); } 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 testLargeMessagePutAllFailureWithDropLargeMessageDisabled() { LargeMessageSafeStore largeMessageSafeKeyValueStore = new LargeMessageSafeStore(store, storeName, false, maxMessageSize); byte[] key = new byte[16]; byte[] largeMessage = new byte[maxMessageSize + 1]; List<Entry<byte[], byte[]>> entries = new ArrayList<>(); entries.add(new Entry<>(key, largeMessage)); try { largeMessageSafeKeyValueStore.putAll(entries); 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 testLargeMessagePutWithSerdeAndDropLargeMessageDisabled() { LargeMessageSafeStore largeMessageSafeKeyValueStore = new LargeMessageSafeStore(store, storeName, false, maxMessageSize); Serde<Long> longSerde = new LongSerde(); long longObj = 1000L; byte[] key = longSerde.toBytes(longObj); Serde<String> stringSerde = new StringSerde(); String largeString = StringUtils.repeat("a", maxMessageSize + 1); byte[] largeMessage = stringSerde.toBytes(largeString); List<Entry<byte[], byte[]>> entries = new ArrayList<>(); entries.add(new Entry<>(key, largeMessage)); try { largeMessageSafeKeyValueStore.putAll(entries); } catch (RecordTooLargeException e) { Mockito.verifyZeroInteractions(store); } }