SAMZA-2174: Throw a record too large exception for oversized records in changelog

docs/learn/documentation/versioned/jobs/configuration-table.html

@@ -1687,6 +1687,49 @@ <h1>Samza Configuration Reference</h1>
                     </td>
                 </tr>
 
+                <tr>
+                    <td class="property" id="stores-changelog-max-message-size-bytes">stores.<span class="store">store-name</span>.changelog.max.message.size.bytes</td>
+                    <td class="default">1048576</td>
+                    <td class="description">
+                        This property sets the maximum size of the messages allowed in the changelog.
+                        The default value is 1 MB.
+                    </td>
+                </tr>
+
+                <tr>
+                    <td class="property" id="stores-disallow-large-messages">stores.<span class="store">store-name</span>.disallow.large.messages</td>
+                    <td class="default">false</td>
+                    <td class="description">
+                        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
+                        <a href="#stores-changelog-max-message-size-bytes" class="property">stores.*.changelog.max.message.size.bytes</a>
+                        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 <a href="#stores-drop-large-messages" class="property">stores.*.drop.large.messages</a>
+                        configuration is ignored. The default value for this config is false. When this property is not set,
+                        <a href="#stores-drop-large-messages" class="property">stores.*.drop.large.messages</a>
+                        determines the large message handling behavior.
+                    </td>
+                </tr>
+
+                <tr>
+                    <td class="property" id="stores-drop-large-messages">stores.<span class="store">store-name</span>.drop.large.messages</td>
+                    <td class="default">false</td>
+                    <td class="description">
+                        This property, when turned on, causes messages larger than
+                        <a href="#stores-changelog-max-message-size-bytes" class="property">stores.*.changelog.max.message.size.bytes</a>
+                        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 <a href="#stores-rocksdb-object-cache-size" class="property">stores.*.object.cache.size</a>
+                        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.
+                    </td>
+                </tr>
+
                 <tr>
                     <th colspan="3" class="section" id="keyvalue-rocksdb">
                         Using RocksDB for key-value storage<br>

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.|
 
 ##### <a name="advanced-storage-configurations"></a>[4.1 Advanced Storage Configurations](#advanced-storage-configurations)

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.
    */

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));
+  }
 }

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<byte[], byte[]> {
+
+  private static final Logger LOG = LoggerFactory.getLogger(LargeMessageSafeStore.class);
+  private final KeyValueStore<byte[], byte[]> store;
+  private final String storeName;
+  private final boolean dropLargeMessages;
+  private final int maxMessageSize;
+  private final LargeMessageSafeStoreMetrics largeMessageSafeStoreMetrics;
+
+  public LargeMessageSafeStore(KeyValueStore<byte[], byte[]> 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<Entry<byte[], byte[]>> entries) {
+    entries.forEach(entry -> {
+        validateMessageSize(entry.getValue());
+      });
+    List<Entry<byte[], byte[]>> largeMessageSafeEntries = removeLargeMessages(entries);
+    store.putAll(largeMessageSafeEntries);
+  }
+
+  @Override
+  public void delete(byte[] key) {
+    store.delete(key);
+  }
+
+  @Override
+  public void deleteAll(List<byte[]> keys) {
+    store.deleteAll(keys);
+  }
+
+  @Override
+  public KeyValueIterator<byte[], byte[]> range(byte[] from, byte[] to) {
+    return store.range(from, to);
+  }
+
+  @Override
+  public KeyValueSnapshot<byte[], byte[]> snapshot(byte[] from, byte[] to) {
+    return store.snapshot(from, to);
+  }
+
+  @Override
+  public KeyValueIterator<byte[], byte[]> 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<Entry<byte[], byte[]>> removeLargeMessages(List<Entry<byte[], byte[]>> entries) {
+    List<Entry<byte[], byte[]>> 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;
+  }
+}