[hotfix] Disable WAL in RocksDB state "clear"

flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/AbstractRocksDBState.java

@@ -28,6 +28,7 @@
 import org.rocksdb.ColumnFamilyHandle;
 import org.rocksdb.RocksDBException;
 
+import org.rocksdb.WriteOptions;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -62,6 +63,11 @@ public abstract class AbstractRocksDBState<K, N, S extends State, SD extends Sta
 	/** The column family of this particular instance of state */
 	protected ColumnFamilyHandle columnFamily;
 
+	/**
+	 * We disable writes to the write-ahead-log here.
+	 */
+	private final WriteOptions writeOptions;
+
 	/**
 	 * Creates a new RocksDB backed state.
 	 *
@@ -75,6 +81,9 @@ protected AbstractRocksDBState(ColumnFamilyHandle columnFamily,
 		this.backend = backend;
 
 		this.columnFamily = columnFamily;
+
+		writeOptions = new WriteOptions();
+		writeOptions.setDisableWAL(true);
 	}
 
 	// ------------------------------------------------------------------------
@@ -86,7 +95,7 @@ public void clear() {
 		try {
 			writeKeyAndNamespace(out);
 			byte[] key = baos.toByteArray();
-			backend.db.remove(columnFamily, key);
+			backend.db.remove(columnFamily, writeOptions, key);
 		} catch (IOException|RocksDBException e) {
 			throw new RuntimeException("Error while removing entry from RocksDB", e);
 		}