From 43e6112c0b48afe70763185b5f5fb1cbc281aee1 Mon Sep 17 00:00:00 2001 From: pchhokra Date: Wed, 24 Apr 2019 13:51:40 -0700 Subject: [PATCH 01/10] Throw a record too large exception for changelog oversized records --- .../kv/BaseKeyValueStorageEngineFactory.scala | 2 +- .../apache/samza/storage/kv/LoggedStore.scala | 43 +++++-- .../samza/storage/kv/TestLoggedStore.scala | 105 ++++++++++++++++++ 3 files changed, 138 insertions(+), 12 deletions(-) create mode 100644 samza-kv/src/test/scala/org/apache/samza/storage/kv/TestLoggedStore.scala diff --git a/samza-kv/src/main/scala/org/apache/samza/storage/kv/BaseKeyValueStorageEngineFactory.scala b/samza-kv/src/main/scala/org/apache/samza/storage/kv/BaseKeyValueStorageEngineFactory.scala index 6f1e0f6a33..70cb04a111 100644 --- a/samza-kv/src/main/scala/org/apache/samza/storage/kv/BaseKeyValueStorageEngineFactory.scala +++ b/samza-kv/src/main/scala/org/apache/samza/storage/kv/BaseKeyValueStorageEngineFactory.scala @@ -118,7 +118,7 @@ trait BaseKeyValueStorageEngineFactory[K, V] extends StorageEngineFactory[K, V] } else { val loggedStoreMetrics = new LoggedStoreMetrics(storeName, registry) storePropertiesBuilder = storePropertiesBuilder.setLoggedStore(true) - new LoggedStore(rawStore, changeLogSystemStreamPartition, collector, loggedStoreMetrics) + new LoggedStore(rawStore, storeName, storageConfig, changeLogSystemStreamPartition, collector, loggedStoreMetrics) } // wrap with serialization diff --git a/samza-kv/src/main/scala/org/apache/samza/storage/kv/LoggedStore.scala b/samza-kv/src/main/scala/org/apache/samza/storage/kv/LoggedStore.scala index e5f4ca43bf..9c027f02a5 100644 --- a/samza-kv/src/main/scala/org/apache/samza/storage/kv/LoggedStore.scala +++ b/samza-kv/src/main/scala/org/apache/samza/storage/kv/LoggedStore.scala @@ -19,34 +19,44 @@ package org.apache.samza.storage.kv +import org.apache.samza.SamzaException +import org.apache.samza.config.Config import org.apache.samza.util.Logging import org.apache.samza.system.{OutgoingMessageEnvelope, SystemStreamPartition} import org.apache.samza.task.MessageCollector +import scala.collection.JavaConverters.asScalaBufferConverter + /** * A key/value store decorator that adds a changelog for any changes made to the underlying store */ -class LoggedStore[K, V]( - val store: KeyValueStore[K, V], +class LoggedStore( + val store: KeyValueStore[Array[Byte], Array[Byte]], + val storeName: String, + val storeConfig: Config, val systemStreamPartition: SystemStreamPartition, val collector: MessageCollector, - val metrics: LoggedStoreMetrics = new LoggedStoreMetrics) extends KeyValueStore[K, V] with Logging { + val metrics: LoggedStoreMetrics = new LoggedStoreMetrics) extends KeyValueStore[Array[Byte], Array[Byte]] with Logging { val systemStream = systemStreamPartition.getSystemStream val partitionId = systemStreamPartition.getPartition.getPartitionId + private val DEFAULT_CHANGELOG_MAX_MSG_SIZE_BYTES = 1000000 + private val CHANGELOG_MAX_MSG_SIZE_BYTES = "changelog.max.message.size.bytes" + private val maxMessageSize = storeConfig.getInt(CHANGELOG_MAX_MSG_SIZE_BYTES, DEFAULT_CHANGELOG_MAX_MSG_SIZE_BYTES) // slightly less than 1 MB + /* pass through methods */ - def get(key: K) = { + def get(key: Array[Byte]) = { metrics.gets.inc store.get(key) } - override def getAll(keys: java.util.List[K]): java.util.Map[K, V] = { + override def getAll(keys: java.util.List[Array[Byte]]): java.util.Map[Array[Byte], Array[Byte]] = { metrics.gets.inc(keys.size) store.getAll(keys) } - def range(from: K, to: K) = { + def range(from: Array[Byte], to: Array[Byte]) = { metrics.ranges.inc store.range(from, to) } @@ -59,7 +69,8 @@ class LoggedStore[K, V]( /** * Perform the local update and log it out to the changelog */ - def put(key: K, value: V) { + def put(key: Array[Byte], value: Array[Byte]) { + validateMessageSize(key, value) metrics.puts.inc collector.send(new OutgoingMessageEnvelope(systemStream, partitionId, key, value)) store.put(key, value) @@ -68,7 +79,10 @@ class LoggedStore[K, V]( /** * Perform multiple local updates and log out all changes to the changelog */ - def putAll(entries: java.util.List[Entry[K, V]]) { + def putAll(entries: java.util.List[Entry[Array[Byte], Array[Byte]]]) { + entries.asScala.foreach(entry => { + validateMessageSize(entry.getKey, entry.getValue) + }) metrics.puts.inc(entries.size) val iter = entries.iterator while (iter.hasNext) { @@ -81,7 +95,7 @@ class LoggedStore[K, V]( /** * Perform the local delete and log it out to the changelog */ - def delete(key: K) { + def delete(key: Array[Byte]) { metrics.deletes.inc collector.send(new OutgoingMessageEnvelope(systemStream, partitionId, key, null)) store.delete(key) @@ -90,7 +104,7 @@ class LoggedStore[K, V]( /** * Perform the local deletes and log them out to the changelog */ - override def deleteAll(keys: java.util.List[K]) = { + override def deleteAll(keys: java.util.List[Array[Byte]]) = { metrics.deletes.inc(keys.size) val keysIterator = keys.iterator while (keysIterator.hasNext) { @@ -114,7 +128,14 @@ class LoggedStore[K, V]( store.close } - override def snapshot(from: K, to: K): KeyValueSnapshot[K, V] = { + override def snapshot(from: Array[Byte], to: Array[Byte]): KeyValueSnapshot[Array[Byte], Array[Byte]] = { store.snapshot(from, to) } + + private def validateMessageSize(key: Array[Byte], message: Array[Byte]): Unit = { + if (message.length > maxMessageSize) { + throw new SamzaException("RocksDB message size " + message.length + " for store " + storeName + + " was larger than the maximum allowed message size " + maxMessageSize + ".") + } + } } diff --git a/samza-kv/src/test/scala/org/apache/samza/storage/kv/TestLoggedStore.scala b/samza-kv/src/test/scala/org/apache/samza/storage/kv/TestLoggedStore.scala new file mode 100644 index 0000000000..1ffef380fa --- /dev/null +++ b/samza-kv/src/test/scala/org/apache/samza/storage/kv/TestLoggedStore.scala @@ -0,0 +1,105 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.samza.storage.kv + +import java.util._ + +import org.apache.samza.config.Config +import org.apache.samza.metrics.Counter +import org.apache.samza.system.SystemStreamPartition +import org.apache.samza.task.MessageCollector +import org.apache.samza.{Partition, SamzaException} +import org.junit.{Before, Test} +import org.mockito.Matchers +import org.mockito.Mockito._ + + +class TestLoggedStore { + + val store = mock(classOf[KeyValueStore[Array[Byte], Array[Byte]]]) + val storeName = "testStore" + val storeConfig = mock(classOf[Config]) + val systemStreamPartition = mock(classOf[SystemStreamPartition]) + val collector = mock(classOf[MessageCollector]) + val metrics = mock(classOf[LoggedStoreMetrics]) + val maxMessageSize = 1024 + val counter = mock(classOf[Counter]) + + @Before + def setup: Unit = { + when(systemStreamPartition.getPartition).thenReturn(new Partition(1)) + when(storeConfig.getInt(Matchers.eq("changelog.max.message.size.bytes"), Matchers.any())).thenReturn(maxMessageSize) + when(metrics.puts).thenReturn(counter) + when(counter.inc()).thenReturn(1) + doNothing().when(collector).send(Matchers.any()) + doNothing().when(store).put(Matchers.any(), Matchers.any()) + doNothing().when(store).putAll(Matchers.any()) + } + + @Test + def testLargeMessagePut = { + val safeLoggedStore = new LoggedStore(store, storeName, storeConfig, systemStreamPartition, collector, metrics) + + val key1 = new Array[Byte](16) + val smallMessage = new Array[Byte](32) + safeLoggedStore.put(key1, smallMessage) + verify(store, times(1)).put(Matchers.eq(key1), Matchers.eq(smallMessage)) + + val key2 = new Array[Byte](16) + val largeMessage = new Array[Byte](1025) + try { + safeLoggedStore.put(key2, largeMessage) + } catch { + case e: SamzaException => + verify(store, times(0)).put(Matchers.eq(key2), Matchers.eq(largeMessage)) + } + } + + @Test + def testSmallMessagePutAllSuccess = { + val safeLoggedStore = new LoggedStore(store, storeName, storeConfig, systemStreamPartition, collector, metrics) + val key = new Array[Byte](16) + val smallMessage = new Array[Byte](32) + + val entries = new ArrayList[Entry[Array[Byte], Array[Byte]]] + entries.add(new Entry(key, smallMessage)) + + safeLoggedStore.putAll(entries) + verify(store, times(1)).putAll(Matchers.eq(entries)) + verify(store, times(0)).put(Matchers.any(classOf[Array[Byte]]), Matchers.any(classOf[Array[Byte]])) + } + + @Test + def testLargeMessagePutAllFailure = { + val safeLoggedStore = new LoggedStore(store, storeName, storeConfig, systemStreamPartition, collector, metrics) + val key = new Array[Byte](16) + val largeMessage = new Array[Byte](1025) + + val entries = new ArrayList[Entry[Array[Byte], Array[Byte]]] + entries.add(new Entry(key, largeMessage)) + + try { + safeLoggedStore.putAll(entries) + } catch { + case e: SamzaException => + verify(store, times(0)).putAll(Matchers.eq(entries)) + verify(store, times(0)).put(Matchers.any(classOf[Array[Byte]]), Matchers.any(classOf[Array[Byte]])) + } + } +} From 4e3b65e72792d7fb8f00981c614dac1e5bb4d43e Mon Sep 17 00:00:00 2001 From: pchhokra Date: Wed, 22 May 2019 06:11:03 -0700 Subject: [PATCH 02/10] Change implementation to handle large messages in CachedStore based on user defined configs --- .../kv/BaseKeyValueStorageEngineFactory.scala | 77 ++++++-- .../storage/kv/LargeMessageSafeStore.scala | 107 ++++++++++ .../apache/samza/storage/kv/LoggedStore.scala | 65 +++--- .../kv/TestLargeMessageSafeStore.scala | 187 ++++++++++++++++++ .../samza/storage/kv/TestLoggedStore.scala | 105 ---------- .../TestLargeMessageSafeKeyValueStores.scala | 184 +++++++++++++++++ 6 files changed, 560 insertions(+), 165 deletions(-) create mode 100644 samza-kv/src/main/scala/org/apache/samza/storage/kv/LargeMessageSafeStore.scala create mode 100644 samza-kv/src/test/scala/org/apache/samza/storage/kv/TestLargeMessageSafeStore.scala delete mode 100644 samza-kv/src/test/scala/org/apache/samza/storage/kv/TestLoggedStore.scala create mode 100644 samza-test/src/test/scala/org/apache/samza/storage/kv/TestLargeMessageSafeKeyValueStores.scala diff --git a/samza-kv/src/main/scala/org/apache/samza/storage/kv/BaseKeyValueStorageEngineFactory.scala b/samza-kv/src/main/scala/org/apache/samza/storage/kv/BaseKeyValueStorageEngineFactory.scala index 70cb04a111..37bc4e7490 100644 --- a/samza-kv/src/main/scala/org/apache/samza/storage/kv/BaseKeyValueStorageEngineFactory.scala +++ b/samza-kv/src/main/scala/org/apache/samza/storage/kv/BaseKeyValueStorageEngineFactory.scala @@ -33,15 +33,22 @@ import org.apache.samza.task.MessageCollector import org.apache.samza.util.HighResolutionClock /** - * A key value storage engine factory implementation - * - * This trait encapsulates all the steps needed to create a key value storage engine. It is meant to be extended - * by the specific key value store factory implementations which will in turn override the getKVStore method. - */ + * A key value storage engine factory implementation + * + * This trait encapsulates all the steps needed to create a key value storage engine. It is meant to be extended + * by the specific key value store factory implementations which will in turn override the getKVStore method. + */ trait BaseKeyValueStorageEngineFactory[K, V] extends StorageEngineFactory[K, V] { private val INMEMORY_KV_STORAGE_ENGINE_FACTORY = "org.apache.samza.storage.kv.inmemory.InMemoryKeyValueStorageEngineFactory" + private val CHANGELOG_MAX_MSG_SIZE_BYTES = "changelog.max.message.size.bytes" + private val DEFAULT_CHANGELOG_MAX_MSG_SIZE_BYTES = 1000000 // slightly less than 1 MB + private val EXPECT_LARGE_MESSAGES = "expect.large.message" + private val DEFAULT_EXPECT_LARGE_MESSAGES = false + private val DROP_LARGE_MESSAGES = "drop.large.messages" + private val DEFAULT_DROP_LARGE_MESSAGES = false + /** * Return a KeyValueStore instance for the given store name, * which will be used as the underlying raw store @@ -85,6 +92,9 @@ trait BaseKeyValueStorageEngineFactory[K, V] extends StorageEngineFactory[K, V] val storeFactory = storageConfig.get("factory") var storePropertiesBuilder = new StoreProperties.StorePropertiesBuilder() val accessLog = storageConfig.getBoolean("accesslog.enabled", false) + val maxMessageSize = storageConfig.getInt(CHANGELOG_MAX_MSG_SIZE_BYTES, DEFAULT_CHANGELOG_MAX_MSG_SIZE_BYTES) + val largeMessagesExpected = storageConfig.getBoolean(EXPECT_LARGE_MESSAGES, DEFAULT_EXPECT_LARGE_MESSAGES) + val dropLargeMessage = storageConfig.getBoolean(DROP_LARGE_MESSAGES, DEFAULT_DROP_LARGE_MESSAGES) if (storeFactory == null) { throw new SamzaException("Store factory not defined. Cannot proceed with KV store creation!") @@ -118,25 +128,59 @@ trait BaseKeyValueStorageEngineFactory[K, V] extends StorageEngineFactory[K, V] } else { val loggedStoreMetrics = new LoggedStoreMetrics(storeName, registry) storePropertiesBuilder = storePropertiesBuilder.setLoggedStore(true) - new LoggedStore(rawStore, storeName, storageConfig, changeLogSystemStreamPartition, collector, loggedStoreMetrics) + new LoggedStore(rawStore, changeLogSystemStreamPartition, collector, loggedStoreMetrics) } - // wrap with serialization - val serializedMetrics = new SerializedKeyValueStoreMetrics(storeName, registry) - val serialized = new SerializedKeyValueStore[K, V](maybeLoggedStore, keySerde, msgSerde, serializedMetrics) + var toBeAccessLoggedStore: KeyValueStore[K, V] = null + + if (largeMessagesExpected) { + // Execute "if" when we expect to get large messages and user defined config -> EXPECT_LARGE_MESSAGES = true + // This code path will throw a SamzaException for large message size. + + // maybe wrap with caching + val maybeCachedStore = if (enableCache) { + val cachedStoreMetrics = new CachedStoreMetrics(storeName, registry) + new CachedStore(maybeLoggedStore, cacheSize, batchSize, cachedStoreMetrics) + } else { + maybeLoggedStore + } + + // wrap with large message checking + val largeMessageSafeKeyValueStore = new LargeMessageSafeStore(maybeCachedStore, storeName, false, maxMessageSize) + + // wrap with serialization + val serializedMetrics = new SerializedKeyValueStoreMetrics(storeName, registry) + val serialized = new SerializedKeyValueStore[K, V](largeMessageSafeKeyValueStore, keySerde, msgSerde, serializedMetrics) + toBeAccessLoggedStore = serialized - // maybe wrap with caching - val maybeCachedStore = if (enableCache) { - val cachedStoreMetrics = new CachedStoreMetrics(storeName, registry) - new CachedStore(serialized, cacheSize, batchSize, cachedStoreMetrics) } else { - serialized + + var toBeSerializedStore = maybeLoggedStore + if (dropLargeMessage) { + // Call LargeMessageSafeKeyValueStore when EXPECT_LARGE_MESSAGES = false but DROP_LARGE_MESSAGES = true + // This if statement code path will ignore all large messages and continue to add all other messages to the desired stores. + + // wrap with large message checking + toBeSerializedStore = new LargeMessageSafeStore(maybeLoggedStore, storeName, dropLargeMessage, maxMessageSize) + } + // wrap with serialization + val serializedMetrics = new SerializedKeyValueStoreMetrics(storeName, registry) + val serialized = new SerializedKeyValueStore[K, V](toBeSerializedStore, keySerde, msgSerde, serializedMetrics) + + // maybe wrap with caching + val maybeCachedStore = if (enableCache) { + val cachedStoreMetrics = new CachedStoreMetrics(storeName, registry) + new CachedStore(serialized, cacheSize, batchSize, cachedStoreMetrics) + } else { + serialized + } + toBeAccessLoggedStore = maybeCachedStore } val maybeAccessLoggedStore = if (accessLog) { - new AccessLoggedStore(maybeCachedStore, collector, changeLogSystemStreamPartition, storageConfig, storeName, keySerde) + new AccessLoggedStore(toBeAccessLoggedStore, collector, changeLogSystemStreamPartition, storageConfig, storeName, keySerde) } else { - maybeCachedStore + toBeAccessLoggedStore } // wrap with null value checking @@ -158,5 +202,4 @@ trait BaseKeyValueStorageEngineFactory[K, V] extends StorageEngineFactory[K, V] new KeyValueStorageEngine(storeName, storeDir, storePropertiesBuilder.build(), nullSafeStore, rawStore, keyValueStorageEngineMetrics, batchSize, () => clock.nanoTime()) } - } diff --git a/samza-kv/src/main/scala/org/apache/samza/storage/kv/LargeMessageSafeStore.scala b/samza-kv/src/main/scala/org/apache/samza/storage/kv/LargeMessageSafeStore.scala new file mode 100644 index 0000000000..8c7ab1676e --- /dev/null +++ b/samza-kv/src/main/scala/org/apache/samza/storage/kv/LargeMessageSafeStore.scala @@ -0,0 +1,107 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.samza.storage.kv +import java.util + +import org.apache.samza.SamzaException + +import scala.collection.JavaConverters.iterableAsScalaIterableConverter + +/** + * A wrapper over KeyValueStore that checks to see if any large messages have been passed to the store. + * + * @param store the store to persist to if messages are not large + * @param storeName store name + * @param dropLargeMessage a flag that determines if we want to ignore and drop large messages + * @param maxMessageSize maximum allowed size of a message, after which every message will be considered large + */ +class LargeMessageSafeStore( + store: KeyValueStore[Array[Byte], Array[Byte]], + storeName: String, + dropLargeMessage: Boolean, + maxMessageSize: Int) extends KeyValueStore[Array[Byte], Array[Byte]] { + + override def get(key: Array[Byte]): Array[Byte] = { + store.get(key) + } + + override def put(key: Array[Byte], value: Array[Byte]): Unit = { + validateMessageSize(value) + if (!isLargeMessage(value)) { + store.put(key, value) + } + } + + override def putAll(entries: util.List[Entry[Array[Byte], Array[Byte]]]): Unit = { + entries.asScala.foreach(entry => { + validateMessageSize(entry.getValue) + }) + val largeMessageSafeEntries = removeLargeMessage(entries) + store.putAll(largeMessageSafeEntries) + } + + override def delete(key: Array[Byte]): Unit = { + store.delete(key) + } + + override def deleteAll(keys: java.util.List[Array[Byte]]): Unit = { + store.deleteAll(keys) + } + + override def range(from: Array[Byte], to: Array[Byte]): KeyValueIterator[Array[Byte], Array[Byte]] = { + store.range(from, to) + } + + override def snapshot(from: Array[Byte], to: Array[Byte]): KeyValueSnapshot[Array[Byte], Array[Byte]] = { + store.snapshot(from, to) + } + + override def all(): KeyValueIterator[Array[Byte], Array[Byte]] = { + store.all() + } + + override def close(): Unit = { + store.close() + } + + override def flush(): Unit = { + store.flush() + } + + private def validateMessageSize(message: Array[Byte]): Unit = { + if (!dropLargeMessage && isLargeMessage(message)) { + throw new SamzaException("Logged store message size " + message.length + " for store " + storeName + + " was larger than the maximum allowed message size " + maxMessageSize + ".") + } + } + + private def isLargeMessage(message:Array[Byte]): Boolean = { + message != null && message.length > maxMessageSize + } + + private def removeLargeMessage(entries: util.List[Entry[Array[Byte], Array[Byte]]]): util.List[Entry[Array[Byte], Array[Byte]]] = { + val largeMessageSafeEntries = new util.ArrayList[Entry[Array[Byte], Array[Byte]]] + entries.asScala.foreach(entry => { + if (!isLargeMessage(entry.getValue)) { + largeMessageSafeEntries.add(entry) + } + }) + largeMessageSafeEntries + } +} diff --git a/samza-kv/src/main/scala/org/apache/samza/storage/kv/LoggedStore.scala b/samza-kv/src/main/scala/org/apache/samza/storage/kv/LoggedStore.scala index 9c027f02a5..fa37b21a77 100644 --- a/samza-kv/src/main/scala/org/apache/samza/storage/kv/LoggedStore.scala +++ b/samza-kv/src/main/scala/org/apache/samza/storage/kv/LoggedStore.scala @@ -19,44 +19,34 @@ package org.apache.samza.storage.kv -import org.apache.samza.SamzaException -import org.apache.samza.config.Config import org.apache.samza.util.Logging import org.apache.samza.system.{OutgoingMessageEnvelope, SystemStreamPartition} import org.apache.samza.task.MessageCollector -import scala.collection.JavaConverters.asScalaBufferConverter - /** - * A key/value store decorator that adds a changelog for any changes made to the underlying store - */ -class LoggedStore( - val store: KeyValueStore[Array[Byte], Array[Byte]], - val storeName: String, - val storeConfig: Config, + * A key/value store decorator that adds a changelog for any changes made to the underlying store + */ +class LoggedStore[K, V]( + val store: KeyValueStore[K, V], val systemStreamPartition: SystemStreamPartition, val collector: MessageCollector, - val metrics: LoggedStoreMetrics = new LoggedStoreMetrics) extends KeyValueStore[Array[Byte], Array[Byte]] with Logging { + val metrics: LoggedStoreMetrics = new LoggedStoreMetrics) extends KeyValueStore[K, V] with Logging { val systemStream = systemStreamPartition.getSystemStream val partitionId = systemStreamPartition.getPartition.getPartitionId - private val DEFAULT_CHANGELOG_MAX_MSG_SIZE_BYTES = 1000000 - private val CHANGELOG_MAX_MSG_SIZE_BYTES = "changelog.max.message.size.bytes" - private val maxMessageSize = storeConfig.getInt(CHANGELOG_MAX_MSG_SIZE_BYTES, DEFAULT_CHANGELOG_MAX_MSG_SIZE_BYTES) // slightly less than 1 MB - /* pass through methods */ - def get(key: Array[Byte]) = { + def get(key: K) = { metrics.gets.inc store.get(key) } - override def getAll(keys: java.util.List[Array[Byte]]): java.util.Map[Array[Byte], Array[Byte]] = { + override def getAll(keys: java.util.List[K]): java.util.Map[K, V] = { metrics.gets.inc(keys.size) store.getAll(keys) } - def range(from: Array[Byte], to: Array[Byte]) = { + def range(from: K, to: K) = { metrics.ranges.inc store.range(from, to) } @@ -67,22 +57,18 @@ class LoggedStore( } /** - * Perform the local update and log it out to the changelog - */ - def put(key: Array[Byte], value: Array[Byte]) { - validateMessageSize(key, value) + * Perform the local update and log it out to the changelog + */ + def put(key: K, value: V) { metrics.puts.inc collector.send(new OutgoingMessageEnvelope(systemStream, partitionId, key, value)) store.put(key, value) } /** - * Perform multiple local updates and log out all changes to the changelog - */ - def putAll(entries: java.util.List[Entry[Array[Byte], Array[Byte]]]) { - entries.asScala.foreach(entry => { - validateMessageSize(entry.getKey, entry.getValue) - }) + * Perform multiple local updates and log out all changes to the changelog + */ + def putAll(entries: java.util.List[Entry[K, V]]) { metrics.puts.inc(entries.size) val iter = entries.iterator while (iter.hasNext) { @@ -93,18 +79,18 @@ class LoggedStore( } /** - * Perform the local delete and log it out to the changelog - */ - def delete(key: Array[Byte]) { + * Perform the local delete and log it out to the changelog + */ + def delete(key: K) { metrics.deletes.inc collector.send(new OutgoingMessageEnvelope(systemStream, partitionId, key, null)) store.delete(key) } /** - * Perform the local deletes and log them out to the changelog - */ - override def deleteAll(keys: java.util.List[Array[Byte]]) = { + * Perform the local deletes and log them out to the changelog + */ + override def deleteAll(keys: java.util.List[K]) = { metrics.deletes.inc(keys.size) val keysIterator = keys.iterator while (keysIterator.hasNext) { @@ -128,14 +114,7 @@ class LoggedStore( store.close } - override def snapshot(from: Array[Byte], to: Array[Byte]): KeyValueSnapshot[Array[Byte], Array[Byte]] = { + override def snapshot(from: K, to: K): KeyValueSnapshot[K, V] = { store.snapshot(from, to) } - - private def validateMessageSize(key: Array[Byte], message: Array[Byte]): Unit = { - if (message.length > maxMessageSize) { - throw new SamzaException("RocksDB message size " + message.length + " for store " + storeName - + " was larger than the maximum allowed message size " + maxMessageSize + ".") - } - } -} +} \ No newline at end of file diff --git a/samza-kv/src/test/scala/org/apache/samza/storage/kv/TestLargeMessageSafeStore.scala b/samza-kv/src/test/scala/org/apache/samza/storage/kv/TestLargeMessageSafeStore.scala new file mode 100644 index 0000000000..dea531f0d8 --- /dev/null +++ b/samza-kv/src/test/scala/org/apache/samza/storage/kv/TestLargeMessageSafeStore.scala @@ -0,0 +1,187 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.samza.storage.kv + +import java.util._ + +import org.apache.samza.config.Config +import org.apache.samza.metrics.Counter +import org.apache.samza.serializers.{JsonSerdeV2, LongSerde, StringSerde} +import org.apache.samza.system.SystemStreamPartition +import org.apache.samza.task.MessageCollector +import org.apache.samza.{Partition, SamzaException} +import org.junit.{Before, Test} +import org.mockito.Matchers +import org.mockito.Mockito._ +import org.junit.Assert._ + +class TestLargeMessageSafeStore { + + val store = mock(classOf[KeyValueStore[Array[Byte], Array[Byte]]]) + val storeName = "testStore" + val storeConfig = mock(classOf[Config]) + val systemStreamPartition = mock(classOf[SystemStreamPartition]) + val collector = mock(classOf[MessageCollector]) + val metrics = mock(classOf[LoggedStoreMetrics]) + val maxMessageSize = 1024 + val counter = mock(classOf[Counter]) + + @Before + def setup: Unit = { + when(systemStreamPartition.getPartition).thenReturn(new Partition(1)) + when(storeConfig.getInt(Matchers.eq("changelog.max.message.size.bytes"), Matchers.any())).thenReturn(maxMessageSize) + when(metrics.puts).thenReturn(counter) + when(counter.inc()).thenReturn(1) + doNothing().when(collector).send(Matchers.any()) + doNothing().when(store).put(Matchers.any(), Matchers.any()) + doNothing().when(store).putAll(Matchers.any()) + } + + @Test + def testLargeMessagePutWithDropLargeMessageDisabled = { + val largeMessageSafeKeyValueStore = new LargeMessageSafeStore(store, storeName, false, maxMessageSize) + + val key1 = new Array[Byte](16) + val smallMessage = new Array[Byte](32) + largeMessageSafeKeyValueStore.put(key1, smallMessage) + verify(store, times(1)).put(Matchers.eq(key1), Matchers.eq(smallMessage)) + + val key2 = new Array[Byte](16) + val largeMessage = new Array[Byte](maxMessageSize + 1) + try { + largeMessageSafeKeyValueStore.put(key2, largeMessage) + fail("The test case should have failed due to a large message being passed to the changelog, but it didn't.") + } catch { + case e: SamzaException => + verify(store, times(0)).put(Matchers.eq(key2), Matchers.eq(largeMessage)) + } + } + + @Test + def testSmallMessagePutAllSuccessWithDropLargeMessageDisabled = { + val largeMessageSafeKeyValueStore = new LargeMessageSafeStore(store, storeName, false, maxMessageSize) + val key = new Array[Byte](16) + val smallMessage = new Array[Byte](32) + + val entries = new ArrayList[Entry[Array[Byte], Array[Byte]]] + entries.add(new Entry(key, smallMessage)) + + largeMessageSafeKeyValueStore.putAll(entries) + verify(store, times(1)).putAll(Matchers.eq(entries)) + verify(store, times(0)).put(Matchers.any(classOf[Array[Byte]]), Matchers.any(classOf[Array[Byte]])) + } + + @Test + def testLargeMessagePutAllFailureWithDropLargeMessageDisabled = { + val largeMessageSafeKeyValueStore = new LargeMessageSafeStore(store, storeName, false, maxMessageSize) + val key = new Array[Byte](16) + val largeMessage = new Array[Byte](maxMessageSize + 1) + + val entries = new ArrayList[Entry[Array[Byte], Array[Byte]]] + entries.add(new Entry(key, largeMessage)) + + try { + largeMessageSafeKeyValueStore.putAll(entries) + fail("The test case should have failed due to a large message being passed to the changelog, but it didn't.") + } catch { + case e: SamzaException => + verify(store, times(0)).putAll(Matchers.eq(entries)) + verify(store, times(0)).put(Matchers.any(classOf[Array[Byte]]), Matchers.any(classOf[Array[Byte]])) + } + } + + @Test + def testSmallMessagePutWithSerdeAndDropLargeMessageDisabled(): Unit = { + val largeMessageSafeKeyValueStore = new LargeMessageSafeStore(store, storeName, false, maxMessageSize) + + val longSerde = new LongSerde + val longObj = 1000L + val key = longSerde.toBytes(longObj) + + val jsonSerde = new JsonSerdeV2[java.util.HashMap[String, Object]] + val obj = new java.util.HashMap[String, Object](); + obj.put("jack", "jill") + obj.put("john", new java.lang.Integer(2)) + val smallMessage = jsonSerde.toBytes(obj) + + val entries = new ArrayList[Entry[Array[Byte], Array[Byte]]] + entries.add(new Entry(key, smallMessage)) + + largeMessageSafeKeyValueStore.putAll(entries) + verify(store, times(1)).putAll(Matchers.eq(entries)) + verify(store, times(0)).put(Matchers.any(classOf[Array[Byte]]), Matchers.any(classOf[Array[Byte]])) + } + + @Test + def testLargeMessagePutWithSerdeAndDropLargeMessageDisabled(): Unit = { + val largeMessageSafeKeyValueStore = new LargeMessageSafeStore(store, storeName, false, maxMessageSize) + + val longSerde = new LongSerde + val longObj = 1000L + val key = longSerde.toBytes(longObj) + + val stringSerde = new StringSerde + val largeString = "a" * maxMessageSize + 1 + val largeMessage = stringSerde.toBytes(largeString) + + val entries = new ArrayList[Entry[Array[Byte], Array[Byte]]] + entries.add(new Entry(key, largeMessage)) + + try{ + largeMessageSafeKeyValueStore.putAll(entries) + } + catch { + case e: SamzaException => + verify(store, times(0)).putAll(Matchers.eq(entries)) + verify(store, times(0)).put(Matchers.any(classOf[Array[Byte]]), Matchers.any(classOf[Array[Byte]])) + } + } + + @Test + def testPutSuccessWithDropLargeMessageEnabled = { + val largeMessageSafeKeyValueStore = new LargeMessageSafeStore(store, storeName, true, maxMessageSize) + val key1 = new Array[Byte](16) + val largeMessage = new Array[Byte](maxMessageSize + 1) + + largeMessageSafeKeyValueStore.put(key1, largeMessage) + + verify(store, times(0)).put(Matchers.any(classOf[Array[Byte]]), Matchers.any(classOf[Array[Byte]])) + } + + @Test + def testPutAllSuccessWithDropLargeMessageEnabled = { + val largeMessageSafeKeyValueStore = new LargeMessageSafeStore(store, storeName, true, maxMessageSize) + val key1 = new Array[Byte](16) + val largeMessage = new Array[Byte](maxMessageSize + 1) + val key2 = new Array[Byte](8) + val smallMessage = new Array[Byte](1) + + val entries = new ArrayList[Entry[Array[Byte], Array[Byte]]] + val largeMessageEntry = new Entry(key1, largeMessage) + val smallMessageEntry = new Entry(key2, smallMessage) + entries.add(largeMessageEntry) + entries.add(smallMessageEntry) + + largeMessageSafeKeyValueStore.putAll(entries) + + entries.remove(largeMessageEntry) + verify(store, times(1)).putAll(Matchers.eq(entries)) + verify(store, times(0)).put(Matchers.any(classOf[Array[Byte]]), Matchers.any(classOf[Array[Byte]])) + } +} diff --git a/samza-kv/src/test/scala/org/apache/samza/storage/kv/TestLoggedStore.scala b/samza-kv/src/test/scala/org/apache/samza/storage/kv/TestLoggedStore.scala deleted file mode 100644 index 1ffef380fa..0000000000 --- a/samza-kv/src/test/scala/org/apache/samza/storage/kv/TestLoggedStore.scala +++ /dev/null @@ -1,105 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ -package org.apache.samza.storage.kv - -import java.util._ - -import org.apache.samza.config.Config -import org.apache.samza.metrics.Counter -import org.apache.samza.system.SystemStreamPartition -import org.apache.samza.task.MessageCollector -import org.apache.samza.{Partition, SamzaException} -import org.junit.{Before, Test} -import org.mockito.Matchers -import org.mockito.Mockito._ - - -class TestLoggedStore { - - val store = mock(classOf[KeyValueStore[Array[Byte], Array[Byte]]]) - val storeName = "testStore" - val storeConfig = mock(classOf[Config]) - val systemStreamPartition = mock(classOf[SystemStreamPartition]) - val collector = mock(classOf[MessageCollector]) - val metrics = mock(classOf[LoggedStoreMetrics]) - val maxMessageSize = 1024 - val counter = mock(classOf[Counter]) - - @Before - def setup: Unit = { - when(systemStreamPartition.getPartition).thenReturn(new Partition(1)) - when(storeConfig.getInt(Matchers.eq("changelog.max.message.size.bytes"), Matchers.any())).thenReturn(maxMessageSize) - when(metrics.puts).thenReturn(counter) - when(counter.inc()).thenReturn(1) - doNothing().when(collector).send(Matchers.any()) - doNothing().when(store).put(Matchers.any(), Matchers.any()) - doNothing().when(store).putAll(Matchers.any()) - } - - @Test - def testLargeMessagePut = { - val safeLoggedStore = new LoggedStore(store, storeName, storeConfig, systemStreamPartition, collector, metrics) - - val key1 = new Array[Byte](16) - val smallMessage = new Array[Byte](32) - safeLoggedStore.put(key1, smallMessage) - verify(store, times(1)).put(Matchers.eq(key1), Matchers.eq(smallMessage)) - - val key2 = new Array[Byte](16) - val largeMessage = new Array[Byte](1025) - try { - safeLoggedStore.put(key2, largeMessage) - } catch { - case e: SamzaException => - verify(store, times(0)).put(Matchers.eq(key2), Matchers.eq(largeMessage)) - } - } - - @Test - def testSmallMessagePutAllSuccess = { - val safeLoggedStore = new LoggedStore(store, storeName, storeConfig, systemStreamPartition, collector, metrics) - val key = new Array[Byte](16) - val smallMessage = new Array[Byte](32) - - val entries = new ArrayList[Entry[Array[Byte], Array[Byte]]] - entries.add(new Entry(key, smallMessage)) - - safeLoggedStore.putAll(entries) - verify(store, times(1)).putAll(Matchers.eq(entries)) - verify(store, times(0)).put(Matchers.any(classOf[Array[Byte]]), Matchers.any(classOf[Array[Byte]])) - } - - @Test - def testLargeMessagePutAllFailure = { - val safeLoggedStore = new LoggedStore(store, storeName, storeConfig, systemStreamPartition, collector, metrics) - val key = new Array[Byte](16) - val largeMessage = new Array[Byte](1025) - - val entries = new ArrayList[Entry[Array[Byte], Array[Byte]]] - entries.add(new Entry(key, largeMessage)) - - try { - safeLoggedStore.putAll(entries) - } catch { - case e: SamzaException => - verify(store, times(0)).putAll(Matchers.eq(entries)) - verify(store, times(0)).put(Matchers.any(classOf[Array[Byte]]), Matchers.any(classOf[Array[Byte]])) - } - } -} diff --git a/samza-test/src/test/scala/org/apache/samza/storage/kv/TestLargeMessageSafeKeyValueStores.scala b/samza-test/src/test/scala/org/apache/samza/storage/kv/TestLargeMessageSafeKeyValueStores.scala new file mode 100644 index 0000000000..d5f70e8c14 --- /dev/null +++ b/samza-test/src/test/scala/org/apache/samza/storage/kv/TestLargeMessageSafeKeyValueStores.scala @@ -0,0 +1,184 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.samza.storage.kv + +import java.io.File +import java.util.{ArrayList, Arrays, Random} + +import org.apache.samza.config.MapConfig +import org.apache.samza.serializers.StringSerde +import org.apache.samza.storage.kv.TestLargeMessageSafeKeyValueStores.{BatchSize, CacheSize} +import org.apache.samza.storage.kv.inmemory.InMemoryKeyValueStore +import org.apache.samza.system.{OutgoingMessageEnvelope, SystemStreamPartition} +import org.apache.samza.task.MessageCollector +import org.apache.samza.{Partition, SamzaException} +import org.junit.runner.RunWith +import org.junit.runners.Parameterized +import org.junit.runners.Parameterized.Parameters +import org.junit.{After, Before, Test} +import org.scalatest.Assertions.fail + +@RunWith(value = classOf[Parameterized]) +class TestLargeMessageSafeKeyValueStores(typeOfStore: String, storeConfig: String, dropLargeMessageStr: String) { + + val dir = new File(System.getProperty("java.io.tmpdir"), "rocksdb-test-" + new Random().nextInt(Int.MaxValue)) + var store: KeyValueStore[String, String] = null + val stringSerde = new StringSerde + var loggedStore: KeyValueStore[Array[Byte], Array[Byte]] = null + val storeName = "testStore" + val systemStreamPartition = new SystemStreamPartition("test-system", "test-stream", new Partition(0)) + val collector = new MessageCollector { + override def send(envelope: OutgoingMessageEnvelope): Unit = {} + } + val loggedStoreMetrics = new LoggedStoreMetrics(storeName) + val dropLargeMessage = dropLargeMessageStr.toBoolean + val maxMessageSize = 1024 + + @Before + def setup() { + + val kvStore: KeyValueStore[Array[Byte], Array[Byte]] = typeOfStore match { + case "inmemory" => + new InMemoryKeyValueStore + case "rocksdb" => + new RocksDbKeyValueStore( + dir, + new org.rocksdb.Options() + .setCreateIfMissing(true) + .setCompressionType(org.rocksdb.CompressionType.SNAPPY_COMPRESSION), + new MapConfig(), + false, + "someStore") + case _ => + throw new IllegalArgumentException("Type of store undefined: " + typeOfStore) + } + + loggedStore = new LoggedStore(kvStore, systemStreamPartition, collector, loggedStoreMetrics) + + store = storeConfig match { + case "serde" => + val largeMessageSafeStore = new LargeMessageSafeStore(loggedStore, storeName, dropLargeMessage, maxMessageSize) + new SerializedKeyValueStore(largeMessageSafeStore, stringSerde, stringSerde) + case "cache-then-serde" => + var toBeSerializedStore: KeyValueStore[Array[Byte], Array[Byte]] = loggedStore + toBeSerializedStore = new LargeMessageSafeStore(loggedStore, storeName, dropLargeMessage, maxMessageSize) + val serializedStore = new SerializedKeyValueStore(toBeSerializedStore, stringSerde, stringSerde) + new CachedStore(serializedStore, CacheSize, BatchSize) + case "serde-then-cache" => + val cachedStore = new CachedStore(loggedStore, CacheSize, BatchSize) + val largeMessageSafeStore = new LargeMessageSafeStore(cachedStore, storeName, dropLargeMessage, maxMessageSize) + new SerializedKeyValueStore(largeMessageSafeStore, stringSerde, stringSerde) + case _ => + throw new IllegalArgumentException("Store config undefined: " + storeConfig) + } + store = new NullSafeKeyValueStore(store) + } + + @After + def teardown() { + store.close + if (dir != null && dir.listFiles() != null) { + for (file <- dir.listFiles) + file.delete() + dir.delete() + } + } + + @Test + def testLargeMessagePutFailureForLoggedStoreWithWrappedStore(): Unit = { + val key = "test" + val largeMessage = "x" * (maxMessageSize + 1) + + if (dropLargeMessage) { + store.put(key, largeMessage) + assert(loggedStore.get(stringSerde.toBytes(key)) == null, "The large message was stored while it shouldn't have been.") + } else { + try { + store.put(key, largeMessage) + fail("Failure since putAll() method invocation incorrectly completed.") + } catch { + case e: SamzaException => + assert(store.get(key) == null, "The large message was stored while it shouldn't have been.") + } + } + } + + @Test + def testLargeMessagePutAllFailureForLoggedStoreWithWrappedStore(): Unit = { + val key = "test" + val largeMessage = "x" * (maxMessageSize + 1) + val entries = new ArrayList[Entry[String, String]] + entries.add(new Entry(key, largeMessage)) + + if (dropLargeMessage) { + store.putAll(entries) + assert(loggedStore.get(stringSerde.toBytes(key)) == null, "The large message was stored while it shouldn't have been.") + } else { + try { + store.putAll(entries) + fail("Failure since putAll() method invocation incorrectly completed.") + } catch { + case e: SamzaException => + assert(store.get(key) == null, "The large message was stored while it shouldn't have been.") + } + } + } + + @Test + def testSmallMessagePutSuccessForLoggedStoreWithWrappedStore(): Unit = { + val key = "test" + val smallMessage = "x" * (maxMessageSize - 1) + + store.put(key, smallMessage) + assert(store.get(key).equals(smallMessage)) + } + + @Test + def testSmallMessagePutAllSuccessForLoggedStoreWithWrappedStore(): Unit = { + val key = "test" + val smallMessage = "x" * (maxMessageSize - 1) + + val entries = new ArrayList[Entry[String, String]] + entries.add(new Entry(key, smallMessage)) + + store.putAll(entries) + assert(store.get(key).equals(smallMessage)) + } +} + +object TestLargeMessageSafeKeyValueStores { + val CacheSize = 1024 + val BatchSize = 1 + + @Parameters + def parameters: java.util.Collection[Array[String]] = Arrays.asList( + //Inmemory + Array("inmemory", "serde", "true"), + Array("inmemory", "serde", "false"), + Array("inmemory", "cache-then-serde", "true"), + Array("inmemory", "serde-then-cache", "false"), + Array("inmemory", "serde-then-cache", "true"), + //RocksDB + Array("rocksdb", "serde", "true"), + Array("rocksdb", "serde", "false"), + Array("rocksdb", "cache-then-serde", "true"), + Array("rocksdb", "serde-then-cache", "false"), + Array("rocksdb", "serde-then-cache", "true") + ) +} From 8fa0dab02f2588848924d33b441cb3ff1f355fae Mon Sep 17 00:00:00 2001 From: pchhokra Date: Fri, 24 May 2019 13:43:39 -0700 Subject: [PATCH 03/10] Address review and change new Scala classes to Java --- .../versioned/jobs/configuration-table.html | 44 ++++ .../apache/samza/config/StorageConfig.scala | 6 + .../storage/kv/LargeMessageSafeStore.java | 117 +++++++++ .../kv/BaseKeyValueStorageEngineFactory.scala | 69 +++--- .../apache/samza/storage/kv/CachedStore.scala | 3 +- .../storage/kv/LargeMessageSafeStore.scala | 107 --------- .../storage/kv/TestLargeMessageSafeStore.java | 180 ++++++++++++++ .../kv/TestLargeMessageSafeStore.scala | 187 --------------- .../TestLargeMessageSafeKeyValueStores.java | 225 ++++++++++++++++++ .../TestLargeMessageSafeKeyValueStores.scala | 184 -------------- 10 files changed, 610 insertions(+), 512 deletions(-) create mode 100644 samza-kv/src/main/java/org/apache/samza/storage/kv/LargeMessageSafeStore.java delete mode 100644 samza-kv/src/main/scala/org/apache/samza/storage/kv/LargeMessageSafeStore.scala create mode 100644 samza-kv/src/test/java/org/apache/samza/storage/kv/TestLargeMessageSafeStore.java delete mode 100644 samza-kv/src/test/scala/org/apache/samza/storage/kv/TestLargeMessageSafeStore.scala create mode 100644 samza-test/src/test/java/org/apache/samza/storage/kv/TestLargeMessageSafeKeyValueStores.java delete mode 100644 samza-test/src/test/scala/org/apache/samza/storage/kv/TestLargeMessageSafeKeyValueStores.scala diff --git a/docs/learn/documentation/versioned/jobs/configuration-table.html b/docs/learn/documentation/versioned/jobs/configuration-table.html index 6887c35e08..1a85dcd5eb 100644 --- a/docs/learn/documentation/versioned/jobs/configuration-table.html +++ b/docs/learn/documentation/versioned/jobs/configuration-table.html @@ -1695,6 +1695,50 @@

Samza Configuration Reference

+ + stores.store-name.changelog.max.message.size.bytes + + + This property sets the maximum size of the messages allowed in the changelog. + The maximum allowed value and the default value both are 1 MB. + + + + + stores.store-name.expect.large.messages + + + This property, when turned on, tells the system to expect large messages to be put in the stores. + It will then look out for any large messages greater than + stores.*.changelog.max.message.size.bytes + and throw a SamzaException when it finds one, stating that the record is too large. + In the case of using CachedStore, it will serialize the message first, validate + its size and then cache it if the size is of permissible limit. + This particular case of using CachedStore causes a performance degradation since + we end up serializing every time before putting the values in the cache. + When this property is turned on, we ignore the value of + stores.*.drop.large.messages. + The default value for this config is false. When this property is not set, + stores.*.drop.large.messages + defines the behaviour to be executed. + + + + + stores.store-name.drop.large.messages + + + This property, when turned on, tells the system to drop any large messages instead of + attempting to put them in the stores. It drops all large messages greater than + stores.*.changelog.max.message.size.bytes + and continues to function as if it did not receive any large messages. + No exception is thrown. In the case of using CachedStore, when this config is + turned on, the large message is stored in the Cache but is not written to the + changelog and underlying store, resulting in an inconsistent state temporarily. + When this property is turned off, the system will not be able to handle large messages of any type. + + + Using RocksDB for key-value storage
diff --git a/samza-core/src/main/scala/org/apache/samza/config/StorageConfig.scala b/samza-core/src/main/scala/org/apache/samza/config/StorageConfig.scala index 1577e62837..88a4d9f3c0 100644 --- a/samza-core/src/main/scala/org/apache/samza/config/StorageConfig.scala +++ b/samza-core/src/main/scala/org/apache/samza/config/StorageConfig.scala @@ -38,6 +38,12 @@ object StorageConfig { val ACCESSLOG_SAMPLING_RATIO = "stores.%s.accesslog.sampling.ratio" val ACCESSLOG_ENABLED = "stores.%s.accesslog.enabled" val DEFAULT_ACCESSLOG_SAMPLING_RATIO = 50 + val CHANGELOG_MAX_MSG_SIZE_BYTES = "stores.%s.changelog.max.message.size.bytes" + val DEFAULT_CHANGELOG_MAX_MSG_SIZE_BYTES = 1000000 // slightly less than 1 MB + val EXPECT_LARGE_MESSAGES = "stores.%s.expect.large.messages" + val DEFAULT_EXPECT_LARGE_MESSAGES = false + val DROP_LARGE_MESSAGES = "stores.%s.drop.large.messages" + val DEFAULT_DROP_LARGE_MESSAGES = false implicit def Config2Storage(config: Config) = new StorageConfig(config) diff --git a/samza-kv/src/main/java/org/apache/samza/storage/kv/LargeMessageSafeStore.java b/samza-kv/src/main/java/org/apache/samza/storage/kv/LargeMessageSafeStore.java new file mode 100644 index 0000000000..5135b8b7c8 --- /dev/null +++ b/samza-kv/src/main/java/org/apache/samza/storage/kv/LargeMessageSafeStore.java @@ -0,0 +1,117 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.samza.storage.kv; + +import java.util.ArrayList; +import java.util.List; +import org.apache.samza.SamzaException; + + +public class LargeMessageSafeStore implements KeyValueStore { + + private KeyValueStore store; + private String storeName; + private Boolean dropLargeMessage; + private int maxMessageSize; + + public LargeMessageSafeStore(KeyValueStore store, String storeName, Boolean dropLargeMessage, int maxMessageSize) { + this.store = store; + this.storeName = storeName; + this.dropLargeMessage = dropLargeMessage; + this.maxMessageSize = maxMessageSize; + } + + @Override + public byte[] get(byte[] key) { + return store.get(key); + } + + @Override + public void put(byte[] key, byte[] value) { + validateMessageSize(value); + if (!isLargeMessage(value)) { + store.put(key, value); + } + } + + @Override + public void putAll(List> entries) { + entries.forEach(entry -> { + validateMessageSize(entry.getValue()); + }); + List> largeMessageSafeEntries = removeLargeMessage(entries); + store.putAll(largeMessageSafeEntries); + } + + @Override + public void delete(byte[] key) { + store.delete(key); + } + + @Override + public void deleteAll(List keys) { + store.deleteAll(keys); + } + + @Override + public KeyValueIterator range(byte[] from, byte[] to) { + return store.range(from, to); + } + + @Override + public KeyValueSnapshot snapshot(byte[] from, byte[] to) { + return store.snapshot(from, to); + } + + @Override + public KeyValueIterator all() { + return store.all(); + } + + @Override + public void close() { + store.close(); + } + + @Override + public void flush() { + store.flush(); + } + + private void validateMessageSize(byte[] message) { + if (!dropLargeMessage && isLargeMessage(message)) { + throw new SamzaException("Logged store message size " + message.length + " for store " + storeName + + " was larger than the maximum allowed message size " + maxMessageSize + "."); + } + } + + private boolean isLargeMessage(byte[] message) { + return message != null && message.length > maxMessageSize; + } + + private List> removeLargeMessage(List> entries) { + List> largeMessageSafeEntries = new ArrayList<>(); + entries.forEach(entry -> { + if (!isLargeMessage(entry.getValue())) { + largeMessageSafeEntries.add(entry); + } + }); + return largeMessageSafeEntries; + } +} diff --git a/samza-kv/src/main/scala/org/apache/samza/storage/kv/BaseKeyValueStorageEngineFactory.scala b/samza-kv/src/main/scala/org/apache/samza/storage/kv/BaseKeyValueStorageEngineFactory.scala index 37bc4e7490..d846be9290 100644 --- a/samza-kv/src/main/scala/org/apache/samza/storage/kv/BaseKeyValueStorageEngineFactory.scala +++ b/samza-kv/src/main/scala/org/apache/samza/storage/kv/BaseKeyValueStorageEngineFactory.scala @@ -23,6 +23,7 @@ import java.io.File import org.apache.samza.SamzaException import org.apache.samza.config.MetricsConfig.Config2Metrics +import org.apache.samza.config.StorageConfig import org.apache.samza.context.{ContainerContext, JobContext} import org.apache.samza.metrics.MetricsRegistry import org.apache.samza.serializers.Serde @@ -42,12 +43,6 @@ trait BaseKeyValueStorageEngineFactory[K, V] extends StorageEngineFactory[K, V] private val INMEMORY_KV_STORAGE_ENGINE_FACTORY = "org.apache.samza.storage.kv.inmemory.InMemoryKeyValueStorageEngineFactory" - private val CHANGELOG_MAX_MSG_SIZE_BYTES = "changelog.max.message.size.bytes" - private val DEFAULT_CHANGELOG_MAX_MSG_SIZE_BYTES = 1000000 // slightly less than 1 MB - private val EXPECT_LARGE_MESSAGES = "expect.large.message" - private val DEFAULT_EXPECT_LARGE_MESSAGES = false - private val DROP_LARGE_MESSAGES = "drop.large.messages" - private val DEFAULT_DROP_LARGE_MESSAGES = false /** * Return a KeyValueStore instance for the given store name, @@ -92,9 +87,9 @@ trait BaseKeyValueStorageEngineFactory[K, V] extends StorageEngineFactory[K, V] val storeFactory = storageConfig.get("factory") var storePropertiesBuilder = new StoreProperties.StorePropertiesBuilder() val accessLog = storageConfig.getBoolean("accesslog.enabled", false) - val maxMessageSize = storageConfig.getInt(CHANGELOG_MAX_MSG_SIZE_BYTES, DEFAULT_CHANGELOG_MAX_MSG_SIZE_BYTES) - val largeMessagesExpected = storageConfig.getBoolean(EXPECT_LARGE_MESSAGES, DEFAULT_EXPECT_LARGE_MESSAGES) - val dropLargeMessage = storageConfig.getBoolean(DROP_LARGE_MESSAGES, DEFAULT_DROP_LARGE_MESSAGES) + val maxMessageSize = storageConfig.getInt(StorageConfig.CHANGELOG_MAX_MSG_SIZE_BYTES, StorageConfig.DEFAULT_CHANGELOG_MAX_MSG_SIZE_BYTES) + val largeMessagesExpected = storageConfig.getBoolean(StorageConfig.EXPECT_LARGE_MESSAGES, StorageConfig.DEFAULT_EXPECT_LARGE_MESSAGES) + val dropLargeMessage = storageConfig.getBoolean(StorageConfig.DROP_LARGE_MESSAGES, StorageConfig.DEFAULT_DROP_LARGE_MESSAGES) if (storeFactory == null) { throw new SamzaException("Store factory not defined. Cannot proceed with KV store creation!") @@ -134,24 +129,15 @@ trait BaseKeyValueStorageEngineFactory[K, V] extends StorageEngineFactory[K, V] var toBeAccessLoggedStore: KeyValueStore[K, V] = null if (largeMessagesExpected) { - // Execute "if" when we expect to get large messages and user defined config -> EXPECT_LARGE_MESSAGES = true + // Execute "if" when we expect to get large messages and user defined config -> EXPECT_LARGE_MESSAGES = true, no matter the value of DROP_LARGE_MESSAGES // This code path will throw a SamzaException for large message size. // maybe wrap with caching - val maybeCachedStore = if (enableCache) { - val cachedStoreMetrics = new CachedStoreMetrics(storeName, registry) - new CachedStore(maybeLoggedStore, cacheSize, batchSize, cachedStoreMetrics) - } else { - maybeLoggedStore - } - + var maybeCachedStore = createCachedStore(enableCache, storeName, registry, maybeLoggedStore, cacheSize, batchSize) // wrap with large message checking - val largeMessageSafeKeyValueStore = new LargeMessageSafeStore(maybeCachedStore, storeName, false, maxMessageSize) - + val largeMessageSafeKeyValueStore = createLargeMessageSafeStore(maybeCachedStore, storeName, false, maxMessageSize) // wrap with serialization - val serializedMetrics = new SerializedKeyValueStoreMetrics(storeName, registry) - val serialized = new SerializedKeyValueStore[K, V](largeMessageSafeKeyValueStore, keySerde, msgSerde, serializedMetrics) - toBeAccessLoggedStore = serialized + toBeAccessLoggedStore = createSerializedStore(largeMessageSafeKeyValueStore, storeName, registry, keySerde, msgSerde) } else { @@ -161,20 +147,12 @@ trait BaseKeyValueStorageEngineFactory[K, V] extends StorageEngineFactory[K, V] // This if statement code path will ignore all large messages and continue to add all other messages to the desired stores. // wrap with large message checking - toBeSerializedStore = new LargeMessageSafeStore(maybeLoggedStore, storeName, dropLargeMessage, maxMessageSize) + toBeSerializedStore = createLargeMessageSafeStore(maybeLoggedStore, storeName, dropLargeMessage, maxMessageSize) } // wrap with serialization - val serializedMetrics = new SerializedKeyValueStoreMetrics(storeName, registry) - val serialized = new SerializedKeyValueStore[K, V](toBeSerializedStore, keySerde, msgSerde, serializedMetrics) - + val serializedStore = createSerializedStore(toBeSerializedStore, storeName, registry, keySerde, msgSerde) // maybe wrap with caching - val maybeCachedStore = if (enableCache) { - val cachedStoreMetrics = new CachedStoreMetrics(storeName, registry) - new CachedStore(serialized, cacheSize, batchSize, cachedStoreMetrics) - } else { - serialized - } - toBeAccessLoggedStore = maybeCachedStore + toBeAccessLoggedStore = createCachedStore(enableCache, storeName, registry, serializedStore, cacheSize, batchSize) } val maybeAccessLoggedStore = if (accessLog) { @@ -202,4 +180,29 @@ trait BaseKeyValueStorageEngineFactory[K, V] extends StorageEngineFactory[K, V] new KeyValueStorageEngine(storeName, storeDir, storePropertiesBuilder.build(), nullSafeStore, rawStore, keyValueStorageEngineMetrics, batchSize, () => clock.nanoTime()) } + + def createCachedStore[K, V](enableCache: Boolean, storeName: String, registry: MetricsRegistry, + underlyingStore: KeyValueStore[K, V], cacheSize: Int, batchSize: Int): KeyValueStore[K, V] = { + // maybe wrap with caching + val maybeCachedStore = if (enableCache) { + val cachedStoreMetrics = new CachedStoreMetrics(storeName, registry) + new CachedStore(underlyingStore, cacheSize, batchSize, cachedStoreMetrics) + } else { + underlyingStore + } + maybeCachedStore + } + + def createLargeMessageSafeStore(underlyingStore: KeyValueStore[Array[Byte], Array[Byte]], storeName: String, + dropLargeMessage: Boolean, maxMessageSize: Int ): LargeMessageSafeStore = { + // wrap with large message checking + new LargeMessageSafeStore(underlyingStore, storeName, dropLargeMessage, maxMessageSize) + } + + def createSerializedStore(underlyingStore: KeyValueStore[Array[Byte], Array[Byte]], storeName: String, + registry: MetricsRegistry, keySerde: Serde[K], msgSerde: Serde[V]): SerializedKeyValueStore[K, V] = { + // wrap with serialization + val serializedMetrics = new SerializedKeyValueStoreMetrics(storeName, registry) + new SerializedKeyValueStore[K, V](underlyingStore, keySerde, msgSerde, serializedMetrics) + } } diff --git a/samza-kv/src/main/scala/org/apache/samza/storage/kv/CachedStore.scala b/samza-kv/src/main/scala/org/apache/samza/storage/kv/CachedStore.scala index fa8b1b2958..959e49e12f 100644 --- a/samza-kv/src/main/scala/org/apache/samza/storage/kv/CachedStore.scala +++ b/samza-kv/src/main/scala/org/apache/samza/storage/kv/CachedStore.scala @@ -27,7 +27,8 @@ import java.util.Arrays * A write-behind caching layer around the rocksdb store. The purpose of this cache is three-fold: * 1. Batch together writes to rocksdb, this turns out to be a great optimization * 2. Avoid duplicate writes and duplicate log entries within a commit interval. i.e. if there are two updates to the same key, log only the later. - * 3. Avoid deserialization cost for gets on very common keys + * 3. Avoid deserialization cost for gets on very common keys (unless the keys itself are an array of bytes, + * in which case they don't need to be deserialized and there is no performance benefit). * * This caching does introduce a few odd corner cases :-( * 1. Items in the cache have pass-by-reference semantics but items in rocksdb have pass-by-value semantics. Modifying items after a put is a bad idea. diff --git a/samza-kv/src/main/scala/org/apache/samza/storage/kv/LargeMessageSafeStore.scala b/samza-kv/src/main/scala/org/apache/samza/storage/kv/LargeMessageSafeStore.scala deleted file mode 100644 index 8c7ab1676e..0000000000 --- a/samza-kv/src/main/scala/org/apache/samza/storage/kv/LargeMessageSafeStore.scala +++ /dev/null @@ -1,107 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ -package org.apache.samza.storage.kv -import java.util - -import org.apache.samza.SamzaException - -import scala.collection.JavaConverters.iterableAsScalaIterableConverter - -/** - * A wrapper over KeyValueStore that checks to see if any large messages have been passed to the store. - * - * @param store the store to persist to if messages are not large - * @param storeName store name - * @param dropLargeMessage a flag that determines if we want to ignore and drop large messages - * @param maxMessageSize maximum allowed size of a message, after which every message will be considered large - */ -class LargeMessageSafeStore( - store: KeyValueStore[Array[Byte], Array[Byte]], - storeName: String, - dropLargeMessage: Boolean, - maxMessageSize: Int) extends KeyValueStore[Array[Byte], Array[Byte]] { - - override def get(key: Array[Byte]): Array[Byte] = { - store.get(key) - } - - override def put(key: Array[Byte], value: Array[Byte]): Unit = { - validateMessageSize(value) - if (!isLargeMessage(value)) { - store.put(key, value) - } - } - - override def putAll(entries: util.List[Entry[Array[Byte], Array[Byte]]]): Unit = { - entries.asScala.foreach(entry => { - validateMessageSize(entry.getValue) - }) - val largeMessageSafeEntries = removeLargeMessage(entries) - store.putAll(largeMessageSafeEntries) - } - - override def delete(key: Array[Byte]): Unit = { - store.delete(key) - } - - override def deleteAll(keys: java.util.List[Array[Byte]]): Unit = { - store.deleteAll(keys) - } - - override def range(from: Array[Byte], to: Array[Byte]): KeyValueIterator[Array[Byte], Array[Byte]] = { - store.range(from, to) - } - - override def snapshot(from: Array[Byte], to: Array[Byte]): KeyValueSnapshot[Array[Byte], Array[Byte]] = { - store.snapshot(from, to) - } - - override def all(): KeyValueIterator[Array[Byte], Array[Byte]] = { - store.all() - } - - override def close(): Unit = { - store.close() - } - - override def flush(): Unit = { - store.flush() - } - - private def validateMessageSize(message: Array[Byte]): Unit = { - if (!dropLargeMessage && isLargeMessage(message)) { - throw new SamzaException("Logged store message size " + message.length + " for store " + storeName - + " was larger than the maximum allowed message size " + maxMessageSize + ".") - } - } - - private def isLargeMessage(message:Array[Byte]): Boolean = { - message != null && message.length > maxMessageSize - } - - private def removeLargeMessage(entries: util.List[Entry[Array[Byte], Array[Byte]]]): util.List[Entry[Array[Byte], Array[Byte]]] = { - val largeMessageSafeEntries = new util.ArrayList[Entry[Array[Byte], Array[Byte]]] - entries.asScala.foreach(entry => { - if (!isLargeMessage(entry.getValue)) { - largeMessageSafeEntries.add(entry) - } - }) - largeMessageSafeEntries - } -} diff --git a/samza-kv/src/test/java/org/apache/samza/storage/kv/TestLargeMessageSafeStore.java b/samza-kv/src/test/java/org/apache/samza/storage/kv/TestLargeMessageSafeStore.java new file mode 100644 index 0000000000..71d52ec970 --- /dev/null +++ b/samza-kv/src/test/java/org/apache/samza/storage/kv/TestLargeMessageSafeStore.java @@ -0,0 +1,180 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.samza.storage.kv; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import org.apache.commons.lang3.StringUtils; +import org.apache.samza.SamzaException; +import org.apache.samza.serializers.JsonSerdeV2; +import org.apache.samza.serializers.LongSerde; +import org.apache.samza.serializers.Serde; +import org.apache.samza.serializers.StringSerde; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Matchers; +import org.mockito.Mock; +import org.mockito.Mockito; +import org.mockito.runners.MockitoJUnitRunner; + + +@RunWith(MockitoJUnitRunner.class) +public class TestLargeMessageSafeStore { + + @Mock + KeyValueStore store; + int maxMessageSize = 1024; + String storeName = "testStore"; + + @Before + public void setup() { + Mockito.doNothing().when(store).put(Matchers.any(), Matchers.any()); + Mockito.doNothing().when(store).putAll(Matchers.any()); + } + + @Test + public void testLargeMessagePutWithDropLargeMessageDisabled() { + LargeMessageSafeStore largeMessageSafeKeyValueStore = new LargeMessageSafeStore(store, storeName, false, maxMessageSize); + + byte[] key1 = new byte[16]; + byte[] smallMessage = new byte[32]; + largeMessageSafeKeyValueStore.put(key1, smallMessage); + Mockito.verify(store).put(Matchers.eq(key1), Matchers.eq(smallMessage)); + + byte[] key2 = new byte[16]; + byte[] largeMessage = new byte[maxMessageSize + 1]; + try { + largeMessageSafeKeyValueStore.put(key2, largeMessage); + Assert.fail("The test case should have failed due to a large message being passed to the changelog, but it didn't."); + } catch (SamzaException e) { + Mockito.verifyZeroInteractions(store); + } + } + + @Test + public void testSmallMessagePutAllSuccessWithDropLargeMessageDisabled() { + LargeMessageSafeStore largeMessageSafeKeyValueStore = new LargeMessageSafeStore(store, storeName, false, maxMessageSize); + byte[] key = new byte[16]; + byte[] smallMessage = new byte[32]; + + List> entries = new ArrayList<>(); + entries.add(new Entry<>(key, smallMessage)); + + largeMessageSafeKeyValueStore.putAll(entries); + Mockito.verify(store).putAll(Matchers.eq(entries)); + Mockito.verify(store, Mockito.never()).put(Matchers.any(byte[].class), Matchers.any(byte[].class)); + } + + @Test + public void testLargeMessagePutAllFailureWithDropLargeMessageDisabled() { + LargeMessageSafeStore largeMessageSafeKeyValueStore = new LargeMessageSafeStore(store, storeName, false, maxMessageSize); + byte[] key = new byte[16]; + byte[] largeMessage = new byte[maxMessageSize + 1]; + + List> 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 (SamzaException e) { + Mockito.verifyZeroInteractions(store); + } + } + + @Test + public void testSmallMessagePutWithSerdeAndDropLargeMessageDisabled() { + LargeMessageSafeStore largeMessageSafeKeyValueStore = new LargeMessageSafeStore(store, storeName, false, maxMessageSize); + + Serde longSerde = new LongSerde(); + long longObj = 1000L; + byte[] key = longSerde.toBytes(longObj); + + JsonSerdeV2> jsonSerde = new JsonSerdeV2<>(); + Map obj = new HashMap<>(); + obj.put("jack", "jill"); + obj.put("john", 2); + byte[] smallMessage = jsonSerde.toBytes(obj); + + List> entries = new ArrayList<>(); + entries.add(new Entry<>(key, smallMessage)); + + largeMessageSafeKeyValueStore.putAll(entries); + Mockito.verify(store).putAll(Matchers.eq(entries)); + Mockito.verify(store, Mockito.never()).put(Matchers.any(byte[].class), Matchers.any(byte[].class)); + } + + @Test + public void testLargeMessagePutWithSerdeAndDropLargeMessageDisabled() { + LargeMessageSafeStore largeMessageSafeKeyValueStore = new LargeMessageSafeStore(store, storeName, false, maxMessageSize); + + Serde longSerde = new LongSerde(); + long longObj = 1000L; + byte[] key = longSerde.toBytes(longObj); + + Serde stringSerde = new StringSerde(); + String largeString = StringUtils.repeat("a", maxMessageSize + 1); + byte[] largeMessage = stringSerde.toBytes(largeString); + + List> entries = new ArrayList<>(); + entries.add(new Entry<>(key, largeMessage)); + + try { + largeMessageSafeKeyValueStore.putAll(entries); + } catch (SamzaException e) { + Mockito.verifyZeroInteractions(store); + } + } + + @Test + public void testPutSuccessWithDropLargeMessageEnabled() { + LargeMessageSafeStore largeMessageSafeKeyValueStore = new LargeMessageSafeStore(store, storeName, true, maxMessageSize); + byte[] key1 = new byte[16]; + byte[] largeMessage = new byte[maxMessageSize + 1]; + + largeMessageSafeKeyValueStore.put(key1, largeMessage); + + Mockito.verifyZeroInteractions(store); + } + + @Test + public void testPutAllSuccessWithDropLargeMessageEnabled() { + LargeMessageSafeStore largeMessageSafeKeyValueStore = new LargeMessageSafeStore(store, storeName, true, maxMessageSize); + byte[] key1 = new byte[16]; + byte[] largeMessage = new byte[maxMessageSize + 1]; + byte[] key2 = new byte[8]; + byte[] smallMessage = new byte[1]; + + List> entries = new ArrayList<>(); + Entry largeMessageEntry = new Entry<>(key1, largeMessage); + Entry smallMessageEntry = new Entry<>(key2, smallMessage); + entries.add(largeMessageEntry); + entries.add(smallMessageEntry); + + largeMessageSafeKeyValueStore.putAll(entries); + + entries.remove(largeMessageEntry); + Mockito.verify(store).putAll(Matchers.eq(entries)); + Mockito.verify(store, Mockito.never()).put(Matchers.any(byte[].class), Matchers.any(byte[].class)); + } +} diff --git a/samza-kv/src/test/scala/org/apache/samza/storage/kv/TestLargeMessageSafeStore.scala b/samza-kv/src/test/scala/org/apache/samza/storage/kv/TestLargeMessageSafeStore.scala deleted file mode 100644 index dea531f0d8..0000000000 --- a/samza-kv/src/test/scala/org/apache/samza/storage/kv/TestLargeMessageSafeStore.scala +++ /dev/null @@ -1,187 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ -package org.apache.samza.storage.kv - -import java.util._ - -import org.apache.samza.config.Config -import org.apache.samza.metrics.Counter -import org.apache.samza.serializers.{JsonSerdeV2, LongSerde, StringSerde} -import org.apache.samza.system.SystemStreamPartition -import org.apache.samza.task.MessageCollector -import org.apache.samza.{Partition, SamzaException} -import org.junit.{Before, Test} -import org.mockito.Matchers -import org.mockito.Mockito._ -import org.junit.Assert._ - -class TestLargeMessageSafeStore { - - val store = mock(classOf[KeyValueStore[Array[Byte], Array[Byte]]]) - val storeName = "testStore" - val storeConfig = mock(classOf[Config]) - val systemStreamPartition = mock(classOf[SystemStreamPartition]) - val collector = mock(classOf[MessageCollector]) - val metrics = mock(classOf[LoggedStoreMetrics]) - val maxMessageSize = 1024 - val counter = mock(classOf[Counter]) - - @Before - def setup: Unit = { - when(systemStreamPartition.getPartition).thenReturn(new Partition(1)) - when(storeConfig.getInt(Matchers.eq("changelog.max.message.size.bytes"), Matchers.any())).thenReturn(maxMessageSize) - when(metrics.puts).thenReturn(counter) - when(counter.inc()).thenReturn(1) - doNothing().when(collector).send(Matchers.any()) - doNothing().when(store).put(Matchers.any(), Matchers.any()) - doNothing().when(store).putAll(Matchers.any()) - } - - @Test - def testLargeMessagePutWithDropLargeMessageDisabled = { - val largeMessageSafeKeyValueStore = new LargeMessageSafeStore(store, storeName, false, maxMessageSize) - - val key1 = new Array[Byte](16) - val smallMessage = new Array[Byte](32) - largeMessageSafeKeyValueStore.put(key1, smallMessage) - verify(store, times(1)).put(Matchers.eq(key1), Matchers.eq(smallMessage)) - - val key2 = new Array[Byte](16) - val largeMessage = new Array[Byte](maxMessageSize + 1) - try { - largeMessageSafeKeyValueStore.put(key2, largeMessage) - fail("The test case should have failed due to a large message being passed to the changelog, but it didn't.") - } catch { - case e: SamzaException => - verify(store, times(0)).put(Matchers.eq(key2), Matchers.eq(largeMessage)) - } - } - - @Test - def testSmallMessagePutAllSuccessWithDropLargeMessageDisabled = { - val largeMessageSafeKeyValueStore = new LargeMessageSafeStore(store, storeName, false, maxMessageSize) - val key = new Array[Byte](16) - val smallMessage = new Array[Byte](32) - - val entries = new ArrayList[Entry[Array[Byte], Array[Byte]]] - entries.add(new Entry(key, smallMessage)) - - largeMessageSafeKeyValueStore.putAll(entries) - verify(store, times(1)).putAll(Matchers.eq(entries)) - verify(store, times(0)).put(Matchers.any(classOf[Array[Byte]]), Matchers.any(classOf[Array[Byte]])) - } - - @Test - def testLargeMessagePutAllFailureWithDropLargeMessageDisabled = { - val largeMessageSafeKeyValueStore = new LargeMessageSafeStore(store, storeName, false, maxMessageSize) - val key = new Array[Byte](16) - val largeMessage = new Array[Byte](maxMessageSize + 1) - - val entries = new ArrayList[Entry[Array[Byte], Array[Byte]]] - entries.add(new Entry(key, largeMessage)) - - try { - largeMessageSafeKeyValueStore.putAll(entries) - fail("The test case should have failed due to a large message being passed to the changelog, but it didn't.") - } catch { - case e: SamzaException => - verify(store, times(0)).putAll(Matchers.eq(entries)) - verify(store, times(0)).put(Matchers.any(classOf[Array[Byte]]), Matchers.any(classOf[Array[Byte]])) - } - } - - @Test - def testSmallMessagePutWithSerdeAndDropLargeMessageDisabled(): Unit = { - val largeMessageSafeKeyValueStore = new LargeMessageSafeStore(store, storeName, false, maxMessageSize) - - val longSerde = new LongSerde - val longObj = 1000L - val key = longSerde.toBytes(longObj) - - val jsonSerde = new JsonSerdeV2[java.util.HashMap[String, Object]] - val obj = new java.util.HashMap[String, Object](); - obj.put("jack", "jill") - obj.put("john", new java.lang.Integer(2)) - val smallMessage = jsonSerde.toBytes(obj) - - val entries = new ArrayList[Entry[Array[Byte], Array[Byte]]] - entries.add(new Entry(key, smallMessage)) - - largeMessageSafeKeyValueStore.putAll(entries) - verify(store, times(1)).putAll(Matchers.eq(entries)) - verify(store, times(0)).put(Matchers.any(classOf[Array[Byte]]), Matchers.any(classOf[Array[Byte]])) - } - - @Test - def testLargeMessagePutWithSerdeAndDropLargeMessageDisabled(): Unit = { - val largeMessageSafeKeyValueStore = new LargeMessageSafeStore(store, storeName, false, maxMessageSize) - - val longSerde = new LongSerde - val longObj = 1000L - val key = longSerde.toBytes(longObj) - - val stringSerde = new StringSerde - val largeString = "a" * maxMessageSize + 1 - val largeMessage = stringSerde.toBytes(largeString) - - val entries = new ArrayList[Entry[Array[Byte], Array[Byte]]] - entries.add(new Entry(key, largeMessage)) - - try{ - largeMessageSafeKeyValueStore.putAll(entries) - } - catch { - case e: SamzaException => - verify(store, times(0)).putAll(Matchers.eq(entries)) - verify(store, times(0)).put(Matchers.any(classOf[Array[Byte]]), Matchers.any(classOf[Array[Byte]])) - } - } - - @Test - def testPutSuccessWithDropLargeMessageEnabled = { - val largeMessageSafeKeyValueStore = new LargeMessageSafeStore(store, storeName, true, maxMessageSize) - val key1 = new Array[Byte](16) - val largeMessage = new Array[Byte](maxMessageSize + 1) - - largeMessageSafeKeyValueStore.put(key1, largeMessage) - - verify(store, times(0)).put(Matchers.any(classOf[Array[Byte]]), Matchers.any(classOf[Array[Byte]])) - } - - @Test - def testPutAllSuccessWithDropLargeMessageEnabled = { - val largeMessageSafeKeyValueStore = new LargeMessageSafeStore(store, storeName, true, maxMessageSize) - val key1 = new Array[Byte](16) - val largeMessage = new Array[Byte](maxMessageSize + 1) - val key2 = new Array[Byte](8) - val smallMessage = new Array[Byte](1) - - val entries = new ArrayList[Entry[Array[Byte], Array[Byte]]] - val largeMessageEntry = new Entry(key1, largeMessage) - val smallMessageEntry = new Entry(key2, smallMessage) - entries.add(largeMessageEntry) - entries.add(smallMessageEntry) - - largeMessageSafeKeyValueStore.putAll(entries) - - entries.remove(largeMessageEntry) - verify(store, times(1)).putAll(Matchers.eq(entries)) - verify(store, times(0)).put(Matchers.any(classOf[Array[Byte]]), Matchers.any(classOf[Array[Byte]])) - } -} diff --git a/samza-test/src/test/java/org/apache/samza/storage/kv/TestLargeMessageSafeKeyValueStores.java b/samza-test/src/test/java/org/apache/samza/storage/kv/TestLargeMessageSafeKeyValueStores.java new file mode 100644 index 0000000000..9b2d1ec95c --- /dev/null +++ b/samza-test/src/test/java/org/apache/samza/storage/kv/TestLargeMessageSafeKeyValueStores.java @@ -0,0 +1,225 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.samza.storage.kv; + +import java.io.File; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.List; +import java.util.Random; +import org.apache.commons.lang3.StringUtils; +import org.apache.samza.Partition; +import org.apache.samza.SamzaException; +import org.apache.samza.config.MapConfig; +import org.apache.samza.metrics.MetricsRegistry; +import org.apache.samza.metrics.MetricsRegistryMap; +import org.apache.samza.serializers.Serde; +import org.apache.samza.serializers.StringSerde; +import org.apache.samza.storage.kv.inmemory.InMemoryKeyValueStore; +import org.apache.samza.system.SystemStreamPartition; +import org.apache.samza.task.MessageCollector; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; +import org.rocksdb.FlushOptions; +import org.rocksdb.WriteOptions; +import scala.Int; + + +/** + * Test suite to check handling of large messages when attempting to write to the store. + */ +@RunWith(value = Parameterized.class) +public class TestLargeMessageSafeKeyValueStores { + + private String typeOfStore; + private String storeConfig; + private boolean dropLargeMessage; + + private static File dir = new File(System.getProperty("java.io.tmpdir"), "rocksdb-test-" + new Random().nextInt(Int.MaxValue())); + private static Serde stringSerde = new StringSerde(); + private static String storeName = "testStore"; + private static SystemStreamPartition systemStreamPartition = new SystemStreamPartition("test-system", "test-stream", new Partition(0)); + private static MessageCollector collector = envelope -> { + }; + private static MetricsRegistry metricsRegistry = new MetricsRegistryMap(); + private static LoggedStoreMetrics loggedStoreMetrics = new LoggedStoreMetrics(storeName, metricsRegistry); + private static KeyValueStoreMetrics keyValueStoreMetrics = new KeyValueStoreMetrics(storeName, metricsRegistry); + private static SerializedKeyValueStoreMetrics serializedKeyValueStoreMetrics = new SerializedKeyValueStoreMetrics(storeName, metricsRegistry); + private static CachedStoreMetrics cachedStoreMetrics = new CachedStoreMetrics(storeName, metricsRegistry); + private static int maxMessageSize = 1024; + private static int cacheSize = 1024; + private static int batchSize = 1; + + private KeyValueStore store = null; + private KeyValueStore loggedStore = null; + + /** + * @param typeOfStore Defines type of key-value store (Eg: "rocksdb" / "inmemory") + * @param storeConfig Defines what order we are invoking the stores in - serde / cache-then-serde / serde-then-cache + * @param dropLargeMessageStr Defines the value of the drop.large.message config which drops large messages if true + */ + public TestLargeMessageSafeKeyValueStores(String typeOfStore, String storeConfig, String dropLargeMessageStr) { + this.typeOfStore = typeOfStore; + this.storeConfig = storeConfig; + this.dropLargeMessage = Boolean.valueOf(dropLargeMessageStr); + } + + @Parameterized.Parameters + public static Collection data() { + return Arrays.asList(new String[][] { + {"inmemory", "serde", "true"}, + {"inmemory", "serde", "false"}, + {"inmemory", "cache-then-serde", "true"}, + {"inmemory", "serde-then-cache", "false"}, + {"inmemory", "serde-then-cache", "true"}, + //RocksDB + {"rocksdb", "serde", "true"}, + {"rocksdb", "serde", "false"}, + {"rocksdb", "cache-then-serde", "true"}, + {"rocksdb", "serde-then-cache", "false"}, + {"rocksdb", "serde-then-cache", "true"} + }); + } + + @Before + public void setup() { + + KeyValueStore kvStore; + switch (typeOfStore) { + case "inmemory" : { + kvStore = new InMemoryKeyValueStore(keyValueStoreMetrics); + break; + } + case "rocksdb" : { + kvStore = new RocksDbKeyValueStore(dir, new org.rocksdb.Options().setCreateIfMissing(true).setCompressionType(org.rocksdb.CompressionType.SNAPPY_COMPRESSION), + new MapConfig(), false, storeName, + new WriteOptions(), new FlushOptions(), keyValueStoreMetrics); + break; + } + default : + throw new IllegalArgumentException("Type of store undefined: " + typeOfStore); + } + + loggedStore = new LoggedStore<>(kvStore, systemStreamPartition, collector, loggedStoreMetrics); + + switch (storeConfig) { + case "serde" : { + KeyValueStore largeMessageSafeStore = + new LargeMessageSafeStore(loggedStore, storeName, dropLargeMessage, maxMessageSize); + store = new SerializedKeyValueStore<>(largeMessageSafeStore, stringSerde, stringSerde, + serializedKeyValueStoreMetrics); + break; + } + case "cache-then-serde" : { + KeyValueStore toBeSerializedStore = + new LargeMessageSafeStore(loggedStore, storeName, dropLargeMessage, maxMessageSize); + KeyValueStore serializedStore = + new SerializedKeyValueStore<>(toBeSerializedStore, stringSerde, stringSerde, serializedKeyValueStoreMetrics); + store = new CachedStore<>(serializedStore, cacheSize, batchSize, cachedStoreMetrics); + break; + } + case "serde-then-cache" : { + KeyValueStore cachedStore = + new CachedStore<>(loggedStore, cacheSize, batchSize, cachedStoreMetrics); + KeyValueStore largeMessageSafeStore = + new LargeMessageSafeStore(cachedStore, storeName, dropLargeMessage, maxMessageSize); + store = new SerializedKeyValueStore<>(largeMessageSafeStore, stringSerde, stringSerde, + serializedKeyValueStoreMetrics); + break; + } + default : + throw new IllegalArgumentException("Store config undefined: " + storeConfig); + } + store = new NullSafeKeyValueStore<>(store); + } + + @After + public void teardown() { + store.close(); + if (dir != null && dir.listFiles() != null) { + for (File file : dir.listFiles()) + file.delete(); + dir.delete(); + } + } + + @Test + public void testLargeMessagePutFailureForLoggedStoreWithWrappedStore() { + String key = "test"; + String largeMessage = StringUtils.repeat("a", maxMessageSize + 1); + + if (dropLargeMessage) { + store.put(key, largeMessage); + Assert.assertNull("The large message was stored while it shouldn't have been.", loggedStore.get(stringSerde.toBytes(key))); + } else { + try { + store.put(key, largeMessage); + Assert.fail("Failure since putAll() method invocation incorrectly completed."); + } catch (SamzaException e) { + Assert.assertNull("The large message was stored while it shouldn't have been.", store.get(key)); + } + } + } + + @Test + public void testLargeMessagePutAllFailureForLoggedStoreWithWrappedStore() { + String key = "test"; + String largeMessage = StringUtils.repeat("a", maxMessageSize + 1); + List> entries = new ArrayList<>(); + entries.add(new Entry<>(key, largeMessage)); + + if (dropLargeMessage) { + store.putAll(entries); + Assert.assertNull("The large message was stored while it shouldn't have been.", loggedStore.get(stringSerde.toBytes(key))); + } else { + try { + store.putAll(entries); + Assert.fail("Failure since putAll() method invocation incorrectly completed."); + } catch (SamzaException e) { + Assert.assertNull("The large message was stored while it shouldn't have been.", store.get(key)); + } + } + } + + @Test + public void testSmallMessagePutSuccessForLoggedStoreWithWrappedStore() { + String key = "test"; + String smallMessage = StringUtils.repeat("a", maxMessageSize - 1); + + store.put(key, smallMessage); + Assert.assertEquals(store.get(key), smallMessage); + } + + @Test + public void testSmallMessagePutAllSuccessForLoggedStoreWithWrappedStore() { + String key = "test"; + String smallMessage = StringUtils.repeat("a", maxMessageSize - 1); + + List> entries = new ArrayList<>(); + entries.add(new Entry<>(key, smallMessage)); + + store.putAll(entries); + Assert.assertEquals(store.get(key), smallMessage); + } +} diff --git a/samza-test/src/test/scala/org/apache/samza/storage/kv/TestLargeMessageSafeKeyValueStores.scala b/samza-test/src/test/scala/org/apache/samza/storage/kv/TestLargeMessageSafeKeyValueStores.scala deleted file mode 100644 index d5f70e8c14..0000000000 --- a/samza-test/src/test/scala/org/apache/samza/storage/kv/TestLargeMessageSafeKeyValueStores.scala +++ /dev/null @@ -1,184 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ -package org.apache.samza.storage.kv - -import java.io.File -import java.util.{ArrayList, Arrays, Random} - -import org.apache.samza.config.MapConfig -import org.apache.samza.serializers.StringSerde -import org.apache.samza.storage.kv.TestLargeMessageSafeKeyValueStores.{BatchSize, CacheSize} -import org.apache.samza.storage.kv.inmemory.InMemoryKeyValueStore -import org.apache.samza.system.{OutgoingMessageEnvelope, SystemStreamPartition} -import org.apache.samza.task.MessageCollector -import org.apache.samza.{Partition, SamzaException} -import org.junit.runner.RunWith -import org.junit.runners.Parameterized -import org.junit.runners.Parameterized.Parameters -import org.junit.{After, Before, Test} -import org.scalatest.Assertions.fail - -@RunWith(value = classOf[Parameterized]) -class TestLargeMessageSafeKeyValueStores(typeOfStore: String, storeConfig: String, dropLargeMessageStr: String) { - - val dir = new File(System.getProperty("java.io.tmpdir"), "rocksdb-test-" + new Random().nextInt(Int.MaxValue)) - var store: KeyValueStore[String, String] = null - val stringSerde = new StringSerde - var loggedStore: KeyValueStore[Array[Byte], Array[Byte]] = null - val storeName = "testStore" - val systemStreamPartition = new SystemStreamPartition("test-system", "test-stream", new Partition(0)) - val collector = new MessageCollector { - override def send(envelope: OutgoingMessageEnvelope): Unit = {} - } - val loggedStoreMetrics = new LoggedStoreMetrics(storeName) - val dropLargeMessage = dropLargeMessageStr.toBoolean - val maxMessageSize = 1024 - - @Before - def setup() { - - val kvStore: KeyValueStore[Array[Byte], Array[Byte]] = typeOfStore match { - case "inmemory" => - new InMemoryKeyValueStore - case "rocksdb" => - new RocksDbKeyValueStore( - dir, - new org.rocksdb.Options() - .setCreateIfMissing(true) - .setCompressionType(org.rocksdb.CompressionType.SNAPPY_COMPRESSION), - new MapConfig(), - false, - "someStore") - case _ => - throw new IllegalArgumentException("Type of store undefined: " + typeOfStore) - } - - loggedStore = new LoggedStore(kvStore, systemStreamPartition, collector, loggedStoreMetrics) - - store = storeConfig match { - case "serde" => - val largeMessageSafeStore = new LargeMessageSafeStore(loggedStore, storeName, dropLargeMessage, maxMessageSize) - new SerializedKeyValueStore(largeMessageSafeStore, stringSerde, stringSerde) - case "cache-then-serde" => - var toBeSerializedStore: KeyValueStore[Array[Byte], Array[Byte]] = loggedStore - toBeSerializedStore = new LargeMessageSafeStore(loggedStore, storeName, dropLargeMessage, maxMessageSize) - val serializedStore = new SerializedKeyValueStore(toBeSerializedStore, stringSerde, stringSerde) - new CachedStore(serializedStore, CacheSize, BatchSize) - case "serde-then-cache" => - val cachedStore = new CachedStore(loggedStore, CacheSize, BatchSize) - val largeMessageSafeStore = new LargeMessageSafeStore(cachedStore, storeName, dropLargeMessage, maxMessageSize) - new SerializedKeyValueStore(largeMessageSafeStore, stringSerde, stringSerde) - case _ => - throw new IllegalArgumentException("Store config undefined: " + storeConfig) - } - store = new NullSafeKeyValueStore(store) - } - - @After - def teardown() { - store.close - if (dir != null && dir.listFiles() != null) { - for (file <- dir.listFiles) - file.delete() - dir.delete() - } - } - - @Test - def testLargeMessagePutFailureForLoggedStoreWithWrappedStore(): Unit = { - val key = "test" - val largeMessage = "x" * (maxMessageSize + 1) - - if (dropLargeMessage) { - store.put(key, largeMessage) - assert(loggedStore.get(stringSerde.toBytes(key)) == null, "The large message was stored while it shouldn't have been.") - } else { - try { - store.put(key, largeMessage) - fail("Failure since putAll() method invocation incorrectly completed.") - } catch { - case e: SamzaException => - assert(store.get(key) == null, "The large message was stored while it shouldn't have been.") - } - } - } - - @Test - def testLargeMessagePutAllFailureForLoggedStoreWithWrappedStore(): Unit = { - val key = "test" - val largeMessage = "x" * (maxMessageSize + 1) - val entries = new ArrayList[Entry[String, String]] - entries.add(new Entry(key, largeMessage)) - - if (dropLargeMessage) { - store.putAll(entries) - assert(loggedStore.get(stringSerde.toBytes(key)) == null, "The large message was stored while it shouldn't have been.") - } else { - try { - store.putAll(entries) - fail("Failure since putAll() method invocation incorrectly completed.") - } catch { - case e: SamzaException => - assert(store.get(key) == null, "The large message was stored while it shouldn't have been.") - } - } - } - - @Test - def testSmallMessagePutSuccessForLoggedStoreWithWrappedStore(): Unit = { - val key = "test" - val smallMessage = "x" * (maxMessageSize - 1) - - store.put(key, smallMessage) - assert(store.get(key).equals(smallMessage)) - } - - @Test - def testSmallMessagePutAllSuccessForLoggedStoreWithWrappedStore(): Unit = { - val key = "test" - val smallMessage = "x" * (maxMessageSize - 1) - - val entries = new ArrayList[Entry[String, String]] - entries.add(new Entry(key, smallMessage)) - - store.putAll(entries) - assert(store.get(key).equals(smallMessage)) - } -} - -object TestLargeMessageSafeKeyValueStores { - val CacheSize = 1024 - val BatchSize = 1 - - @Parameters - def parameters: java.util.Collection[Array[String]] = Arrays.asList( - //Inmemory - Array("inmemory", "serde", "true"), - Array("inmemory", "serde", "false"), - Array("inmemory", "cache-then-serde", "true"), - Array("inmemory", "serde-then-cache", "false"), - Array("inmemory", "serde-then-cache", "true"), - //RocksDB - Array("rocksdb", "serde", "true"), - Array("rocksdb", "serde", "false"), - Array("rocksdb", "cache-then-serde", "true"), - Array("rocksdb", "serde-then-cache", "false"), - Array("rocksdb", "serde-then-cache", "true") - ) -} From 1ddcc623a63ed8aa514b4b6ad1b8a8a05c9a8d36 Mon Sep 17 00:00:00 2001 From: pchhokra Date: Thu, 30 May 2019 00:56:03 -0700 Subject: [PATCH 04/10] Address review and add a test case --- .../versioned/jobs/configuration-table.html | 6 ++-- .../apache/samza/config/StorageConfig.java | 2 +- .../TestLargeMessageSafeKeyValueStores.java | 31 ++++++++++++++----- 3 files changed, 27 insertions(+), 12 deletions(-) diff --git a/docs/learn/documentation/versioned/jobs/configuration-table.html b/docs/learn/documentation/versioned/jobs/configuration-table.html index dd1a7d694a..ce820132d2 100644 --- a/docs/learn/documentation/versioned/jobs/configuration-table.html +++ b/docs/learn/documentation/versioned/jobs/configuration-table.html @@ -1689,7 +1689,7 @@

Samza Configuration Reference

stores.store-name.changelog.max.message.size.bytes - + 1000000 This property sets the maximum size of the messages allowed in the changelog. The maximum allowed value and the default value both are 1 MB. @@ -1698,7 +1698,7 @@

Samza Configuration Reference

stores.store-name.expect.large.messages - + false This property, when turned on, tells the system to expect large messages to be put in the stores. It will then look out for any large messages greater than @@ -1718,7 +1718,7 @@

Samza Configuration Reference

stores.store-name.drop.large.messages - + false This property, when turned on, tells the system to drop any large messages instead of attempting to put them in the stores. It drops all large messages greater than diff --git a/samza-core/src/main/java/org/apache/samza/config/StorageConfig.java b/samza-core/src/main/java/org/apache/samza/config/StorageConfig.java index 9269e04766..83e4114d75 100644 --- a/samza-core/src/main/java/org/apache/samza/config/StorageConfig.java +++ b/samza-core/src/main/java/org/apache/samza/config/StorageConfig.java @@ -47,7 +47,7 @@ public class StorageConfig extends MapConfig { public static final String ACCESSLOG_STREAM_SUFFIX = "access-log"; public static final String CHANGELOG_REPLICATION_FACTOR = STORE_PREFIX + "%s.changelog.replication.factor"; public static final String CHANGELOG_MAX_MSG_SIZE_BYTES = "stores.%s.changelog.max.message.size.bytes"; - public static final int DEFAULT_CHANGELOG_MAX_MSG_SIZE_BYTES = 1000000; // slightly less than 1 MB + public static final int DEFAULT_CHANGELOG_MAX_MSG_SIZE_BYTES = 5000000; // slightly less than 5 MB public static final String EXPECT_LARGE_MESSAGES = "stores.%s.expect.large.messages"; public static final boolean DEFAULT_EXPECT_LARGE_MESSAGES = false; public static final String DROP_LARGE_MESSAGES = "stores.%s.drop.large.messages"; diff --git a/samza-test/src/test/java/org/apache/samza/storage/kv/TestLargeMessageSafeKeyValueStores.java b/samza-test/src/test/java/org/apache/samza/storage/kv/TestLargeMessageSafeKeyValueStores.java index 9b2d1ec95c..94018e2b4a 100644 --- a/samza-test/src/test/java/org/apache/samza/storage/kv/TestLargeMessageSafeKeyValueStores.java +++ b/samza-test/src/test/java/org/apache/samza/storage/kv/TestLargeMessageSafeKeyValueStores.java @@ -60,8 +60,6 @@ public class TestLargeMessageSafeKeyValueStores { private static Serde stringSerde = new StringSerde(); private static String storeName = "testStore"; private static SystemStreamPartition systemStreamPartition = new SystemStreamPartition("test-system", "test-stream", new Partition(0)); - private static MessageCollector collector = envelope -> { - }; private static MetricsRegistry metricsRegistry = new MetricsRegistryMap(); private static LoggedStoreMetrics loggedStoreMetrics = new LoggedStoreMetrics(storeName, metricsRegistry); private static KeyValueStoreMetrics keyValueStoreMetrics = new KeyValueStoreMetrics(storeName, metricsRegistry); @@ -91,12 +89,14 @@ public static Collection data() { {"inmemory", "serde", "true"}, {"inmemory", "serde", "false"}, {"inmemory", "cache-then-serde", "true"}, + {"inmemory", "cache-then-serde", "false"}, {"inmemory", "serde-then-cache", "false"}, {"inmemory", "serde-then-cache", "true"}, //RocksDB {"rocksdb", "serde", "true"}, {"rocksdb", "serde", "false"}, {"rocksdb", "cache-then-serde", "true"}, + {"rocksdb", "cache-then-serde", "false"}, {"rocksdb", "serde-then-cache", "false"}, {"rocksdb", "serde-then-cache", "true"} }); @@ -121,6 +121,13 @@ public void setup() { throw new IllegalArgumentException("Type of store undefined: " + typeOfStore); } + MessageCollector collector = envelope -> { + int messageLength = ((byte[]) envelope.getMessage()).length; + if (messageLength > maxMessageSize) { + throw new SamzaException("Logged store message size " + messageLength + " for store " + storeName + + " was larger than the maximum allowed message size " + maxMessageSize + "."); + } + }; loggedStore = new LoggedStore<>(kvStore, systemStreamPartition, collector, loggedStoreMetrics); switch (storeConfig) { @@ -132,13 +139,17 @@ public void setup() { break; } case "cache-then-serde" : { - KeyValueStore toBeSerializedStore = - new LargeMessageSafeStore(loggedStore, storeName, dropLargeMessage, maxMessageSize); + KeyValueStore toBeSerializedStore = loggedStore; + if (dropLargeMessage) { + toBeSerializedStore = new LargeMessageSafeStore(loggedStore, storeName, dropLargeMessage, maxMessageSize); + } KeyValueStore serializedStore = new SerializedKeyValueStore<>(toBeSerializedStore, stringSerde, stringSerde, serializedKeyValueStoreMetrics); store = new CachedStore<>(serializedStore, cacheSize, batchSize, cachedStoreMetrics); break; } + //For this case, the value of dropLargeMessage doesn't matter since we are testing the case when + // large messages are expected and StorageConfig.EXPECT_LARGE_MESSAGES is true. case "serde-then-cache" : { KeyValueStore cachedStore = new CachedStore<>(loggedStore, cacheSize, batchSize, cachedStoreMetrics); @@ -156,7 +167,11 @@ public void setup() { @After public void teardown() { - store.close(); + try { + store.close(); + } catch (SamzaException e) { + loggedStore.close(); + } if (dir != null && dir.listFiles() != null) { for (File file : dir.listFiles()) file.delete(); @@ -175,9 +190,9 @@ public void testLargeMessagePutFailureForLoggedStoreWithWrappedStore() { } else { try { store.put(key, largeMessage); - Assert.fail("Failure since putAll() method invocation incorrectly completed."); + Assert.fail("Failure since put() method invocation incorrectly completed."); } catch (SamzaException e) { - Assert.assertNull("The large message was stored while it shouldn't have been.", store.get(key)); + Assert.assertNull("The large message was stored while it shouldn't have been.", loggedStore.get(stringSerde.toBytes(key))); } } } @@ -197,7 +212,7 @@ public void testLargeMessagePutAllFailureForLoggedStoreWithWrappedStore() { store.putAll(entries); Assert.fail("Failure since putAll() method invocation incorrectly completed."); } catch (SamzaException e) { - Assert.assertNull("The large message was stored while it shouldn't have been.", store.get(key)); + Assert.assertNull("The large message was stored while it shouldn't have been.", loggedStore.get(stringSerde.toBytes(key))); } } } From 1ea72ff2b02951175c9b4d47bb7e9c0e4aee0235 Mon Sep 17 00:00:00 2001 From: pchhokra Date: Thu, 30 May 2019 10:39:24 -0700 Subject: [PATCH 05/10] Address review --- .../documentation/versioned/jobs/configuration-table.html | 2 +- .../samza/storage/kv/BaseKeyValueStorageEngineFactory.scala | 6 +++--- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/docs/learn/documentation/versioned/jobs/configuration-table.html b/docs/learn/documentation/versioned/jobs/configuration-table.html index ce820132d2..077d46bb6e 100644 --- a/docs/learn/documentation/versioned/jobs/configuration-table.html +++ b/docs/learn/documentation/versioned/jobs/configuration-table.html @@ -1692,7 +1692,7 @@

Samza Configuration Reference

1000000 This property sets the maximum size of the messages allowed in the changelog. - The maximum allowed value and the default value both are 1 MB. + The maximum allowed value is 5 MB and the default value is 1 MB. diff --git a/samza-kv/src/main/scala/org/apache/samza/storage/kv/BaseKeyValueStorageEngineFactory.scala b/samza-kv/src/main/scala/org/apache/samza/storage/kv/BaseKeyValueStorageEngineFactory.scala index 412638f64a..801d49a235 100644 --- a/samza-kv/src/main/scala/org/apache/samza/storage/kv/BaseKeyValueStorageEngineFactory.scala +++ b/samza-kv/src/main/scala/org/apache/samza/storage/kv/BaseKeyValueStorageEngineFactory.scala @@ -90,9 +90,9 @@ trait BaseKeyValueStorageEngineFactory[K, V] extends StorageEngineFactory[K, V] var storePropertiesBuilder = new StoreProperties.StorePropertiesBuilder() val accessLog = storageConfig.getAccessLogEnabled(storeName) - val maxMessageSize = storageConfig.getInt(StorageConfig.CHANGELOG_MAX_MSG_SIZE_BYTES, StorageConfig.DEFAULT_CHANGELOG_MAX_MSG_SIZE_BYTES) - val largeMessagesExpected = storageConfig.getBoolean(StorageConfig.EXPECT_LARGE_MESSAGES, StorageConfig.DEFAULT_EXPECT_LARGE_MESSAGES) - val dropLargeMessage = storageConfig.getBoolean(StorageConfig.DROP_LARGE_MESSAGES, StorageConfig.DEFAULT_DROP_LARGE_MESSAGES) + val maxMessageSize = storageConfigSubset.getInt(StorageConfig.CHANGELOG_MAX_MSG_SIZE_BYTES, StorageConfig.DEFAULT_CHANGELOG_MAX_MSG_SIZE_BYTES) + val largeMessagesExpected = storageConfigSubset.getBoolean(StorageConfig.EXPECT_LARGE_MESSAGES, StorageConfig.DEFAULT_EXPECT_LARGE_MESSAGES) + val dropLargeMessage = storageConfigSubset.getBoolean(StorageConfig.DROP_LARGE_MESSAGES, StorageConfig.DEFAULT_DROP_LARGE_MESSAGES) if (storeFactory.isEmpty) { throw new SamzaException("Store factory not defined. Cannot proceed with KV store creation!") From 95648eddddcc01248948dd81f9583abe0477b3e6 Mon Sep 17 00:00:00 2001 From: pchhokra Date: Thu, 6 Jun 2019 08:47:49 -0700 Subject: [PATCH 06/10] Address review --- .../versioned/jobs/configuration-table.html | 2 +- .../apache/samza/config/StorageConfig.java | 20 ++++++++++--- .../samza/config/TestStorageConfig.java | 30 +++++++++++++++++++ .../kv/BaseKeyValueStorageEngineFactory.scala | 16 ++++++---- 4 files changed, 58 insertions(+), 10 deletions(-) diff --git a/docs/learn/documentation/versioned/jobs/configuration-table.html b/docs/learn/documentation/versioned/jobs/configuration-table.html index 077d46bb6e..ef89f4a397 100644 --- a/docs/learn/documentation/versioned/jobs/configuration-table.html +++ b/docs/learn/documentation/versioned/jobs/configuration-table.html @@ -1692,7 +1692,7 @@

Samza Configuration Reference

1000000 This property sets the maximum size of the messages allowed in the changelog. - The maximum allowed value is 5 MB and the default value is 1 MB. + The maximum allowed value is 1 MB and the default value is 1 MB as well. diff --git a/samza-core/src/main/java/org/apache/samza/config/StorageConfig.java b/samza-core/src/main/java/org/apache/samza/config/StorageConfig.java index 83e4114d75..0e75ab6398 100644 --- a/samza-core/src/main/java/org/apache/samza/config/StorageConfig.java +++ b/samza-core/src/main/java/org/apache/samza/config/StorageConfig.java @@ -46,11 +46,11 @@ public class StorageConfig extends MapConfig { public static final String CHANGELOG_STREAM = STORE_PREFIX + "%s" + CHANGELOG_SUFFIX; public static final String ACCESSLOG_STREAM_SUFFIX = "access-log"; public static final String CHANGELOG_REPLICATION_FACTOR = STORE_PREFIX + "%s.changelog.replication.factor"; - public static final String CHANGELOG_MAX_MSG_SIZE_BYTES = "stores.%s.changelog.max.message.size.bytes"; - public static final int DEFAULT_CHANGELOG_MAX_MSG_SIZE_BYTES = 5000000; // slightly less than 5 MB - public static final String EXPECT_LARGE_MESSAGES = "stores.%s.expect.large.messages"; + public static final String CHANGELOG_MAX_MSG_SIZE_BYTES = STORE_PREFIX + "%s.changelog.max.message.size.bytes"; + public static final int DEFAULT_CHANGELOG_MAX_MSG_SIZE_BYTES = 1000000; // slightly less than 1 MB + public static final String EXPECT_LARGE_MESSAGES = STORE_PREFIX + "%s.expect.large.messages"; public static final boolean DEFAULT_EXPECT_LARGE_MESSAGES = false; - public static final String DROP_LARGE_MESSAGES = "stores.%s.drop.large.messages"; + public static final String DROP_LARGE_MESSAGES = STORE_PREFIX + "%s.drop.large.messages"; public static final boolean DEFAULT_DROP_LARGE_MESSAGES = false; static final String CHANGELOG_SYSTEM = "job.changelog.system"; @@ -193,6 +193,18 @@ public long getChangeLogDeleteRetentionInMs(String storeName) { return getLong(String.format(CHANGELOG_DELETE_RETENTION_MS, storeName), DEFAULT_CHANGELOG_DELETE_RETENTION_MS); } + public int getChangelogMaxMsgSizeBytes(String storeName) { + return getInt(String.format(CHANGELOG_MAX_MSG_SIZE_BYTES, storeName), DEFAULT_CHANGELOG_MAX_MSG_SIZE_BYTES); + } + + public boolean getExpectLargeMessages(String storeName) { + return getBoolean(String.format(EXPECT_LARGE_MESSAGES, storeName), DEFAULT_EXPECT_LARGE_MESSAGES); + } + + public boolean getDropLargeMessages(String storeName) { + return getBoolean(String.format(DROP_LARGE_MESSAGES, storeName), DEFAULT_DROP_LARGE_MESSAGES); + } + /** * Helper method to check if a system has a changelog attached to it. */ diff --git a/samza-core/src/test/java/org/apache/samza/config/TestStorageConfig.java b/samza-core/src/test/java/org/apache/samza/config/TestStorageConfig.java index 8fc77270b4..250f35d58e 100644 --- a/samza-core/src/test/java/org/apache/samza/config/TestStorageConfig.java +++ b/samza-core/src/test/java/org/apache/samza/config/TestStorageConfig.java @@ -251,4 +251,34 @@ public void testHasDurableStores() { String.format(StorageConfig.CHANGELOG_STREAM, STORE_NAME0), "system0.changelog-stream"))); assertTrue(storageConfig.hasDurableStores()); } + + @Test + public void testGetChangelogMaxMsgSizeBytes() { + // empty config, return default size + assertEquals(StorageConfig.DEFAULT_CHANGELOG_MAX_MSG_SIZE_BYTES, new StorageConfig(new MapConfig()).getChangelogMaxMsgSizeBytes(STORE_NAME0)); + + StorageConfig storageConfig = new StorageConfig( + new MapConfig(ImmutableMap.of(String.format(StorageConfig.CHANGELOG_MAX_MSG_SIZE_BYTES, STORE_NAME0), "10"))); + assertEquals(10, storageConfig.getChangelogMaxMsgSizeBytes(STORE_NAME0)); + } + + @Test + public void testGetExpectLargeMessages() { + // empty config, return default size + assertEquals(StorageConfig.DEFAULT_EXPECT_LARGE_MESSAGES, new StorageConfig(new MapConfig()).getExpectLargeMessages(STORE_NAME0)); + + StorageConfig storageConfig = new StorageConfig( + new MapConfig(ImmutableMap.of(String.format(StorageConfig.EXPECT_LARGE_MESSAGES, STORE_NAME0), "true"))); + assertEquals(true, storageConfig.getExpectLargeMessages(STORE_NAME0)); + } + + @Test + public void testGetDropLargeMessages() { + // empty config, return default size + assertEquals(StorageConfig.DEFAULT_DROP_LARGE_MESSAGES, new StorageConfig(new MapConfig()).getDropLargeMessages(STORE_NAME0)); + + StorageConfig storageConfig = new StorageConfig( + new MapConfig(ImmutableMap.of(String.format(StorageConfig.DROP_LARGE_MESSAGES, STORE_NAME0), "true"))); + assertEquals(true, storageConfig.getDropLargeMessages(STORE_NAME0)); + } } diff --git a/samza-kv/src/main/scala/org/apache/samza/storage/kv/BaseKeyValueStorageEngineFactory.scala b/samza-kv/src/main/scala/org/apache/samza/storage/kv/BaseKeyValueStorageEngineFactory.scala index 801d49a235..3375dc0729 100644 --- a/samza-kv/src/main/scala/org/apache/samza/storage/kv/BaseKeyValueStorageEngineFactory.scala +++ b/samza-kv/src/main/scala/org/apache/samza/storage/kv/BaseKeyValueStorageEngineFactory.scala @@ -32,7 +32,7 @@ import org.apache.samza.storage.{StorageEngine, StorageEngineFactory, StorePrope import org.apache.samza.system.SystemStreamPartition import org.apache.samza.task.MessageCollector import org.apache.samza.util.ScalaJavaUtil.JavaOptionals -import org.apache.samza.util.{HighResolutionClock, ScalaJavaUtil} +import org.apache.samza.util.{HighResolutionClock, Logging} /** * A key value storage engine factory implementation @@ -40,7 +40,7 @@ import org.apache.samza.util.{HighResolutionClock, ScalaJavaUtil} * This trait encapsulates all the steps needed to create a key value storage engine. It is meant to be extended * by the specific key value store factory implementations which will in turn override the getKVStore method. */ -trait BaseKeyValueStorageEngineFactory[K, V] extends StorageEngineFactory[K, V] { +trait BaseKeyValueStorageEngineFactory[K, V] extends StorageEngineFactory[K, V] with Logging { private val INMEMORY_KV_STORAGE_ENGINE_FACTORY = "org.apache.samza.storage.kv.inmemory.InMemoryKeyValueStorageEngineFactory" @@ -90,9 +90,15 @@ trait BaseKeyValueStorageEngineFactory[K, V] extends StorageEngineFactory[K, V] var storePropertiesBuilder = new StoreProperties.StorePropertiesBuilder() val accessLog = storageConfig.getAccessLogEnabled(storeName) - val maxMessageSize = storageConfigSubset.getInt(StorageConfig.CHANGELOG_MAX_MSG_SIZE_BYTES, StorageConfig.DEFAULT_CHANGELOG_MAX_MSG_SIZE_BYTES) - val largeMessagesExpected = storageConfigSubset.getBoolean(StorageConfig.EXPECT_LARGE_MESSAGES, StorageConfig.DEFAULT_EXPECT_LARGE_MESSAGES) - val dropLargeMessage = storageConfigSubset.getBoolean(StorageConfig.DROP_LARGE_MESSAGES, StorageConfig.DEFAULT_DROP_LARGE_MESSAGES) + var maxMessageSize = storageConfig.getChangelogMaxMsgSizeBytes(storeName) + if (maxMessageSize > StorageConfig.DEFAULT_CHANGELOG_MAX_MSG_SIZE_BYTES) { + warn("Defined " + String.format(StorageConfig.CHANGELOG_MAX_MSG_SIZE_BYTES, storeName) + " value: " + maxMessageSize + + " bytes is greater than maximum allowed value of " + StorageConfig.DEFAULT_CHANGELOG_MAX_MSG_SIZE_BYTES + + " bytes. Defaulting this value to " + StorageConfig.DEFAULT_CHANGELOG_MAX_MSG_SIZE_BYTES + " bytes instead.") + maxMessageSize = StorageConfig.DEFAULT_CHANGELOG_MAX_MSG_SIZE_BYTES + } + val largeMessagesExpected = storageConfig.getExpectLargeMessages(storeName) + val dropLargeMessage = storageConfig.getDropLargeMessages(storeName) if (storeFactory.isEmpty) { throw new SamzaException("Store factory not defined. Cannot proceed with KV store creation!") From e26a6ffec498297ce054609101ad3c22b962276d Mon Sep 17 00:00:00 2001 From: pchhokra Date: Tue, 25 Jun 2019 16:40:31 -0700 Subject: [PATCH 07/10] Address review --- .../documentation/versioned/jobs/configuration-table.html | 2 +- .../samza/storage/kv/BaseKeyValueStorageEngineFactory.scala | 6 ------ 2 files changed, 1 insertion(+), 7 deletions(-) diff --git a/docs/learn/documentation/versioned/jobs/configuration-table.html b/docs/learn/documentation/versioned/jobs/configuration-table.html index ef89f4a397..7f714bd73e 100644 --- a/docs/learn/documentation/versioned/jobs/configuration-table.html +++ b/docs/learn/documentation/versioned/jobs/configuration-table.html @@ -1692,7 +1692,7 @@

Samza Configuration Reference

1000000 This property sets the maximum size of the messages allowed in the changelog. - The maximum allowed value is 1 MB and the default value is 1 MB as well. + The default value is 1 MB. diff --git a/samza-kv/src/main/scala/org/apache/samza/storage/kv/BaseKeyValueStorageEngineFactory.scala b/samza-kv/src/main/scala/org/apache/samza/storage/kv/BaseKeyValueStorageEngineFactory.scala index 3375dc0729..51045f7e4f 100644 --- a/samza-kv/src/main/scala/org/apache/samza/storage/kv/BaseKeyValueStorageEngineFactory.scala +++ b/samza-kv/src/main/scala/org/apache/samza/storage/kv/BaseKeyValueStorageEngineFactory.scala @@ -91,12 +91,6 @@ trait BaseKeyValueStorageEngineFactory[K, V] extends StorageEngineFactory[K, V] val accessLog = storageConfig.getAccessLogEnabled(storeName) var maxMessageSize = storageConfig.getChangelogMaxMsgSizeBytes(storeName) - if (maxMessageSize > StorageConfig.DEFAULT_CHANGELOG_MAX_MSG_SIZE_BYTES) { - warn("Defined " + String.format(StorageConfig.CHANGELOG_MAX_MSG_SIZE_BYTES, storeName) + " value: " + maxMessageSize - + " bytes is greater than maximum allowed value of " + StorageConfig.DEFAULT_CHANGELOG_MAX_MSG_SIZE_BYTES - + " bytes. Defaulting this value to " + StorageConfig.DEFAULT_CHANGELOG_MAX_MSG_SIZE_BYTES + " bytes instead.") - maxMessageSize = StorageConfig.DEFAULT_CHANGELOG_MAX_MSG_SIZE_BYTES - } val largeMessagesExpected = storageConfig.getExpectLargeMessages(storeName) val dropLargeMessage = storageConfig.getDropLargeMessages(storeName) From 53fb9a593fcdd94288c9eaed6de67a7c0d47e07e Mon Sep 17 00:00:00 2001 From: pchhokra Date: Thu, 25 Jul 2019 10:18:09 -0700 Subject: [PATCH 08/10] Address review --- .../versioned/jobs/configuration-table.html | 35 +++++----- .../versioned/jobs/samza-configurations.md | 3 + .../apache/samza/config/StorageConfig.java | 8 +-- .../samza/config/TestStorageConfig.java | 8 +-- .../storage/kv/LargeMessageSafeStore.java | 55 ++++++++++++--- .../kv/BaseKeyValueStorageEngineFactory.scala | 67 ++++++++----------- .../kv/LargeMessageSafeStoreMetrics.scala | 30 +++++++++ .../storage/kv/RecordTooLargeException.java | 29 ++++++++ .../TestLargeMessageSafeKeyValueStores.java | 2 +- 9 files changed, 160 insertions(+), 77 deletions(-) create mode 100644 samza-kv/src/main/scala/org/apache/samza/storage/kv/LargeMessageSafeStoreMetrics.scala create mode 100644 samza-kv/src/main/scala/org/apache/samza/storage/kv/RecordTooLargeException.java diff --git a/docs/learn/documentation/versioned/jobs/configuration-table.html b/docs/learn/documentation/versioned/jobs/configuration-table.html index 7f714bd73e..ce4be96051 100644 --- a/docs/learn/documentation/versioned/jobs/configuration-table.html +++ b/docs/learn/documentation/versioned/jobs/configuration-table.html @@ -1689,7 +1689,7 @@

Samza Configuration Reference

stores.store-name.changelog.max.message.size.bytes - 1000000 + 1048576 This property sets the maximum size of the messages allowed in the changelog. The default value is 1 MB. @@ -1697,22 +1697,20 @@

Samza Configuration Reference

- stores.store-name.expect.large.messages + stores.store-name.disallow.large.messages false - This property, when turned on, tells the system to expect large messages to be put in the stores. - It will then look out for any large messages greater than + This property, when turned on, tells the system to expect large messages to be put in the stores + and disallows them. It looks out for any large messages greater than stores.*.changelog.max.message.size.bytes - and throw a SamzaException when it finds one, stating that the record is too large. + and throws a SamzaException when it finds one, stating that the record is too large. In the case of using CachedStore, it will serialize the message first, validate - its size and then cache it if the size is of permissible limit. - This particular case of using CachedStore causes a performance degradation since - we end up serializing every time before putting the values in the cache. - When this property is turned on, we ignore the value of - stores.*.drop.large.messages. - The default value for this config is false. When this property is not set, + its size and then cache it if the size is under the permissible limit. + Note that if enabled retroactively, this may cause a performance regression due to the pre-caching serialization. + When this property is turned on, the stores.*.drop.large.messages + configuration is ignored. The default value for this config is false. When this property is not set, stores.*.drop.large.messages - defines the behaviour to be executed. + determines the large message handling behavior. @@ -1720,14 +1718,15 @@

Samza Configuration Reference

stores.store-name.drop.large.messages false - This property, when turned on, tells the system to drop any large messages instead of - attempting to put them in the stores. It drops all large messages greater than + This property, when turned on, causes messages larger than stores.*.changelog.max.message.size.bytes - and continues to function as if it did not receive any large messages. - No exception is thrown. In the case of using CachedStore, when this config is - turned on, the large message is stored in the Cache but is not written to the + to be dropped from the underlying store and changelog. No exception is thrown when a large message is encountered. + For the case when storing messages in the cache is enabled along with this config being + turned on (look at the stores.*.object.cache.size + config), the large message is stored in the cache but is not written to the changelog and underlying store, resulting in an inconsistent state temporarily. - When this property is turned off, the system will not be able to handle large messages of any type. + When this property is turned off, large messages will be sent to the changelog topic as is, + and may cause the container to fail during commit. diff --git a/docs/learn/documentation/versioned/jobs/samza-configurations.md b/docs/learn/documentation/versioned/jobs/samza-configurations.md index 4fcc87bfae..535bfbec1b 100644 --- a/docs/learn/documentation/versioned/jobs/samza-configurations.md +++ b/docs/learn/documentation/versioned/jobs/samza-configurations.md @@ -254,6 +254,9 @@ These properties define Samza's storage mechanism for efficient [stateful stream |stores.**_store-name_**.key.serde| |If the storage engine expects keys in the store to be simple byte arrays, this [serde](../container/serialization.html) allows the stream task to access the store using another object type as key. The value of this property must be a serde-name that is registered with serializers.registry.*.class. If this property is not set, keys are passed unmodified to the storage engine (and the changelog stream, if appropriate).| |stores.**_store-name_**.msg.serde| |If the storage engine expects values in the store to be simple byte arrays, this [serde](../container/serialization.html) allows the stream task to access the store using another object type as value. The value of this property must be a serde-name that is registered with serializers.registry.*.class. If this property is not set, values are passed unmodified to the storage engine (and the changelog stream, if appropriate).| |stores.**_store-name_**.changelog| |Samza stores are local to a container. If the container fails, the contents of the store are lost. To prevent loss of data, you need to set this property to configure a changelog stream: Samza then ensures that writes to the store are replicated to this stream, and the store is restored from this stream after a failure. The value of this property is given in the form system-name.stream-name. The "system-name" part is optional. If it is omitted you must specify the system in job.changelog.system config. Any output stream can be used as changelog, but you must ensure that only one job ever writes to a given changelog stream (each instance of a job and each store needs its own changelog stream).| +|stores.**_store-name_**.changelog.max.message.size.bytes|1048576|This property sets the maximum size of the messages allowed in the changelog. The default value is 1 MB.| +|stores.**_store-name_**.disallow.large.messages|false|This property, when turned on, tells the system to expect large messages to be put in the stores and disallows them. It looks out for any large messages greater than `stores.*.changelog.max.message.size.bytes` and throws a SamzaException when it finds one, stating that the record is too large. In the case of using CachedStore, it will serialize the message first, validate its size and then cache it if the size is under the permissible limit. Note that if enabled retroactively, this may cause a performance regression due to the pre-caching serialization. When this property is turned on, the `stores.*.drop.large.messages` configuration is ignored. The default value for this config is false. When this property is not set, `stores.*.drop.large.messages` determines the large message handling behavior.| +|stores.**_store-name_**.drop.large.messages|false|This property, when turned on, causes messages larger than `stores.*.changelog.max.message.size.bytes` to be dropped from the underlying store and changelog. No exception is thrown when a large message is encountered. For the case when storing messages in the cache is enabled along with this config being turned on (look at the `stores.*.object.cache.size` config for reference), the large message is stored in the cache but is not written to the changelog and underlying store, resulting in an inconsistent state temporarily. When this property is turned off, large messages will be sent to the changelog topic as is, and may cause the container to fail during commit.| |stores.**_store-name_**.rocksdb.ttl.ms| |__For RocksDB:__ The time-to-live of the store. Please note it's not a strict TTL limit (removed only after compaction). Please use caution opening a database with and without TTL, as it might corrupt the database. Please make sure to read the [constraints](https://github.com/facebook/rocksdb/wiki/Time-to-Live) before using.| ##### [4.1 Advanced Storage Configurations](#advanced-storage-configurations) diff --git a/samza-core/src/main/java/org/apache/samza/config/StorageConfig.java b/samza-core/src/main/java/org/apache/samza/config/StorageConfig.java index 0e75ab6398..bba3750bc9 100644 --- a/samza-core/src/main/java/org/apache/samza/config/StorageConfig.java +++ b/samza-core/src/main/java/org/apache/samza/config/StorageConfig.java @@ -48,8 +48,8 @@ public class StorageConfig extends MapConfig { public static final String CHANGELOG_REPLICATION_FACTOR = STORE_PREFIX + "%s.changelog.replication.factor"; public static final String CHANGELOG_MAX_MSG_SIZE_BYTES = STORE_PREFIX + "%s.changelog.max.message.size.bytes"; public static final int DEFAULT_CHANGELOG_MAX_MSG_SIZE_BYTES = 1000000; // slightly less than 1 MB - public static final String EXPECT_LARGE_MESSAGES = STORE_PREFIX + "%s.expect.large.messages"; - public static final boolean DEFAULT_EXPECT_LARGE_MESSAGES = false; + public static final String DISALLOW_LARGE_MESSAGES = STORE_PREFIX + "%s.disallow.large.messages"; + public static final boolean DEFAULT_DISALLOW_LARGE_MESSAGES = false; public static final String DROP_LARGE_MESSAGES = STORE_PREFIX + "%s.drop.large.messages"; public static final boolean DEFAULT_DROP_LARGE_MESSAGES = false; @@ -197,8 +197,8 @@ public int getChangelogMaxMsgSizeBytes(String storeName) { return getInt(String.format(CHANGELOG_MAX_MSG_SIZE_BYTES, storeName), DEFAULT_CHANGELOG_MAX_MSG_SIZE_BYTES); } - public boolean getExpectLargeMessages(String storeName) { - return getBoolean(String.format(EXPECT_LARGE_MESSAGES, storeName), DEFAULT_EXPECT_LARGE_MESSAGES); + public boolean getDisallowLargeMessages(String storeName) { + return getBoolean(String.format(DISALLOW_LARGE_MESSAGES, storeName), DEFAULT_DISALLOW_LARGE_MESSAGES); } public boolean getDropLargeMessages(String storeName) { diff --git a/samza-core/src/test/java/org/apache/samza/config/TestStorageConfig.java b/samza-core/src/test/java/org/apache/samza/config/TestStorageConfig.java index 250f35d58e..5ec4640686 100644 --- a/samza-core/src/test/java/org/apache/samza/config/TestStorageConfig.java +++ b/samza-core/src/test/java/org/apache/samza/config/TestStorageConfig.java @@ -263,13 +263,13 @@ public void testGetChangelogMaxMsgSizeBytes() { } @Test - public void testGetExpectLargeMessages() { + public void testGetDisallowLargeMessages() { // empty config, return default size - assertEquals(StorageConfig.DEFAULT_EXPECT_LARGE_MESSAGES, new StorageConfig(new MapConfig()).getExpectLargeMessages(STORE_NAME0)); + assertEquals(StorageConfig.DEFAULT_DISALLOW_LARGE_MESSAGES, new StorageConfig(new MapConfig()).getDisallowLargeMessages(STORE_NAME0)); StorageConfig storageConfig = new StorageConfig( - new MapConfig(ImmutableMap.of(String.format(StorageConfig.EXPECT_LARGE_MESSAGES, STORE_NAME0), "true"))); - assertEquals(true, storageConfig.getExpectLargeMessages(STORE_NAME0)); + new MapConfig(ImmutableMap.of(String.format(StorageConfig.DISALLOW_LARGE_MESSAGES, STORE_NAME0), "true"))); + assertEquals(true, storageConfig.getDisallowLargeMessages(STORE_NAME0)); } @Test diff --git a/samza-kv/src/main/java/org/apache/samza/storage/kv/LargeMessageSafeStore.java b/samza-kv/src/main/java/org/apache/samza/storage/kv/LargeMessageSafeStore.java index 5135b8b7c8..f936f51b91 100644 --- a/samza-kv/src/main/java/org/apache/samza/storage/kv/LargeMessageSafeStore.java +++ b/samza-kv/src/main/java/org/apache/samza/storage/kv/LargeMessageSafeStore.java @@ -20,21 +20,30 @@ import java.util.ArrayList; import java.util.List; -import org.apache.samza.SamzaException; +import org.apache.samza.metrics.MetricsRegistryMap; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +/** + * This class checks for the size of the message being stored and either throws an exception when a large message is + * encountered or ignores the large message and continues processing, depending on how it is configured. + */ public class LargeMessageSafeStore implements KeyValueStore { - private KeyValueStore store; - private String storeName; - private Boolean dropLargeMessage; - private int maxMessageSize; + private static final Logger LOG = LoggerFactory.getLogger(LargeMessageSafeStore.class); + private final KeyValueStore store; + private final String storeName; + private final boolean dropLargeMessages; + private final int maxMessageSize; + private final LargeMessageSafeStoreMetrics largeMessageSafeStoreMetrics; - public LargeMessageSafeStore(KeyValueStore store, String storeName, Boolean dropLargeMessage, int maxMessageSize) { + public LargeMessageSafeStore(KeyValueStore store, String storeName, Boolean dropLargeMessages, int maxMessageSize) { this.store = store; this.storeName = storeName; - this.dropLargeMessage = dropLargeMessage; + this.dropLargeMessages = dropLargeMessages; this.maxMessageSize = maxMessageSize; + this.largeMessageSafeStoreMetrics = new LargeMessageSafeStoreMetrics(storeName, new MetricsRegistryMap()); } @Override @@ -42,20 +51,40 @@ public byte[] get(byte[] key) { return store.get(key); } + /** + * This function puts a message in the store after validating its size. + * It drops the large message if it has been configured to do so. + * Otherwise, it throws an exception stating that a large message was encountered. + * + * @param key the key with which the specified {@code value} is to be associated. + * @param value the value with which the specified {@code key} is to be associated. + */ @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(); } } + /** + * This function puts messages in the store after validating their size. + * It drops any large messages in the entry list if it has been configured to do so. + * Otherwise, it throws an exception stating that a large message was encountered, + * and does not put any of the messages in the store. + * + * @param entries the updated mappings to put into this key-value store. + */ @Override public void putAll(List> entries) { entries.forEach(entry -> { validateMessageSize(entry.getValue()); }); - List> largeMessageSafeEntries = removeLargeMessage(entries); + List> largeMessageSafeEntries = removeLargeMessages(entries); store.putAll(largeMessageSafeEntries); } @@ -95,8 +124,8 @@ public void flush() { } private void validateMessageSize(byte[] message) { - if (!dropLargeMessage && isLargeMessage(message)) { - throw new SamzaException("Logged store message size " + message.length + " for store " + storeName + if (!dropLargeMessages && isLargeMessage(message)) { + throw new RecordTooLargeException("The message size " + message.length + " for store " + storeName + " was larger than the maximum allowed message size " + maxMessageSize + "."); } } @@ -105,11 +134,15 @@ private boolean isLargeMessage(byte[] message) { return message != null && message.length > maxMessageSize; } - private List> removeLargeMessage(List> entries) { + private List> removeLargeMessages(List> entries) { List> largeMessageSafeEntries = new ArrayList<>(); entries.forEach(entry -> { if (!isLargeMessage(entry.getValue())) { largeMessageSafeEntries.add(entry); + } else { + LOG.info("Ignoring a large message with size " + entry.getValue().length + " since it is greater than " + + "the maximum allowed value of " + maxMessageSize); + largeMessageSafeStoreMetrics.ignoredLargeMessages().inc(); } }); return largeMessageSafeEntries; diff --git a/samza-kv/src/main/scala/org/apache/samza/storage/kv/BaseKeyValueStorageEngineFactory.scala b/samza-kv/src/main/scala/org/apache/samza/storage/kv/BaseKeyValueStorageEngineFactory.scala index 51045f7e4f..e6c3835109 100644 --- a/samza-kv/src/main/scala/org/apache/samza/storage/kv/BaseKeyValueStorageEngineFactory.scala +++ b/samza-kv/src/main/scala/org/apache/samza/storage/kv/BaseKeyValueStorageEngineFactory.scala @@ -40,7 +40,7 @@ import org.apache.samza.util.{HighResolutionClock, Logging} * This trait encapsulates all the steps needed to create a key value storage engine. It is meant to be extended * by the specific key value store factory implementations which will in turn override the getKVStore method. */ -trait BaseKeyValueStorageEngineFactory[K, V] extends StorageEngineFactory[K, V] with Logging { +trait BaseKeyValueStorageEngineFactory[K, V] extends StorageEngineFactory[K, V] { private val INMEMORY_KV_STORAGE_ENGINE_FACTORY = "org.apache.samza.storage.kv.inmemory.InMemoryKeyValueStorageEngineFactory" @@ -91,7 +91,7 @@ trait BaseKeyValueStorageEngineFactory[K, V] extends StorageEngineFactory[K, V] val accessLog = storageConfig.getAccessLogEnabled(storeName) var maxMessageSize = storageConfig.getChangelogMaxMsgSizeBytes(storeName) - val largeMessagesExpected = storageConfig.getExpectLargeMessages(storeName) + val disallowLargeMessages = storageConfig.getDisallowLargeMessages(storeName) val dropLargeMessage = storageConfig.getDropLargeMessages(storeName) if (storeFactory.isEmpty) { @@ -131,31 +131,38 @@ trait BaseKeyValueStorageEngineFactory[K, V] extends StorageEngineFactory[K, V] var toBeAccessLoggedStore: KeyValueStore[K, V] = null - if (largeMessagesExpected) { - // Execute "if" when we expect to get large messages and user defined config -> EXPECT_LARGE_MESSAGES = true, no matter the value of DROP_LARGE_MESSAGES - // This code path will throw a SamzaException for large message size. - + // If large messages are disallowed in config, then this creates a LargeMessageSafeKeyValueStore that throws a + // RecordTooLargeException when a large message is encountered. + if (disallowLargeMessages) { // maybe wrap with caching - var maybeCachedStore = createCachedStore(enableCache, storeName, registry, maybeLoggedStore, cacheSize, batchSize) + val maybeCachedStore = if (enableCache) { + createCachedStore(storeName, registry, maybeLoggedStore, cacheSize, batchSize) + } else { + maybeLoggedStore + } + // wrap with large message checking - val largeMessageSafeKeyValueStore = createLargeMessageSafeStore(maybeCachedStore, storeName, false, maxMessageSize) + val largeMessageSafeKeyValueStore = new LargeMessageSafeStore(maybeCachedStore, storeName, false, maxMessageSize) // wrap with serialization - toBeAccessLoggedStore = createSerializedStore(largeMessageSafeKeyValueStore, storeName, registry, keySerde, msgSerde) - - } else { + val serializedMetrics = new SerializedKeyValueStoreMetrics(storeName, registry) + toBeAccessLoggedStore = new SerializedKeyValueStore[K, V](largeMessageSafeKeyValueStore, keySerde, msgSerde, serializedMetrics) + } + else { var toBeSerializedStore = maybeLoggedStore if (dropLargeMessage) { - // Call LargeMessageSafeKeyValueStore when EXPECT_LARGE_MESSAGES = false but DROP_LARGE_MESSAGES = true - // This if statement code path will ignore all large messages and continue to add all other messages to the desired stores. - // wrap with large message checking - toBeSerializedStore = createLargeMessageSafeStore(maybeLoggedStore, storeName, dropLargeMessage, maxMessageSize) + toBeSerializedStore = new LargeMessageSafeStore(maybeLoggedStore, storeName, dropLargeMessage, maxMessageSize) } // wrap with serialization - val serializedStore = createSerializedStore(toBeSerializedStore, storeName, registry, keySerde, msgSerde) + val serializedMetrics = new SerializedKeyValueStoreMetrics(storeName, registry) + val serializedStore = new SerializedKeyValueStore[K, V](toBeSerializedStore, keySerde, msgSerde, serializedMetrics) // maybe wrap with caching - toBeAccessLoggedStore = createCachedStore(enableCache, storeName, registry, serializedStore, cacheSize, batchSize) + toBeAccessLoggedStore = if (enableCache) { + createCachedStore(storeName, registry, serializedStore, cacheSize, batchSize) + } else { + serializedStore + } } val maybeAccessLoggedStore = if (accessLog) { @@ -184,28 +191,10 @@ trait BaseKeyValueStorageEngineFactory[K, V] extends StorageEngineFactory[K, V] keyValueStorageEngineMetrics, batchSize, () => clock.nanoTime()) } - def createCachedStore[K, V](enableCache: Boolean, storeName: String, registry: MetricsRegistry, + def createCachedStore[K, V](storeName: String, registry: MetricsRegistry, underlyingStore: KeyValueStore[K, V], cacheSize: Int, batchSize: Int): KeyValueStore[K, V] = { - // maybe wrap with caching - val maybeCachedStore = if (enableCache) { - val cachedStoreMetrics = new CachedStoreMetrics(storeName, registry) - new CachedStore(underlyingStore, cacheSize, batchSize, cachedStoreMetrics) - } else { - underlyingStore - } - maybeCachedStore - } - - def createLargeMessageSafeStore(underlyingStore: KeyValueStore[Array[Byte], Array[Byte]], storeName: String, - dropLargeMessage: Boolean, maxMessageSize: Int ): LargeMessageSafeStore = { - // wrap with large message checking - new LargeMessageSafeStore(underlyingStore, storeName, dropLargeMessage, maxMessageSize) - } - - def createSerializedStore(underlyingStore: KeyValueStore[Array[Byte], Array[Byte]], storeName: String, - registry: MetricsRegistry, keySerde: Serde[K], msgSerde: Serde[V]): SerializedKeyValueStore[K, V] = { - // wrap with serialization - val serializedMetrics = new SerializedKeyValueStoreMetrics(storeName, registry) - new SerializedKeyValueStore[K, V](underlyingStore, keySerde, msgSerde, serializedMetrics) + // wrap with caching + val cachedStoreMetrics = new CachedStoreMetrics(storeName, registry) + new CachedStore(underlyingStore, cacheSize, batchSize, cachedStoreMetrics) } } diff --git a/samza-kv/src/main/scala/org/apache/samza/storage/kv/LargeMessageSafeStoreMetrics.scala b/samza-kv/src/main/scala/org/apache/samza/storage/kv/LargeMessageSafeStoreMetrics.scala new file mode 100644 index 0000000000..5dcbbea9a3 --- /dev/null +++ b/samza-kv/src/main/scala/org/apache/samza/storage/kv/LargeMessageSafeStoreMetrics.scala @@ -0,0 +1,30 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.samza.storage.kv + +import org.apache.samza.metrics.{MetricsHelper, MetricsRegistry, MetricsRegistryMap} + +class LargeMessageSafeStoreMetrics( + val storeName: String = "unknown", + val registry: MetricsRegistry = new MetricsRegistryMap) extends MetricsHelper { + + val ignoredLargeMessages = newCounter("ignored-large-messages") + + override def getPrefix = storeName + "-" +} diff --git a/samza-kv/src/main/scala/org/apache/samza/storage/kv/RecordTooLargeException.java b/samza-kv/src/main/scala/org/apache/samza/storage/kv/RecordTooLargeException.java new file mode 100644 index 0000000000..6d3e244575 --- /dev/null +++ b/samza-kv/src/main/scala/org/apache/samza/storage/kv/RecordTooLargeException.java @@ -0,0 +1,29 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.samza.storage.kv; + +import org.apache.samza.SamzaException; + + +public class RecordTooLargeException extends SamzaException { + + public RecordTooLargeException(String s) { + super(s); + } +} diff --git a/samza-test/src/test/java/org/apache/samza/storage/kv/TestLargeMessageSafeKeyValueStores.java b/samza-test/src/test/java/org/apache/samza/storage/kv/TestLargeMessageSafeKeyValueStores.java index 94018e2b4a..afb0efa6ee 100644 --- a/samza-test/src/test/java/org/apache/samza/storage/kv/TestLargeMessageSafeKeyValueStores.java +++ b/samza-test/src/test/java/org/apache/samza/storage/kv/TestLargeMessageSafeKeyValueStores.java @@ -149,7 +149,7 @@ public void setup() { break; } //For this case, the value of dropLargeMessage doesn't matter since we are testing the case when - // large messages are expected and StorageConfig.EXPECT_LARGE_MESSAGES is true. + // large messages are expected and StorageConfig.DISALLOW_LARGE_MESSAGES is true. case "serde-then-cache" : { KeyValueStore cachedStore = new CachedStore<>(loggedStore, cacheSize, batchSize, cachedStoreMetrics); From 723d6ffc458978a5fce4227b464ee44fc372e839 Mon Sep 17 00:00:00 2001 From: pchhokra Date: Tue, 30 Jul 2019 13:15:42 -0700 Subject: [PATCH 09/10] Address review --- .../apache/samza/config/StorageConfig.java | 2 +- .../storage/kv/LargeMessageSafeStore.java | 2 +- .../kv/BaseKeyValueStorageEngineFactory.scala | 7 ++-- .../kv/LargeMessageSafeStoreMetrics.scala | 4 +- .../storage/kv/TestLargeMessageSafeStore.java | 41 +++++++++++++------ .../TestLargeMessageSafeKeyValueStores.java | 8 ++-- 6 files changed, 40 insertions(+), 24 deletions(-) diff --git a/samza-core/src/main/java/org/apache/samza/config/StorageConfig.java b/samza-core/src/main/java/org/apache/samza/config/StorageConfig.java index bba3750bc9..9141dfe4d0 100644 --- a/samza-core/src/main/java/org/apache/samza/config/StorageConfig.java +++ b/samza-core/src/main/java/org/apache/samza/config/StorageConfig.java @@ -47,7 +47,7 @@ public class StorageConfig extends MapConfig { public static final String ACCESSLOG_STREAM_SUFFIX = "access-log"; public static final String CHANGELOG_REPLICATION_FACTOR = STORE_PREFIX + "%s.changelog.replication.factor"; public static final String CHANGELOG_MAX_MSG_SIZE_BYTES = STORE_PREFIX + "%s.changelog.max.message.size.bytes"; - public static final int DEFAULT_CHANGELOG_MAX_MSG_SIZE_BYTES = 1000000; // slightly less than 1 MB + public static final int DEFAULT_CHANGELOG_MAX_MSG_SIZE_BYTES = 1048576; // slightly less than 1 MB public static final String DISALLOW_LARGE_MESSAGES = STORE_PREFIX + "%s.disallow.large.messages"; public static final boolean DEFAULT_DISALLOW_LARGE_MESSAGES = false; public static final String DROP_LARGE_MESSAGES = STORE_PREFIX + "%s.drop.large.messages"; diff --git a/samza-kv/src/main/java/org/apache/samza/storage/kv/LargeMessageSafeStore.java b/samza-kv/src/main/java/org/apache/samza/storage/kv/LargeMessageSafeStore.java index f936f51b91..f53f9e8e83 100644 --- a/samza-kv/src/main/java/org/apache/samza/storage/kv/LargeMessageSafeStore.java +++ b/samza-kv/src/main/java/org/apache/samza/storage/kv/LargeMessageSafeStore.java @@ -38,7 +38,7 @@ public class LargeMessageSafeStore implements KeyValueStore { private final int maxMessageSize; private final LargeMessageSafeStoreMetrics largeMessageSafeStoreMetrics; - public LargeMessageSafeStore(KeyValueStore store, String storeName, Boolean dropLargeMessages, int maxMessageSize) { + public LargeMessageSafeStore(KeyValueStore store, String storeName, boolean dropLargeMessages, int maxMessageSize) { this.store = store; this.storeName = storeName; this.dropLargeMessages = dropLargeMessages; diff --git a/samza-kv/src/main/scala/org/apache/samza/storage/kv/BaseKeyValueStorageEngineFactory.scala b/samza-kv/src/main/scala/org/apache/samza/storage/kv/BaseKeyValueStorageEngineFactory.scala index e6c3835109..0f4244168a 100644 --- a/samza-kv/src/main/scala/org/apache/samza/storage/kv/BaseKeyValueStorageEngineFactory.scala +++ b/samza-kv/src/main/scala/org/apache/samza/storage/kv/BaseKeyValueStorageEngineFactory.scala @@ -149,10 +149,11 @@ trait BaseKeyValueStorageEngineFactory[K, V] extends StorageEngineFactory[K, V] } else { - var toBeSerializedStore = maybeLoggedStore - if (dropLargeMessage) { + val toBeSerializedStore = if (dropLargeMessage) { // wrap with large message checking - toBeSerializedStore = new LargeMessageSafeStore(maybeLoggedStore, storeName, dropLargeMessage, maxMessageSize) + new LargeMessageSafeStore(maybeLoggedStore, storeName, dropLargeMessage, maxMessageSize) + } else { + maybeLoggedStore } // wrap with serialization val serializedMetrics = new SerializedKeyValueStoreMetrics(storeName, registry) diff --git a/samza-kv/src/main/scala/org/apache/samza/storage/kv/LargeMessageSafeStoreMetrics.scala b/samza-kv/src/main/scala/org/apache/samza/storage/kv/LargeMessageSafeStoreMetrics.scala index 5dcbbea9a3..015f2da9d4 100644 --- a/samza-kv/src/main/scala/org/apache/samza/storage/kv/LargeMessageSafeStoreMetrics.scala +++ b/samza-kv/src/main/scala/org/apache/samza/storage/kv/LargeMessageSafeStoreMetrics.scala @@ -21,8 +21,8 @@ package org.apache.samza.storage.kv import org.apache.samza.metrics.{MetricsHelper, MetricsRegistry, MetricsRegistryMap} class LargeMessageSafeStoreMetrics( - val storeName: String = "unknown", - val registry: MetricsRegistry = new MetricsRegistryMap) extends MetricsHelper { + val storeName: String, + val registry: MetricsRegistry) extends MetricsHelper { val ignoredLargeMessages = newCounter("ignored-large-messages") diff --git a/samza-kv/src/test/java/org/apache/samza/storage/kv/TestLargeMessageSafeStore.java b/samza-kv/src/test/java/org/apache/samza/storage/kv/TestLargeMessageSafeStore.java index 71d52ec970..f1a303c595 100644 --- a/samza-kv/src/test/java/org/apache/samza/storage/kv/TestLargeMessageSafeStore.java +++ b/samza-kv/src/test/java/org/apache/samza/storage/kv/TestLargeMessageSafeStore.java @@ -23,7 +23,6 @@ import java.util.List; import java.util.Map; import org.apache.commons.lang3.StringUtils; -import org.apache.samza.SamzaException; import org.apache.samza.serializers.JsonSerdeV2; import org.apache.samza.serializers.LongSerde; import org.apache.samza.serializers.Serde; @@ -53,20 +52,25 @@ public void setup() { } @Test - public void testLargeMessagePutWithDropLargeMessageDisabled() { + public void testSmallMessagePut() { LargeMessageSafeStore largeMessageSafeKeyValueStore = new LargeMessageSafeStore(store, storeName, false, maxMessageSize); - byte[] key1 = new byte[16]; + byte[] key = new byte[16]; byte[] smallMessage = new byte[32]; - largeMessageSafeKeyValueStore.put(key1, smallMessage); - Mockito.verify(store).put(Matchers.eq(key1), Matchers.eq(smallMessage)); + 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[] key2 = new byte[16]; + byte[] key = new byte[16]; byte[] largeMessage = new byte[maxMessageSize + 1]; try { - largeMessageSafeKeyValueStore.put(key2, largeMessage); + 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 (SamzaException e) { + } catch (RecordTooLargeException e) { Mockito.verifyZeroInteractions(store); } } @@ -97,7 +101,7 @@ public void testLargeMessagePutAllFailureWithDropLargeMessageDisabled() { 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 (SamzaException e) { + } catch (RecordTooLargeException e) { Mockito.verifyZeroInteractions(store); } } @@ -141,18 +145,29 @@ public void testLargeMessagePutWithSerdeAndDropLargeMessageDisabled() { try { largeMessageSafeKeyValueStore.putAll(entries); - } catch (SamzaException e) { + } catch (RecordTooLargeException e) { Mockito.verifyZeroInteractions(store); } } @Test - public void testPutSuccessWithDropLargeMessageEnabled() { + public void testSmallMessagePutSuccessWithDropLargeMessageEnabled() { LargeMessageSafeStore largeMessageSafeKeyValueStore = new LargeMessageSafeStore(store, storeName, true, maxMessageSize); - byte[] key1 = new byte[16]; + 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(key1, largeMessage); + largeMessageSafeKeyValueStore.put(key, largeMessage); Mockito.verifyZeroInteractions(store); } diff --git a/samza-test/src/test/java/org/apache/samza/storage/kv/TestLargeMessageSafeKeyValueStores.java b/samza-test/src/test/java/org/apache/samza/storage/kv/TestLargeMessageSafeKeyValueStores.java index afb0efa6ee..f6cc400cf3 100644 --- a/samza-test/src/test/java/org/apache/samza/storage/kv/TestLargeMessageSafeKeyValueStores.java +++ b/samza-test/src/test/java/org/apache/samza/storage/kv/TestLargeMessageSafeKeyValueStores.java @@ -180,7 +180,7 @@ public void teardown() { } @Test - public void testLargeMessagePutFailureForLoggedStoreWithWrappedStore() { + public void testLargeMessagePut() { String key = "test"; String largeMessage = StringUtils.repeat("a", maxMessageSize + 1); @@ -198,7 +198,7 @@ public void testLargeMessagePutFailureForLoggedStoreWithWrappedStore() { } @Test - public void testLargeMessagePutAllFailureForLoggedStoreWithWrappedStore() { + public void testLargeMessagePutAll() { String key = "test"; String largeMessage = StringUtils.repeat("a", maxMessageSize + 1); List> entries = new ArrayList<>(); @@ -218,7 +218,7 @@ public void testLargeMessagePutAllFailureForLoggedStoreWithWrappedStore() { } @Test - public void testSmallMessagePutSuccessForLoggedStoreWithWrappedStore() { + public void testSmallMessagePut() { String key = "test"; String smallMessage = StringUtils.repeat("a", maxMessageSize - 1); @@ -227,7 +227,7 @@ public void testSmallMessagePutSuccessForLoggedStoreWithWrappedStore() { } @Test - public void testSmallMessagePutAllSuccessForLoggedStoreWithWrappedStore() { + public void testSmallMessagePutAll() { String key = "test"; String smallMessage = StringUtils.repeat("a", maxMessageSize - 1); From 2d8a5838a89ea603856570728b5b76fa3c3c5af9 Mon Sep 17 00:00:00 2001 From: pchhokra Date: Tue, 30 Jul 2019 16:05:22 -0700 Subject: [PATCH 10/10] Address review --- .../src/main/java/org/apache/samza/config/StorageConfig.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/samza-core/src/main/java/org/apache/samza/config/StorageConfig.java b/samza-core/src/main/java/org/apache/samza/config/StorageConfig.java index 9141dfe4d0..c4ff478acf 100644 --- a/samza-core/src/main/java/org/apache/samza/config/StorageConfig.java +++ b/samza-core/src/main/java/org/apache/samza/config/StorageConfig.java @@ -47,7 +47,7 @@ public class StorageConfig extends MapConfig { public static final String ACCESSLOG_STREAM_SUFFIX = "access-log"; public static final String CHANGELOG_REPLICATION_FACTOR = STORE_PREFIX + "%s.changelog.replication.factor"; public static final String CHANGELOG_MAX_MSG_SIZE_BYTES = STORE_PREFIX + "%s.changelog.max.message.size.bytes"; - public static final int DEFAULT_CHANGELOG_MAX_MSG_SIZE_BYTES = 1048576; // slightly less than 1 MB + public static final int DEFAULT_CHANGELOG_MAX_MSG_SIZE_BYTES = 1048576; public static final String DISALLOW_LARGE_MESSAGES = STORE_PREFIX + "%s.disallow.large.messages"; public static final boolean DEFAULT_DISALLOW_LARGE_MESSAGES = false; public static final String DROP_LARGE_MESSAGES = STORE_PREFIX + "%s.drop.large.messages";