diff --git a/docs/learn/documentation/versioned/jobs/configuration-table.html b/docs/learn/documentation/versioned/jobs/configuration-table.html
index d44cacd36b..ce4be96051 100644
--- a/docs/learn/documentation/versioned/jobs/configuration-table.html
+++ b/docs/learn/documentation/versioned/jobs/configuration-table.html
@@ -1687,6 +1687,49 @@ Samza Configuration Reference
+
+ | 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), 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.
+ |
+
+
Using RocksDB for key-value storage
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 bc38e5675b..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
@@ -46,6 +46,12 @@ 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 = STORE_PREFIX + "%s.changelog.max.message.size.bytes";
+ 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";
+ public static final boolean DEFAULT_DROP_LARGE_MESSAGES = false;
static final String CHANGELOG_SYSTEM = "job.changelog.system";
static final String CHANGELOG_DELETE_RETENTION_MS = STORE_PREFIX + "%s.changelog.delete.retention.ms";
@@ -187,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 getDisallowLargeMessages(String storeName) {
+ return getBoolean(String.format(DISALLOW_LARGE_MESSAGES, storeName), DEFAULT_DISALLOW_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..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
@@ -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 testGetDisallowLargeMessages() {
+ // empty config, return default size
+ assertEquals(StorageConfig.DEFAULT_DISALLOW_LARGE_MESSAGES, new StorageConfig(new MapConfig()).getDisallowLargeMessages(STORE_NAME0));
+
+ StorageConfig storageConfig = new StorageConfig(
+ new MapConfig(ImmutableMap.of(String.format(StorageConfig.DISALLOW_LARGE_MESSAGES, STORE_NAME0), "true")));
+ assertEquals(true, storageConfig.getDisallowLargeMessages(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/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..f53f9e8e83
--- /dev/null
+++ b/samza-kv/src/main/java/org/apache/samza/storage/kv/LargeMessageSafeStore.java
@@ -0,0 +1,150 @@
+/*
+ * 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.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 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 dropLargeMessages, int maxMessageSize) {
+ this.store = store;
+ this.storeName = storeName;
+ this.dropLargeMessages = dropLargeMessages;
+ this.maxMessageSize = maxMessageSize;
+ this.largeMessageSafeStoreMetrics = new LargeMessageSafeStoreMetrics(storeName, new MetricsRegistryMap());
+ }
+
+ @Override
+ 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 = removeLargeMessages(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 (!dropLargeMessages && isLargeMessage(message)) {
+ throw new RecordTooLargeException("The 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> 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 fdb473a7d1..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
@@ -32,18 +32,19 @@ 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
- *
- * 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"
+
/**
* Return a KeyValueStore instance for the given store name,
* which will be used as the underlying raw store
@@ -89,6 +90,10 @@ trait BaseKeyValueStorageEngineFactory[K, V] extends StorageEngineFactory[K, V]
var storePropertiesBuilder = new StoreProperties.StorePropertiesBuilder()
val accessLog = storageConfig.getAccessLogEnabled(storeName)
+ var maxMessageSize = storageConfig.getChangelogMaxMsgSizeBytes(storeName)
+ val disallowLargeMessages = storageConfig.getDisallowLargeMessages(storeName)
+ val dropLargeMessage = storageConfig.getDropLargeMessages(storeName)
+
if (storeFactory.isEmpty) {
throw new SamzaException("Store factory not defined. Cannot proceed with KV store creation!")
}
@@ -124,22 +129,47 @@ trait BaseKeyValueStorageEngineFactory[K, V] extends StorageEngineFactory[K, V]
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
- // maybe wrap with caching
- val maybeCachedStore = if (enableCache) {
- val cachedStoreMetrics = new CachedStoreMetrics(storeName, registry)
- new CachedStore(serialized, cacheSize, batchSize, cachedStoreMetrics)
- } else {
- serialized
+ // 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
+ val maybeCachedStore = if (enableCache) {
+ createCachedStore(storeName, registry, maybeLoggedStore, cacheSize, batchSize)
+ } else {
+ maybeLoggedStore
+ }
+
+ // wrap with large message checking
+ val largeMessageSafeKeyValueStore = new LargeMessageSafeStore(maybeCachedStore, storeName, false, maxMessageSize)
+ // wrap with serialization
+ val serializedMetrics = new SerializedKeyValueStoreMetrics(storeName, registry)
+ toBeAccessLoggedStore = new SerializedKeyValueStore[K, V](largeMessageSafeKeyValueStore, keySerde, msgSerde, serializedMetrics)
+
+ }
+ else {
+ val toBeSerializedStore = if (dropLargeMessage) {
+ // wrap with large message checking
+ new LargeMessageSafeStore(maybeLoggedStore, storeName, dropLargeMessage, maxMessageSize)
+ } else {
+ maybeLoggedStore
+ }
+ // wrap with serialization
+ val serializedMetrics = new SerializedKeyValueStoreMetrics(storeName, registry)
+ val serializedStore = new SerializedKeyValueStore[K, V](toBeSerializedStore, keySerde, msgSerde, serializedMetrics)
+ // maybe wrap with caching
+ toBeAccessLoggedStore = if (enableCache) {
+ createCachedStore(storeName, registry, serializedStore, cacheSize, batchSize)
+ } else {
+ serializedStore
+ }
}
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
@@ -162,4 +192,10 @@ trait BaseKeyValueStorageEngineFactory[K, V] extends StorageEngineFactory[K, V]
keyValueStorageEngineMetrics, batchSize, () => clock.nanoTime())
}
+ def createCachedStore[K, V](storeName: String, registry: MetricsRegistry,
+ underlyingStore: KeyValueStore[K, V], cacheSize: Int, batchSize: Int): KeyValueStore[K, V] = {
+ // 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/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/LargeMessageSafeStoreMetrics.scala b/samza-kv/src/main/scala/org/apache/samza/storage/kv/LargeMessageSafeStoreMetrics.scala
new file mode 100644
index 0000000000..015f2da9d4
--- /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,
+ val registry: MetricsRegistry) 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/LoggedStore.scala b/samza-kv/src/main/scala/org/apache/samza/storage/kv/LoggedStore.scala
index e5f4ca43bf..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
@@ -24,8 +24,8 @@ import org.apache.samza.system.{OutgoingMessageEnvelope, SystemStreamPartition}
import org.apache.samza.task.MessageCollector
/**
- * A key/value store decorator that adds a changelog for any changes made to the underlying store
- */
+ * 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,
@@ -57,8 +57,8 @@ class LoggedStore[K, V](
}
/**
- * Perform the local update and log it out to the changelog
- */
+ * 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))
@@ -66,8 +66,8 @@ class LoggedStore[K, V](
}
/**
- * Perform multiple local updates and log out all changes to the changelog
- */
+ * 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
@@ -79,8 +79,8 @@ class LoggedStore[K, V](
}
/**
- * Perform the local delete and log it out to the changelog
- */
+ * 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))
@@ -88,8 +88,8 @@ class LoggedStore[K, V](
}
/**
- * Perform the local deletes and log them out to the changelog
- */
+ * 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
@@ -117,4 +117,4 @@ class LoggedStore[K, V](
override def snapshot(from: K, to: K): KeyValueSnapshot[K, V] = {
store.snapshot(from, to)
}
-}
+}
\ No newline at end of file
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-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..f1a303c595
--- /dev/null
+++ b/samza-kv/src/test/java/org/apache/samza/storage/kv/TestLargeMessageSafeStore.java
@@ -0,0 +1,195 @@
+/*
+ * 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.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 testSmallMessagePut() {
+ LargeMessageSafeStore largeMessageSafeKeyValueStore = new LargeMessageSafeStore(store, storeName, false, maxMessageSize);
+
+ byte[] key = new byte[16];
+ byte[] smallMessage = new byte[32];
+ largeMessageSafeKeyValueStore.put(key, smallMessage);
+ Mockito.verify(store).put(Matchers.eq(key), Matchers.eq(smallMessage));
+ }
+
+ @Test
+ public void testLargeMessagePutWithDropLargeMessageDisabled() {
+ LargeMessageSafeStore largeMessageSafeKeyValueStore = new LargeMessageSafeStore(store, storeName, false, maxMessageSize);
+
+ byte[] key = new byte[16];
+ byte[] largeMessage = new byte[maxMessageSize + 1];
+ try {
+ largeMessageSafeKeyValueStore.put(key, largeMessage);
+ Assert.fail("The test case should have failed due to a large message being passed to the changelog, but it didn't.");
+ } catch (RecordTooLargeException e) {
+ Mockito.verifyZeroInteractions(store);
+ }
+ }
+
+ @Test
+ public void 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 (RecordTooLargeException 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 (RecordTooLargeException e) {
+ Mockito.verifyZeroInteractions(store);
+ }
+ }
+
+ @Test
+ public void testSmallMessagePutSuccessWithDropLargeMessageEnabled() {
+ LargeMessageSafeStore largeMessageSafeKeyValueStore = new LargeMessageSafeStore(store, storeName, true, maxMessageSize);
+ byte[] key = new byte[16];
+ byte[] smallMessage = new byte[32];
+
+ largeMessageSafeKeyValueStore.put(key, smallMessage);
+
+ Mockito.verify(store).put(Matchers.eq(key), Matchers.eq(smallMessage));
+ }
+
+ @Test
+ public void testLargeMessagePutSuccessWithDropLargeMessageEnabled() {
+ LargeMessageSafeStore largeMessageSafeKeyValueStore = new LargeMessageSafeStore(store, storeName, true, maxMessageSize);
+ byte[] key = new byte[16];
+ byte[] largeMessage = new byte[maxMessageSize + 1];
+
+ largeMessageSafeKeyValueStore.put(key, largeMessage);
+
+ Mockito.verifyZeroInteractions(store);
+ }
+
+ @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-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..f6cc400cf3
--- /dev/null
+++ b/samza-test/src/test/java/org/apache/samza/storage/kv/TestLargeMessageSafeKeyValueStores.java
@@ -0,0 +1,240 @@
+/*
+ * 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 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", "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"}
+ });
+ }
+
+ @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);
+ }
+
+ 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) {
+ case "serde" : {
+ KeyValueStore largeMessageSafeStore =
+ new LargeMessageSafeStore(loggedStore, storeName, dropLargeMessage, maxMessageSize);
+ store = new SerializedKeyValueStore<>(largeMessageSafeStore, stringSerde, stringSerde,
+ serializedKeyValueStoreMetrics);
+ break;
+ }
+ case "cache-then-serde" : {
+ 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.DISALLOW_LARGE_MESSAGES is true.
+ 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() {
+ try {
+ store.close();
+ } catch (SamzaException e) {
+ loggedStore.close();
+ }
+ if (dir != null && dir.listFiles() != null) {
+ for (File file : dir.listFiles())
+ file.delete();
+ dir.delete();
+ }
+ }
+
+ @Test
+ public void testLargeMessagePut() {
+ 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 put() method invocation incorrectly completed.");
+ } catch (SamzaException e) {
+ Assert.assertNull("The large message was stored while it shouldn't have been.", loggedStore.get(stringSerde.toBytes(key)));
+ }
+ }
+ }
+
+ @Test
+ public void testLargeMessagePutAll() {
+ 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.", loggedStore.get(stringSerde.toBytes(key)));
+ }
+ }
+ }
+
+ @Test
+ public void testSmallMessagePut() {
+ String key = "test";
+ String smallMessage = StringUtils.repeat("a", maxMessageSize - 1);
+
+ store.put(key, smallMessage);
+ Assert.assertEquals(store.get(key), smallMessage);
+ }
+
+ @Test
+ public void testSmallMessagePutAll() {
+ 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);
+ }
+}
|