This is an automated email from the ASF dual-hosted git repository.
roman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 7b65644161767e90f0834db76ba116d4d92449d5
Author: Roman Khachatryan <khachatryan.ro...@gmail.com>
AuthorDate: Sun Jan 29 14:20:51 2023 +0000
[FLINK-30328][tests] Use RocksDBMemory factory interface in
TaskManagerWideRocksDbMemorySharingITCase
---
.../TaskManagerWideRocksDbMemorySharingITCase.java | 191 +++++++--------------
1 file changed, 63 insertions(+), 128 deletions(-)
diff --git
a/flink-tests/src/test/java/org/apache/flink/test/state/TaskManagerWideRocksDbMemorySharingITCase.java
b/flink-tests/src/test/java/org/apache/flink/test/state/TaskManagerWideRocksDbMemorySharingITCase.java
index e633bc005ad..bbc4da21655 100644
---
a/flink-tests/src/test/java/org/apache/flink/test/state/TaskManagerWideRocksDbMemorySharingITCase.java
+++
b/flink-tests/src/test/java/org/apache/flink/test/state/TaskManagerWideRocksDbMemorySharingITCase.java
@@ -21,76 +21,57 @@ import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
-import org.apache.flink.api.common.time.Deadline;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.MemorySize;
-import org.apache.flink.configuration.StateBackendOptions;
-import org.apache.flink.contrib.streaming.state.RocksDBNativeMetricOptions;
+import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
+import
org.apache.flink.contrib.streaming.state.RocksDBMemoryControllerUtils.RocksDBMemoryFactory;
import org.apache.flink.contrib.streaming.state.RocksDBOptions;
-import org.apache.flink.metrics.Gauge;
import org.apache.flink.runtime.jobgraph.JobGraph;
-import
org.apache.flink.runtime.metrics.DescriptiveStatisticsHistogramStatistics;
-import org.apache.flink.runtime.testutils.InMemoryReporter;
import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
import org.apache.flink.test.util.MiniClusterWithClientResource;
+import org.apache.flink.testutils.junit.SharedObjects;
+import org.apache.flink.testutils.junit.SharedReference;
+import org.apache.flink.util.TestLogger;
-import org.apache.commons.math3.stat.descriptive.moment.StandardDeviation;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
-import org.junit.Ignore;
+import org.junit.Rule;
import org.junit.Test;
+import org.rocksdb.Cache;
+import org.rocksdb.WriteBufferManager;
-import java.math.BigInteger;
-import java.time.Duration;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.List;
import java.util.Random;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
+import java.util.concurrent.CopyOnWriteArrayList;
import static
org.apache.flink.api.common.restartstrategy.RestartStrategies.noRestart;
-import static
org.apache.flink.contrib.streaming.state.RocksDBMemoryControllerUtils.calculateActualCacheCapacity;
import static
org.apache.flink.runtime.testutils.CommonTestUtils.waitForAllTaskRunning;
-import static org.apache.flink.util.Preconditions.checkState;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
/**
* Tests that {@link RocksDBOptions#FIX_PER_TM_MEMORY_SIZE} works as expected,
i.e. make RocksDB use
* the same BlockCache and WriteBufferManager objects. It does so using
RocksDB metrics.
*/
-public class TaskManagerWideRocksDbMemorySharingITCase {
+public class TaskManagerWideRocksDbMemorySharingITCase extends TestLogger {
private static final int PARALLELISM = 4;
private static final int NUMBER_OF_JOBS = 5;
private static final int NUMBER_OF_TASKS = NUMBER_OF_JOBS * PARALLELISM;
-
private static final MemorySize SHARED_MEMORY =
MemorySize.ofMebiBytes(NUMBER_OF_TASKS * 25);
- private static final double WRITE_BUFFER_RATIO = 0.5;
- private static final double EXPECTED_BLOCK_CACHE_SIZE =
- calculateActualCacheCapacity(SHARED_MEMORY.getBytes(),
WRITE_BUFFER_RATIO);
- // try to check that the memory usage is limited
- // however, there is no hard limit actually
- // because of https://issues.apache.org/jira/browse/FLINK-15532
- private static final double EFFECTIVE_LIMIT = EXPECTED_BLOCK_CACHE_SIZE *
1.5;
-
- private static final int NUM_MEASUREMENTS = 100;
-
- private InMemoryReporter metricsReporter;
private MiniClusterWithClientResource cluster;
+ @Rule public final SharedObjects sharedObjects = SharedObjects.create();
+
@Before
public void init() throws Exception {
- metricsReporter = InMemoryReporter.create();
cluster =
new MiniClusterWithClientResource(
new MiniClusterResourceConfiguration.Builder()
-
.setConfiguration(getConfiguration(metricsReporter))
+ .setConfiguration(getConfiguration())
.setNumberTaskManagers(1)
.setNumberSlotsPerTaskManager(NUMBER_OF_TASKS)
.build());
@@ -100,116 +81,48 @@ public class TaskManagerWideRocksDbMemorySharingITCase {
@After
public void destroy() {
cluster.after();
- metricsReporter.close();
}
- @Ignore
@Test
public void testBlockCache() throws Exception {
+ List<Cache> createdCaches = new CopyOnWriteArrayList<>();
+ List<WriteBufferManager> createdWriteBufferManagers = new
CopyOnWriteArrayList<>();
+ TestingRocksDBMemoryFactory memoryFactory =
+ new TestingRocksDBMemoryFactory(
+ sharedObjects.add(createdCaches),
+ sharedObjects.add(createdWriteBufferManagers));
List<JobID> jobIDs = new ArrayList<>(NUMBER_OF_JOBS);
try {
- // launch jobs
for (int i = 0; i < NUMBER_OF_JOBS; i++) {
-
jobIDs.add(cluster.getRestClusterClient().submitJob(dag()).get());
+
jobIDs.add(cluster.getRestClusterClient().submitJob(dag(memoryFactory)).get());
}
-
- // wait for init
- Deadline initDeadline = Deadline.fromNow(Duration.ofMinutes(1));
for (JobID jid : jobIDs) {
waitForAllTaskRunning(cluster.getMiniCluster(), jid, false);
- waitForAllMetricsReported(jid, initDeadline);
}
+ Assert.assertEquals(1, createdCaches.size());
+ Assert.assertEquals(1, createdWriteBufferManagers.size());
- // check declared capacity
- collectGaugeValues(jobIDs, "rocksdb.block-cache-capacity")
- .forEach(
- size ->
- assertEquals(
- "Unexpected rocksdb block cache
capacity",
- EXPECTED_BLOCK_CACHE_SIZE,
- size,
- 0));
-
- // do some work and check the actual usage of memory
- double[] deviations = new double[NUM_MEASUREMENTS];
- for (int i = 0; i < NUM_MEASUREMENTS; i++) {
- Thread.sleep(50L);
- double[] blockCacheUsages =
- collectGaugeValues(jobIDs, "rocksdb.block-cache-usage")
- .mapToDouble(value -> value)
- .toArray();
- assertTrue(
- String.format(
- "total block cache usage is too high: %s
(limit: %s, effective limit: %s)",
- Arrays.toString(blockCacheUsages),
- EXPECTED_BLOCK_CACHE_SIZE,
- EFFECTIVE_LIMIT),
- Arrays.stream(blockCacheUsages).max().getAsDouble() <=
EFFECTIVE_LIMIT);
- deviations[i] = new
StandardDeviation().evaluate(blockCacheUsages);
- }
- validateDeviations(deviations);
} finally {
for (JobID jobID : jobIDs) {
- cluster.getRestClusterClient().cancel(jobID).get();
+ try {
+ cluster.getRestClusterClient().cancel(jobID).get();
+ } catch (Exception e) {
+ log.warn("Can not cancel job {}", jobID, e);
+ }
}
}
}
- private static void validateDeviations(double[] deviations) {
- DescriptiveStatisticsHistogramStatistics percentile =
- new DescriptiveStatisticsHistogramStatistics(deviations);
- assertTrue(
- String.format(
- "Block cache usage reported by different tasks varies
too much: %s\n"
- + "That likely mean that they use different
cache objects",
- Arrays.toString(deviations)),
- // some deviation is possible because:
- // 1. records are being processed in parallel with requesting
metrics
- // 2. reporting metrics is not synchronized
- percentile.getQuantile(.50d) <= 10_000d
- && percentile.getQuantile(.75d) <= 500_000d);
- }
-
- private void waitForAllMetricsReported(JobID jid, Deadline deadline)
- throws InterruptedException {
- List<Double> gaugeValues = collectGaugeValues(jid,
"rocksdb.block-cache-capacity");
- while (deadline.hasTimeLeft() && isEmptyOrHasZeroes(gaugeValues)) {
- Thread.sleep(100);
- gaugeValues = collectGaugeValues(jid,
"rocksdb.block-cache-capacity");
- }
- if (isEmptyOrHasZeroes(gaugeValues)) {
- Assert.fail(
- String.format(
- "some tasks are still reporting zero cache
capacity: %s", gaugeValues));
- }
- }
-
- private boolean isEmptyOrHasZeroes(List<Double> gaugeValues) {
- return gaugeValues.isEmpty() || gaugeValues.stream().anyMatch(x -> x
== 0);
- }
-
- // collect at least one metric according to the given pattern
- // then convert it to double and return
- private List<Double> collectGaugeValues(JobID jobID, String metricPattern)
{
- //noinspection unchecked
- List<Double> list =
- metricsReporter.findJobMetricGroups(jobID,
metricPattern).stream()
- .map(triple -> ((Gauge<BigInteger>)
triple.f2).getValue().doubleValue())
- .collect(Collectors.toList());
- checkState(!list.isEmpty());
- return list;
- }
-
- private Stream<Double> collectGaugeValues(List<JobID> jobIDs, String
metricPattern) {
- return jobIDs.stream().flatMap(jobID -> collectGaugeValues(jobID,
metricPattern).stream());
- }
-
- private JobGraph dag() {
+ private JobGraph dag(RocksDBMemoryFactory memoryFactory) {
Configuration configuration = new Configuration();
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment(configuration);
env.setParallelism(PARALLELISM);
+ EmbeddedRocksDBStateBackend backend = new
EmbeddedRocksDBStateBackend(true);
+ backend.setRocksDBMemoryFactory(memoryFactory);
+ env.setStateBackend(backend);
+
// don't flush memtables by checkpoints
env.enableCheckpointing(24 * 60 * 60 * 1000,
CheckpointingMode.EXACTLY_ONCE);
env.setRestartStrategy(noRestart());
@@ -246,19 +159,41 @@ public class TaskManagerWideRocksDbMemorySharingITCase {
return env.getStreamGraph().getJobGraph();
}
- private static Configuration getConfiguration(InMemoryReporter
metricsReporter) {
+ private static Configuration getConfiguration() {
Configuration configuration = new Configuration();
-
configuration.set(RocksDBOptions.FIX_PER_TM_MEMORY_SIZE,
SHARED_MEMORY);
-
- configuration.set(StateBackendOptions.STATE_BACKEND, "rocksdb");
configuration.set(RocksDBOptions.USE_MANAGED_MEMORY, false);
- configuration.setDouble(RocksDBOptions.WRITE_BUFFER_RATIO,
WRITE_BUFFER_RATIO);
-
- metricsReporter.addToConfiguration(configuration);
- configuration.set(RocksDBNativeMetricOptions.BLOCK_CACHE_CAPACITY,
true);
- configuration.set(RocksDBNativeMetricOptions.BLOCK_CACHE_USAGE, true);
return configuration;
}
+
+ private static class TestingRocksDBMemoryFactory implements
RocksDBMemoryFactory {
+ private final SharedReference<List<Cache>> createdCaches;
+ private final SharedReference<List<WriteBufferManager>>
createdWriteBufferManagers;
+
+ private TestingRocksDBMemoryFactory(
+ SharedReference<List<Cache>> createdCaches,
+ SharedReference<List<WriteBufferManager>>
createdWriteBufferManagers) {
+ this.createdCaches = createdCaches;
+ this.createdWriteBufferManagers = createdWriteBufferManagers;
+ }
+
+ @Override
+ public Cache createCache(long cacheCapacity, double
highPriorityPoolRatio) {
+ Cache cache =
+ RocksDBMemoryFactory.DEFAULT.createCache(cacheCapacity,
highPriorityPoolRatio);
+ createdCaches.get().add(cache);
+ return cache;
+ }
+
+ @Override
+ public WriteBufferManager createWriteBufferManager(
+ long writeBufferManagerCapacity, Cache cache) {
+ WriteBufferManager writeBufferManager =
+ RocksDBMemoryFactory.DEFAULT.createWriteBufferManager(
+ writeBufferManagerCapacity, cache);
+ createdWriteBufferManagers.get().add(writeBufferManager);
+ return writeBufferManager;
+ }
+ }
}
