You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2020/01/23 14:53:32 UTC
[flink] 02/02: [FLINK-15368][e2e] (follow-up) Minor improvements to
the RocksDB Memory Control e2e test
This is an automated email from the ASF dual-hosted git repository.
sewen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 7a8ac1cc0305bd312e3a80e2c07f28d573aeddc1
Author: Stephan Ewen <se...@apache.org>
AuthorDate: Thu Jan 23 12:23:07 2020 +0100
[FLINK-15368][e2e] (follow-up) Minor improvements to the RocksDB Memory Control e2e test
Some cleanups in RocksDBStateMemoryControlTestProgram
- skip ".returns" when not necessary
- peel off command line parameter abstraction earlier (don't pass ParameterTool to functions).
- Add proper class-level JavaDoc.
Remove script from 'run-nightly-tests.sh' (unused) and add to a different CRON split file instead.
Some improvements to the test script
- Added comment explainig where the upper limit value comes from
- Added fixed managed memory size, so the test is not affected by changes in the managed memory fraction settings
- Always log measured cache size for information
---
.../RocksDBStateMemoryControlTestProgram.java | 38 +++++++++++++---------
flink-end-to-end-tests/run-nightly-tests.sh | 2 --
.../test_rocksdb_state_memory_control.sh | 26 +++++++++++----
tools/travis/splits/split_checkpoints.sh | 2 ++
4 files changed, 43 insertions(+), 25 deletions(-)
diff --git a/flink-end-to-end-tests/flink-rocksdb-state-memory-control-test/src/main/java/org/apache/flink/streaming/tests/RocksDBStateMemoryControlTestProgram.java b/flink-end-to-end-tests/flink-rocksdb-state-memory-control-test/src/main/java/org/apache/flink/streaming/tests/RocksDBStateMemoryControlTestProgram.java
index 77f2324..8f0af03 100644
--- a/flink-end-to-end-tests/flink-rocksdb-state-memory-control-test/src/main/java/org/apache/flink/streaming/tests/RocksDBStateMemoryControlTestProgram.java
+++ b/flink-end-to-end-tests/flink-rocksdb-state-memory-control-test/src/main/java/org/apache/flink/streaming/tests/RocksDBStateMemoryControlTestProgram.java
@@ -42,12 +42,17 @@ import static org.apache.flink.streaming.tests.TestOperatorEnum.EVENT_SOURCE;
import static org.apache.flink.streaming.tests.TestOperatorEnum.TIME_WINDOW_OPER;
/**
- *
+ * The test program for a job that simply accumulates data in various states.
+ * This is used to stress the RocksDB memory and check that the cache/write buffer management work
+ * properly, limiting the overall memory footprint of RocksDB.
*/
public class RocksDBStateMemoryControlTestProgram {
public static void main(String[] args) throws Exception {
final ParameterTool pt = ParameterTool.fromArgs(args);
+ final boolean useValueState = pt.getBoolean("useValueState", false);
+ final boolean useListState = pt.getBoolean("useListState", false);
+ final boolean useMapState = pt.getBoolean("useMapState", false);
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
@@ -58,9 +63,10 @@ public class RocksDBStateMemoryControlTestProgram {
.uid(EVENT_SOURCE.getUid())
.assignTimestampsAndWatermarks(createTimestampExtractor(pt))
.keyBy(Event::getKey);
- keyedStream.map(new ValueStateMapper(pt)).returns(Event.class).name("ValueStateMapper").uid("ValueStateMapper");
- keyedStream.map(new ListStateMapper(pt)).returns(Event.class).name("ListStateMapper").uid("ListStateMapper");
- keyedStream.map(new MapStateMapper(pt)).returns(Event.class).name("MapStateMapper").uid("MapStateMapper");
+
+ keyedStream.map(new ValueStateMapper(useValueState)).name("ValueStateMapper").uid("ValueStateMapper");
+ keyedStream.map(new ListStateMapper(useListState)).name("ListStateMapper").uid("ListStateMapper");
+ keyedStream.map(new MapStateMapper(useMapState)).name("MapStateMapper").uid("MapStateMapper");
boolean useWindow = pt.getBoolean("useWindow", false);
if (useWindow) {
@@ -84,12 +90,12 @@ public class RocksDBStateMemoryControlTestProgram {
private static final long serialVersionUID = 1L;
- private ValueState<String> valueState;
+ private transient ValueState<String> valueState;
- private boolean useValueState = false;
+ private final boolean useValueState;
- ValueStateMapper(ParameterTool parameterTool) {
- this.useValueState = parameterTool.getBoolean("useValueState", false);
+ ValueStateMapper(boolean useValueState) {
+ this.useValueState = useValueState;
}
@Override
@@ -119,12 +125,12 @@ public class RocksDBStateMemoryControlTestProgram {
private static final long serialVersionUID = 1L;
- private ListState<String> listState;
+ private transient ListState<String> listState;
- private boolean useListState = false;
+ private final boolean useListState;
- ListStateMapper(ParameterTool parameterTool) {
- this.useListState = parameterTool.getBoolean("useListState", false);
+ ListStateMapper(boolean useListState) {
+ this.useListState = useListState;
}
@Override
@@ -149,12 +155,12 @@ public class RocksDBStateMemoryControlTestProgram {
private static final long serialVersionUID = 1L;
- private MapState<Long, String> mapState;
+ private transient MapState<Long, String> mapState;
- private boolean useMapState = false;
+ private final boolean useMapState;
- MapStateMapper(ParameterTool parameterTool) {
- this.useMapState = parameterTool.getBoolean("useMapState", false);
+ MapStateMapper(boolean useMapState) {
+ this.useMapState = useMapState;
}
@Override
diff --git a/flink-end-to-end-tests/run-nightly-tests.sh b/flink-end-to-end-tests/run-nightly-tests.sh
index 2518b73..408d80b 100755
--- a/flink-end-to-end-tests/run-nightly-tests.sh
+++ b/flink-end-to-end-tests/run-nightly-tests.sh
@@ -171,8 +171,6 @@ run_test "ConnectedComponents iterations with high parallelism end-to-end test"
run_test "Dependency shading of table modules test" "$END_TO_END_DIR/test-scripts/test_table_shaded_dependencies.sh"
-run_test "RocksDB memory control end-to-end test" "$END_TO_END_DIR/test-scripts/test_rocksdb_state_memory_control.sh" "skip_check_exceptions"
-
################################################################################
# Sticky Scheduling
################################################################################
diff --git a/flink-end-to-end-tests/test-scripts/test_rocksdb_state_memory_control.sh b/flink-end-to-end-tests/test-scripts/test_rocksdb_state_memory_control.sh
index 49c610d..0674f72 100755
--- a/flink-end-to-end-tests/test-scripts/test_rocksdb_state_memory_control.sh
+++ b/flink-end-to-end-tests/test-scripts/test_rocksdb_state_memory_control.sh
@@ -28,23 +28,33 @@ PARALLELISM=2
CHECKPOINT_DIR="$TEST_DATA_DIR/test_rocksdb_state_memory_control-dir"
mkdir -p $CHECKPOINT_DIR
CHECKPOINT_DIR_URI="file://$CHECKPOINT_DIR"
-# 161061276 + 4 * 8 * 1024 * 1024 = 194615708 bytes
-EXPECTED_MAX_MEMORY_USAGE=194615708
+# The managed memory is fixed at 300m which gives an allowed cache size of about 157,000,000b,
+# via the compensation logic for RocksDBs memory exceeding the cache.
+#
+# Due to RocksDB's lenient (non-strict) memory accounting, we add num-states * arena-size MBs in extra tolerance
+# (because in corner cases, especially with slow I/O, cache footprint can temporarily rise)
+# which brings this to about 190,000,000 bytes, rounded up to 200,000,000 for a safety/stability margin.
+#
+# With unrestricted memory use, this test would use more than 400m of RocksDB memory, so we are
+# well below this limit, thus testing that the memory limiting is actually active.
+EXPECTED_MAX_MEMORY_USAGE=200000000
+
+set_config_key "taskmanager.numberOfTaskSlots" "$PARALLELISM"
set_config_key "taskmanager.memory.process.size" "1024m"
+set_config_key "taskmanager.memory.managed.size" "300m"
set_config_key "state.backend.rocksdb.memory.managed" "true"
+set_config_key "state.backend.rocksdb.memory.write-buffer-ratio" "0.8"
set_config_key "state.backend.rocksdb.metrics.size-all-mem-tables" "true"
set_config_key "state.backend.rocksdb.metrics.cur-size-active-mem-table" "true"
set_config_key "state.backend.rocksdb.metrics.num-immutable-mem-table" "true"
-set_config_key "state.backend.rocksdb.memory.`write`-buffer-ratio" "0.8"
set_config_key "state.backend.rocksdb.metrics.block-cache-usage" "true"
set_config_key "state.backend.rocksdb.metrics.estimate-table-readers-mem" "true"
-set_config_key "taskmanager.numberOfTaskSlots" "$PARALLELISM"
set_config_key "metrics.fetcher.update-interval" "1000"
setup_flink_slf4j_metric_reporter
start_cluster
-echo "Running rocksdb state backend memory control test"
+echo "Running RocksDB state backend memory control test"
TEST_PROGRAM_JAR=${END_TO_END_DIR}/flink-rocksdb-state-memory-control-test/target/RocksDBStateMemoryControlTestProgram.jar
@@ -80,11 +90,13 @@ function find_max_block_cache_usage() {
function memory_under_limit() {
local MAX_BLOCK_CACHE_USAGE=$1
local EXPECTED_MAX_MEMORY_USAGE=$2
+
+ echo "[INFO] Current block cache usage for RocksDB instance in slot was $MAX_BLOCK_CACHE_USAGE"
+
if [ "$MAX_BLOCK_CACHE_USAGE" -gt "$EXPECTED_MAX_MEMORY_USAGE" ]; then
- echo "[ERROR] Current block cache uaage $MAX_BLOCK_CACHE_USAGE larger than expected memory limit $EXPECTED_MAX_MEMORY_USAGE"
+ echo "[ERROR] Current block cache usage $MAX_BLOCK_CACHE_USAGE larger than expected memory limit $EXPECTED_MAX_MEMORY_USAGE"
exit 1
fi
-
}
JOB_CMD=`buildBaseJobCmd `
diff --git a/tools/travis/splits/split_checkpoints.sh b/tools/travis/splits/split_checkpoints.sh
index bf29aac..2809354 100755
--- a/tools/travis/splits/split_checkpoints.sh
+++ b/tools/travis/splits/split_checkpoints.sh
@@ -74,5 +74,7 @@ run_test "Resuming Externalized Checkpoint after terminal failure (file, sync) e
run_test "Resuming Externalized Checkpoint after terminal failure (rocks, non-incremental) end-to-end test" "$END_TO_END_DIR/test-scripts/test_resume_externalized_checkpoints.sh 2 2 rocks true false true" "skip_check_exceptions"
run_test "Resuming Externalized Checkpoint after terminal failure (rocks, incremental) end-to-end test" "$END_TO_END_DIR/test-scripts/test_resume_externalized_checkpoints.sh 2 2 rocks true true true" "skip_check_exceptions"
+run_test "RocksDB Memory Management end-to-end test" "$END_TO_END_DIR/test-scripts/test_rocksdb_state_memory_control.sh"
+
printf "\n[PASS] All tests passed\n"
exit 0