You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ha...@apache.org on 2024/04/26 01:48:50 UTC
(flink) branch master updated (259aef0b66b -> 19577b0da7f)
This is an automated email from the ASF dual-hosted git repository.
hangxiang pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from 259aef0b66b [FLINK-31223][sqlgateway] Pass all flink parameters to the SqlGatewayRestEndpoint.
new 629c35712dc [FLINK-35047][state] Introduce forstjni dependency
new 92d7f653860 [FLINK-35047][state] Introduce ForSt related options
new 8411367ab0a [FLINK-35047][state] Introduce Metrics for ForStStateBackend
new 5f5d5022c03 [FLINK-35047][state] Introduce ResourceContainer for ForStStateBackend
new 625d8a5853f [FLINK-35047][state] Support restoring and building ForSt
new 417484e6029 [FLINK-35047][state] Introduce ForStStateBackend and ForStKeyedStateBackend
new 19577b0da7f [FLINK-35047][state] Support FlinkEnv to init ForSt
The 7 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../flink-statebackend-forst/pom.xml | 11 +-
.../forst/ConfigurableForStOptionsFactory.java | 39 +
.../state/forst/ForStConfigurableOptions.java | 377 ++++++++++
.../flink/state/forst/ForStKeyedStateBackend.java | 135 ++++
.../state/forst/ForStKeyedStateBackendBuilder.java | 160 +++++
.../state/forst/ForStMemoryConfiguration.java | 248 +++++++
.../state/forst/ForStMemoryControllerUtils.java | 195 +++++
.../state/forst/ForStNativeMetricMonitor.java | 233 ++++++
.../state/forst/ForStNativeMetricOptions.java | 700 ++++++++++++++++++
.../flink/state/forst/ForStOperationUtils.java | 138 ++++
.../org/apache/flink/state/forst/ForStOptions.java | 133 ++++
.../flink/state/forst/ForStOptionsFactory.java | 118 +++
.../apache/flink/state/forst/ForStProperty.java | 115 +++
.../flink/state/forst/ForStResourceContainer.java | 547 ++++++++++++++
.../flink/state/forst/ForStSharedResources.java | 73 ++
.../state/forst/ForStSharedResourcesFactory.java | 185 +++++
.../flink/state/forst/ForStStateBackend.java | 680 ++++++++++++++++++
.../state/forst/ForStStateBackendFactory.java | 34 +
.../flink/state/forst/restore/ForStHandle.java | 116 +++
.../forst/restore/ForStNoneRestoreOperation.java | 62 ++
.../state/forst/restore/ForStRestoreOperation.java | 29 +
.../state/forst/restore/ForStRestoreResult.java | 55 ++
.../apache/flink/state/forst/ForStExtension.java | 212 ++++++
.../apache/flink/state/forst/ForStInitITCase.java | 64 ++
.../forst/ForStMemoryControllerUtilsTest.java | 129 ++++
.../state/forst/ForStMultiClassLoaderTest.java | 88 +++
.../state/forst/ForStNativeMetricMonitorTest.java | 254 +++++++
.../state/forst/ForStNativeMetricOptionsTest.java | 53 ++
.../state/forst/ForStOperationsUtilsTest.java | 89 +++
.../flink/state/forst/ForStPropertyTest.java | 47 ++
.../state/forst/ForStResourceContainerTest.java | 320 +++++++++
.../state/forst/ForStStateBackendConfigTest.java | 793 +++++++++++++++++++++
.../state/forst/ForStStateBackendFactoryTest.java | 132 ++++
.../apache/flink/state/forst/ForStTestUtils.java | 54 ++
34 files changed, 6612 insertions(+), 6 deletions(-)
create mode 100644 flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ConfigurableForStOptionsFactory.java
create mode 100644 flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStConfigurableOptions.java
create mode 100644 flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStKeyedStateBackend.java
create mode 100644 flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStKeyedStateBackendBuilder.java
create mode 100644 flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStMemoryConfiguration.java
create mode 100644 flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStMemoryControllerUtils.java
create mode 100644 flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStNativeMetricMonitor.java
create mode 100644 flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStNativeMetricOptions.java
create mode 100644 flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStOperationUtils.java
create mode 100644 flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStOptions.java
create mode 100644 flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStOptionsFactory.java
create mode 100644 flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStProperty.java
create mode 100644 flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java
create mode 100644 flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStSharedResources.java
create mode 100644 flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStSharedResourcesFactory.java
create mode 100644 flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateBackend.java
create mode 100644 flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateBackendFactory.java
create mode 100644 flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/restore/ForStHandle.java
create mode 100644 flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/restore/ForStNoneRestoreOperation.java
create mode 100644 flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/restore/ForStRestoreOperation.java
create mode 100644 flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/restore/ForStRestoreResult.java
create mode 100644 flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStExtension.java
create mode 100644 flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStInitITCase.java
create mode 100644 flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStMemoryControllerUtilsTest.java
create mode 100644 flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStMultiClassLoaderTest.java
create mode 100644 flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStNativeMetricMonitorTest.java
create mode 100644 flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStNativeMetricOptionsTest.java
create mode 100644 flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStOperationsUtilsTest.java
create mode 100644 flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStPropertyTest.java
create mode 100644 flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStResourceContainerTest.java
create mode 100644 flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateBackendConfigTest.java
create mode 100644 flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateBackendFactoryTest.java
create mode 100644 flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStTestUtils.java
(flink) 07/07: [FLINK-35047][state] Support FlinkEnv to init ForSt
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hangxiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 19577b0da7f42fe569084f0207bce93a9be08f57
Author: Hangxiang Yu <ma...@gmail.com>
AuthorDate: Fri Apr 19 01:33:21 2024 +0800
[FLINK-35047][state] Support FlinkEnv to init ForSt
---
.../flink/state/forst/ForStKeyedStateBackend.java | 47 +++++------
.../state/forst/ForStKeyedStateBackendBuilder.java | 51 +++---------
.../org/apache/flink/state/forst/ForStOptions.java | 14 +++-
.../flink/state/forst/ForStResourceContainer.java | 91 ++++++++++++++++++++--
.../flink/state/forst/ForStStateBackend.java | 53 +++++++++----
.../state/forst/ForStResourceContainerTest.java | 23 ++++++
.../state/forst/ForStStateBackendConfigTest.java | 68 +++++++++++-----
7 files changed, 242 insertions(+), 105 deletions(-)
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStKeyedStateBackend.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStKeyedStateBackend.java
index 8c7dfd0ba0f..5737ba7ceac 100644
--- a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStKeyedStateBackend.java
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStKeyedStateBackend.java
@@ -19,7 +19,7 @@ package org.apache.flink.state.forst;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.util.FileUtils;
+import org.apache.flink.util.Disposable;
import org.apache.flink.util.IOUtils;
import org.apache.flink.util.Preconditions;
@@ -29,14 +29,14 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
-import java.io.IOException;
+import java.net.URI;
/**
* A KeyedStateBackend that stores its state in {@code ForSt}. This state backend can store very
* large state that exceeds memory even disk to remote storage. TODO: Support to implement the new
* interface of KeyedStateBackend
*/
-public class ForStKeyedStateBackend<K> {
+public class ForStKeyedStateBackend<K> implements Disposable {
private static final Logger LOG = LoggerFactory.getLogger(ForStKeyedStateBackend.class);
@@ -46,9 +46,6 @@ public class ForStKeyedStateBackend<K> {
/** The container of ForSt options. */
private final ForStResourceContainer optionsContainer;
- /** Path where this configured instance stores its local data directory. */
- private final File localBasePath;
-
/**
* We are not using the default column family for Flink state ops, but we still need to remember
* this handle so that we can close it properly when the backend is closed. Note that the one
@@ -70,14 +67,11 @@ public class ForStKeyedStateBackend<K> {
private boolean disposed = false;
public ForStKeyedStateBackend(
- File localBasePath,
ForStResourceContainer optionsContainer,
TypeSerializer<K> keySerializer,
RocksDB db,
ColumnFamilyHandle defaultColumnFamilyHandle,
ForStNativeMetricMonitor nativeMetricMonitor) {
-
- this.localBasePath = Preconditions.checkNotNull(localBasePath);
this.optionsContainer = Preconditions.checkNotNull(optionsContainer);
this.keySerializer = keySerializer;
this.db = db;
@@ -86,6 +80,7 @@ public class ForStKeyedStateBackend<K> {
}
/** Should only be called by one thread, and only after all accesses to the DB happened. */
+ @Override
public void dispose() {
if (this.disposed) {
return;
@@ -108,27 +103,33 @@ public class ForStKeyedStateBackend<K> {
// ... and finally close the DB instance ...
IOUtils.closeQuietly(db);
- IOUtils.closeQuietly(optionsContainer);
+ LOG.info(
+ "Closed ForSt State Backend. Cleaning up ForSt local working directory {}, remote working directory {}.",
+ optionsContainer.getLocalBasePath(),
+ optionsContainer.getRemoteBasePath());
+
+ try {
+ optionsContainer.clearDirectories();
+ } catch (Exception ex) {
+ LOG.warn(
+ "Could not delete ForSt local working directory {}, remote working directory {}.",
+ optionsContainer.getLocalBasePath(),
+ optionsContainer.getRemoteBasePath(),
+ ex);
+ }
- cleanLocalBasePath();
+ IOUtils.closeQuietly(optionsContainer);
}
this.disposed = true;
}
- private void cleanLocalBasePath() {
- LOG.info(
- "Closed ForSt State Backend. Cleaning up ForSt local working directory {}.",
- localBasePath);
-
- try {
- FileUtils.deleteDirectory(localBasePath);
- } catch (IOException ex) {
- LOG.warn("Could not delete ForSt local working directory: {}", localBasePath, ex);
- }
+ @VisibleForTesting
+ File getLocalBasePath() {
+ return optionsContainer.getLocalBasePath();
}
@VisibleForTesting
- File getInstanceBasePath() {
- return localBasePath;
+ URI getRemoteBasePath() {
+ return optionsContainer.getRemoteBasePath();
}
}
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStKeyedStateBackendBuilder.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStKeyedStateBackendBuilder.java
index 1ff68f413a4..8cf45ff7c6a 100644
--- a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStKeyedStateBackendBuilder.java
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStKeyedStateBackendBuilder.java
@@ -28,7 +28,6 @@ import org.apache.flink.state.forst.restore.ForStNoneRestoreOperation;
import org.apache.flink.state.forst.restore.ForStRestoreOperation;
import org.apache.flink.state.forst.restore.ForStRestoreResult;
import org.apache.flink.util.CollectionUtil;
-import org.apache.flink.util.FileUtils;
import org.apache.flink.util.IOUtils;
import org.apache.flink.util.Preconditions;
@@ -41,8 +40,6 @@ import org.slf4j.LoggerFactory;
import javax.annotation.Nonnull;
-import java.io.File;
-import java.io.IOException;
import java.util.Collection;
import java.util.function.Function;
@@ -66,12 +63,6 @@ public class ForStKeyedStateBackendBuilder<K>
/** The container of ForSt option factory and predefined options. */
private final ForStResourceContainer optionsContainer;
- /** Path where this configured instance stores its data directory. */
- private final File localBasePath;
-
- /** Path where this configured instance stores its ForSt database. */
- private final File localForStPath;
-
private final MetricGroup metricGroup;
/** ForSt property-based and statistics-based native metrics options. */
@@ -84,8 +75,6 @@ public class ForStKeyedStateBackendBuilder<K>
MetricGroup metricGroup,
@Nonnull Collection<KeyedStateHandle> stateHandles) {
this.optionsContainer = optionsContainer;
- this.localBasePath = optionsContainer.getLocalBasePath();
- this.localForStPath = optionsContainer.getLocalForStPath();
this.columnFamilyOptionsFactory = Preconditions.checkNotNull(columnFamilyOptionsFactory);
this.keySerializerProvider =
StateSerializerProvider.fromNewRegisteredSerializer(keySerializer);
@@ -108,7 +97,7 @@ public class ForStKeyedStateBackendBuilder<K>
ForStRestoreOperation restoreOperation = null;
try {
- prepareLocalDirectories();
+ optionsContainer.prepareDirectories();
restoreOperation = getForStRestoreOperation();
ForStRestoreResult restoreResult = restoreOperation.restore();
db = restoreResult.getDb();
@@ -124,12 +113,16 @@ public class ForStKeyedStateBackendBuilder<K>
IOUtils.closeQuietly(db);
// it's possible that db has been initialized but later restore steps failed
IOUtils.closeQuietly(restoreOperation);
- IOUtils.closeQuietly(optionsContainer);
try {
- FileUtils.deleteDirectory(localBasePath);
+ optionsContainer.clearDirectories();
} catch (Exception ex) {
- logger.warn("Failed to delete local base path for ForSt: " + localBasePath, ex);
+ logger.warn(
+ "Failed to delete ForSt local base path {}, remote base path {}.",
+ optionsContainer.getLocalBasePath(),
+ optionsContainer.getRemoteBasePath(),
+ ex);
}
+ IOUtils.closeQuietly(optionsContainer);
// Log and rethrow
if (e instanceof BackendBuildingException) {
throw (BackendBuildingException) e;
@@ -140,10 +133,10 @@ public class ForStKeyedStateBackendBuilder<K>
}
}
logger.info(
- "Finished building ForSt keyed state-backend at local base path: {}.",
- localBasePath);
+ "Finished building ForSt keyed state-backend at local base path: {}, remote base path: {}.",
+ optionsContainer.getLocalBasePath(),
+ optionsContainer.getRemoteBasePath());
return new ForStKeyedStateBackend<>(
- this.localBasePath,
this.optionsContainer,
this.keySerializerProvider.currentSchemaSerializer(),
db,
@@ -155,7 +148,7 @@ public class ForStKeyedStateBackendBuilder<K>
DBOptions dbOptions = optionsContainer.getDbOptions();
if (CollectionUtil.isEmptyOrAllElementsNull(restoreStateHandles)) {
return new ForStNoneRestoreOperation(
- localForStPath,
+ optionsContainer.getLocalForStPath(),
dbOptions,
columnFamilyOptionsFactory,
nativeMetricOptions,
@@ -164,24 +157,4 @@ public class ForStKeyedStateBackendBuilder<K>
// TODO: Support Restoring
throw new UnsupportedOperationException("Not support restoring yet for ForStStateBackend");
}
-
- private void prepareLocalDirectories() throws IOException {
- checkAndCreateDirectory(localBasePath);
- if (localForStPath.exists()) {
- // Clear the base directory when the backend is created
- // in case something crashed and the backend never reached dispose()
- FileUtils.deleteDirectory(localBasePath);
- }
- }
-
- private static void checkAndCreateDirectory(File directory) throws IOException {
- if (directory.exists()) {
- if (!directory.isDirectory()) {
- throw new IOException("Not a directory: " + directory);
- }
- } else if (!directory.mkdirs()) {
- throw new IOException(
- String.format("Could not create ForSt data directory at %s.", directory));
- }
- }
}
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStOptions.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStOptions.java
index ba8ce6b2c62..c0bd1e123a6 100644
--- a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStOptions.java
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStOptions.java
@@ -30,7 +30,7 @@ import org.apache.flink.configuration.description.TextElement;
@Experimental
public class ForStOptions {
- /** The local directory (on the TaskManager) where ForSt puts its some files. */
+ /** The local directory (on the TaskManager) where ForSt puts some meta files. */
public static final ConfigOption<String> LOCAL_DIRECTORIES =
ConfigOptions.key("state.backend.forst.localdir")
.stringType()
@@ -38,13 +38,23 @@ public class ForStOptions {
.withDescription(
Description.builder()
.text(
- "The local directory (on the TaskManager) where ForSt puts its some files. Per default, it will be <WORKING_DIR>/tmp. See %s for more details.",
+ "The local directory (on the TaskManager) where ForSt puts some meta files. Per default, it will be <WORKING_DIR>/tmp. See %s for more details.",
TextElement.code(
ClusterOptions
.TASK_MANAGER_PROCESS_WORKING_DIR_BASE
.key()))
.build());
+ /** The remote directory where ForSt puts its SST files. */
+ public static final ConfigOption<String> REMOTE_DIRECTORY =
+ ConfigOptions.key("state.backend.forst.remotedir")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ String.format(
+ "The remote directory where ForSt puts its SST files, fallback to %s if not configured.",
+ LOCAL_DIRECTORIES.key()));
+
/** The options factory class for ForSt to create DBOptions and ColumnFamilyOptions. */
public static final ConfigOption<String> OPTIONS_FACTORY =
ConfigOptions.key("state.backend.forst.options-factory")
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java
index 2f54522d1c7..2b912ca9732 100644
--- a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java
@@ -22,6 +22,7 @@ import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.runtime.memory.OpaqueMemoryResource;
import org.apache.flink.util.FileUtils;
import org.apache.flink.util.IOUtils;
@@ -33,6 +34,7 @@ import org.rocksdb.Cache;
import org.rocksdb.ColumnFamilyOptions;
import org.rocksdb.DBOptions;
import org.rocksdb.Filter;
+import org.rocksdb.FlinkEnv;
import org.rocksdb.IndexType;
import org.rocksdb.PlainTableConfig;
import org.rocksdb.ReadOptions;
@@ -46,6 +48,7 @@ import javax.annotation.Nullable;
import java.io.File;
import java.io.IOException;
+import java.net.URI;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
@@ -66,7 +69,11 @@ public final class ForStResourceContainer implements AutoCloseable {
// the filename length limit is 255 on most operating systems
private static final int INSTANCE_PATH_LENGTH_LIMIT = 255 - FORST_RELOCATE_LOG_SUFFIX.length();
- private static final String LOCAL_DB_DIR_STRING = "db";
+ private static final String DB_DIR_STRING = "db";
+
+ @Nullable private final URI remoteBasePath;
+
+ @Nullable private final URI remoteForStPath;
@Nullable private final File localBasePath;
@@ -94,19 +101,19 @@ public final class ForStResourceContainer implements AutoCloseable {
@VisibleForTesting
public ForStResourceContainer() {
- this(new Configuration(), null, null, null, false);
+ this(new Configuration(), null, null, null, null, false);
}
@VisibleForTesting
public ForStResourceContainer(@Nullable ForStOptionsFactory optionsFactory) {
- this(new Configuration(), optionsFactory, null, null, false);
+ this(new Configuration(), optionsFactory, null, null, null, false);
}
@VisibleForTesting
public ForStResourceContainer(
@Nullable ForStOptionsFactory optionsFactory,
@Nullable OpaqueMemoryResource<ForStSharedResources> sharedResources) {
- this(new Configuration(), optionsFactory, sharedResources, null, false);
+ this(new Configuration(), optionsFactory, sharedResources, null, null, false);
}
public ForStResourceContainer(
@@ -114,6 +121,7 @@ public final class ForStResourceContainer implements AutoCloseable {
@Nullable ForStOptionsFactory optionsFactory,
@Nullable OpaqueMemoryResource<ForStSharedResources> sharedResources,
@Nullable File localBasePath,
+ @Nullable URI remoteBasePath,
boolean enableStatistics) {
this.configuration = configuration;
@@ -121,8 +129,10 @@ public final class ForStResourceContainer implements AutoCloseable {
this.sharedResources = sharedResources;
this.localBasePath = localBasePath;
- this.localForStPath =
- localBasePath != null ? new File(localBasePath, LOCAL_DB_DIR_STRING) : null;
+ this.localForStPath = localBasePath != null ? new File(localBasePath, DB_DIR_STRING) : null;
+ this.remoteBasePath = remoteBasePath;
+ this.remoteForStPath =
+ remoteBasePath != null ? remoteBasePath.resolve(DB_DIR_STRING) : null;
this.enableStatistics = enableStatistics;
this.handlesToClose = new ArrayList<>();
@@ -156,6 +166,13 @@ public final class ForStResourceContainer implements AutoCloseable {
handlesToClose.add(statistics);
}
+ // TODO: Fallback to checkpoint directory when checkpoint feature is ready if not
+ // configured,
+ // fallback to local directory currently temporarily.
+ if (remoteForStPath != null) {
+ opt.setEnv(new FlinkEnv(remoteForStPath.toString()));
+ }
+
return opt;
}
@@ -241,6 +258,68 @@ public final class ForStResourceContainer implements AutoCloseable {
return localForStPath;
}
+ @Nullable
+ public URI getRemoteBasePath() {
+ return remoteBasePath;
+ }
+
+ /**
+ * Prepare local and remote directories.
+ *
+ * @throws Exception if any unexpected behaviors.
+ */
+ public void prepareDirectories() throws Exception {
+ if (remoteBasePath != null && remoteForStPath != null) {
+ prepareDirectories(remoteBasePath, remoteForStPath);
+ }
+ if (localBasePath != null && localForStPath != null) {
+ prepareDirectories(new URI(localBasePath.getPath()), new URI(localForStPath.getPath()));
+ }
+ }
+
+ private static void prepareDirectories(URI basePath, URI dbPath) throws IOException {
+ FileSystem fileSystem = FileSystem.get(basePath);
+ org.apache.flink.core.fs.Path tempBasePath = new org.apache.flink.core.fs.Path(basePath),
+ tempDBPath = new org.apache.flink.core.fs.Path(dbPath);
+ if (fileSystem.exists(tempBasePath)) {
+ if (!fileSystem.getFileStatus(tempBasePath).isDir()) {
+ throw new IOException("Not a directory: " + tempBasePath);
+ }
+ } else if (!fileSystem.mkdirs(tempBasePath)) {
+ throw new IOException(
+ String.format("Could not create ForSt directory at %s.", tempBasePath));
+ }
+ if (fileSystem.exists(tempDBPath)) {
+ fileSystem.delete(tempDBPath, true);
+ }
+ if (!fileSystem.mkdirs(tempDBPath)) {
+ throw new IOException(
+ String.format("Could not create ForSt db directory at %s.", tempDBPath));
+ }
+ }
+
+ /**
+ * Clear local and remote directories.
+ *
+ * @throws Exception if any unexpected behaviors.
+ */
+ public void clearDirectories() throws Exception {
+ if (remoteBasePath != null) {
+ clearDirectories(remoteBasePath);
+ }
+ if (localBasePath != null) {
+ clearDirectories(new URI(localBasePath.getPath()));
+ }
+ }
+
+ private static void clearDirectories(URI basePath) throws IOException {
+ FileSystem fileSystem = FileSystem.get(basePath);
+ org.apache.flink.core.fs.Path tempBasePath = new org.apache.flink.core.fs.Path(basePath);
+ if (fileSystem.exists(tempBasePath)) {
+ fileSystem.delete(tempBasePath, true);
+ }
+ }
+
ForStNativeMetricOptions getMemoryWatcherOptions(
ForStNativeMetricOptions defaultMetricOptions) {
return optionsFactory == null
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateBackend.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateBackend.java
index be21b54db18..1ab6961350b 100644
--- a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateBackend.java
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateBackend.java
@@ -50,6 +50,7 @@ import java.io.File;
import java.io.IOException;
import java.lang.reflect.Field;
import java.net.URI;
+import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
@@ -84,6 +85,13 @@ public class ForStStateBackend extends AbstractManagedMemoryStateBackend
// -- configuration values, set in the application / configuration
+ /**
+ * Base paths for ForSt remote directory, as configured. Null if not yet set, in which case the
+ * configuration values will be used. The configuration will fallback to local directory by
+ * default. TODO: fallback to checkpoint directory if not configured.
+ */
+ @Nullable private URI remoteForStDirectory;
+
/**
* Base paths for ForSt directory, as configured. Null if not yet set, in which case the
* configuration values will be used. The configuration defaults to the TaskManager's temp
@@ -142,6 +150,21 @@ public class ForStStateBackend extends AbstractManagedMemoryStateBackend
original.memoryConfiguration, config);
this.memoryConfiguration.validate();
+ if (original.remoteForStDirectory != null) {
+ this.remoteForStDirectory = original.remoteForStDirectory;
+ } else {
+ String remoteDirStr = config.get(ForStOptions.REMOTE_DIRECTORY);
+ try {
+ this.remoteForStDirectory = remoteDirStr == null ? null : new URI(remoteDirStr);
+ } catch (URISyntaxException e) {
+ throw new RuntimeException(
+ String.format(
+ "Exception when transform %s to URI, the value is: %s",
+ ForStOptions.REMOTE_DIRECTORY.key(), remoteDirStr),
+ e);
+ }
+ }
+
// configure local directories
if (original.localForStDirectories != null) {
this.localForStDirectories = original.localForStDirectories;
@@ -279,15 +302,12 @@ public class ForStStateBackend extends AbstractManagedMemoryStateBackend
lazyInitializeForJob(env, fileCompatibleIdentifier);
- File instanceBasePath =
- new File(
- getNextStoragePath(),
- "job_"
- + jobId
- + "_op_"
- + fileCompatibleIdentifier
- + "_uuid_"
- + UUID.randomUUID());
+ String childPath =
+ "job_" + jobId + "_op_" + fileCompatibleIdentifier + "_uuid_" + UUID.randomUUID();
+
+ File localBasePath = new File(getNextStoragePath(), childPath);
+ URI remoteBasePath =
+ remoteForStDirectory != null ? remoteForStDirectory.resolve(childPath) : null;
final OpaqueMemoryResource<ForStSharedResources> sharedResources =
ForStOperationUtils.allocateSharedCachesIfConfigured(
@@ -302,7 +322,8 @@ public class ForStStateBackend extends AbstractManagedMemoryStateBackend
final ForStResourceContainer resourceContainer =
createOptionsAndResourceContainer(
sharedResources,
- instanceBasePath,
+ localBasePath,
+ remoteBasePath,
nativeMetricOptions.isStatisticsEnabled());
ForStKeyedStateBackendBuilder<K> builder =
@@ -534,21 +555,23 @@ public class ForStStateBackend extends AbstractManagedMemoryStateBackend
}
@VisibleForTesting
- ForStResourceContainer createOptionsAndResourceContainer(@Nullable File instanceBasePath) {
- return createOptionsAndResourceContainer(null, instanceBasePath, false);
+ ForStResourceContainer createOptionsAndResourceContainer(@Nullable File localBasePath) {
+ return createOptionsAndResourceContainer(null, localBasePath, null, false);
}
@VisibleForTesting
private ForStResourceContainer createOptionsAndResourceContainer(
@Nullable OpaqueMemoryResource<ForStSharedResources> sharedResources,
- @Nullable File instanceBasePath,
+ @Nullable File localBasePath,
+ @Nullable URI remoteBasePath,
boolean enableStatistics) {
return new ForStResourceContainer(
configurableOptions != null ? configurableOptions : new Configuration(),
forStOptionsFactory,
sharedResources,
- instanceBasePath,
+ localBasePath,
+ remoteBasePath,
enableStatistics);
}
@@ -557,6 +580,8 @@ public class ForStStateBackend extends AbstractManagedMemoryStateBackend
return "ForStStateBackend{"
+ ", localForStDirectories="
+ Arrays.toString(localForStDirectories)
+ + ", remoteForStDirectory="
+ + remoteForStDirectory
+ '}';
}
diff --git a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStResourceContainerTest.java b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStResourceContainerTest.java
index 53c0524308b..bec699931c6 100644
--- a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStResourceContainerTest.java
+++ b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStResourceContainerTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.state.forst;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.memory.OpaqueMemoryResource;
import org.apache.flink.util.function.ThrowingRunnable;
@@ -30,6 +31,7 @@ import org.rocksdb.BloomFilter;
import org.rocksdb.Cache;
import org.rocksdb.ColumnFamilyOptions;
import org.rocksdb.DBOptions;
+import org.rocksdb.FlinkEnv;
import org.rocksdb.IndexType;
import org.rocksdb.LRUCache;
import org.rocksdb.NativeLibraryLoader;
@@ -38,8 +40,10 @@ import org.rocksdb.TableFormatConfig;
import org.rocksdb.WriteBufferManager;
import org.rocksdb.WriteOptions;
+import java.io.File;
import java.io.IOException;
import java.lang.reflect.Field;
+import java.net.URI;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
@@ -48,6 +52,7 @@ import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.CoreMatchers.not;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
/** Tests to guard {@link ForStResourceContainer}. */
@@ -294,4 +299,22 @@ public class ForStResourceContainerTest {
}
assertFalse("Block based filter is left unclosed.", blockBasedFilter.isOwningHandle());
}
+
+ @Test
+ public void testDirectoryResources() throws Exception {
+ File localBasePath = TMP_FOLDER.newFolder();
+ URI remoteBasePath = TMP_FOLDER.newFolder().toURI();
+ try (final ForStResourceContainer optionsContainer =
+ new ForStResourceContainer(
+ new Configuration(), null, null, localBasePath, remoteBasePath, false)) {
+ optionsContainer.prepareDirectories();
+ assertTrue(localBasePath.exists());
+ assertTrue(new File(remoteBasePath).exists());
+ assertTrue(optionsContainer.getDbOptions().getEnv() instanceof FlinkEnv);
+
+ optionsContainer.clearDirectories();
+ assertFalse(localBasePath.exists());
+ assertFalse(new File(remoteBasePath).exists());
+ }
+ }
}
diff --git a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateBackendConfigTest.java b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateBackendConfigTest.java
index 11733f0ceca..89bd9a527cd 100644
--- a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateBackendConfigTest.java
+++ b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateBackendConfigTest.java
@@ -144,7 +144,7 @@ public class ForStStateBackendConfigTest {
createKeyedStateBackend(forStStateBackend, env, IntSerializer.INSTANCE);
try {
- File instanceBasePath = keyedBackend.getInstanceBasePath();
+ File instanceBasePath = keyedBackend.getLocalBasePath();
assertThat(
instanceBasePath.getAbsolutePath(),
anyOf(startsWith(testDir1), startsWith(testDir2)));
@@ -207,7 +207,7 @@ public class ForStStateBackendConfigTest {
createKeyedStateBackend(forStBackend, env, IntSerializer.INSTANCE);
try {
- File instanceBasePath = keyedBackend.getInstanceBasePath();
+ File instanceBasePath = keyedBackend.getLocalBasePath();
assertThat(
instanceBasePath.getAbsolutePath(), startsWith(expectedPath.getAbsolutePath()));
@@ -253,26 +253,26 @@ public class ForStStateBackendConfigTest {
conf.set(ForStConfigurableOptions.LOG_LEVEL, InfoLogLevel.DEBUG_LEVEL);
conf.set(ForStConfigurableOptions.LOG_FILE_NUM, 4);
conf.set(ForStConfigurableOptions.LOG_MAX_FILE_SIZE, MemorySize.parse("1kb"));
- final ForStStateBackend rocksDbBackend =
+ final ForStStateBackend forStBackend =
new ForStStateBackend().configure(conf, getClass().getClassLoader());
final String dbStoragePath = new Path(folder.toURI().toString()).toString();
- rocksDbBackend.setLocalDbStoragePath(dbStoragePath);
+ forStBackend.setLocalDbStoragePath(dbStoragePath);
final MockEnvironment env = getMockEnvironment(tempFolder.newFolder());
ForStKeyedStateBackend<Integer> keyedBackend =
- createKeyedStateBackend(rocksDbBackend, env, IntSerializer.INSTANCE);
+ createKeyedStateBackend(forStBackend, env, IntSerializer.INSTANCE);
- File instanceBasePath = keyedBackend.getInstanceBasePath();
- File instanceRocksDBPath = new File(instanceBasePath, "db");
+ File localBasePath = keyedBackend.getLocalBasePath();
+ File localForStPath = new File(localBasePath, "db");
// avoid tests without relocate.
- Assume.assumeTrue(instanceRocksDBPath.getAbsolutePath().length() <= 255 - "_LOG".length());
+ Assume.assumeTrue(localForStPath.getAbsolutePath().length() <= 255 - "_LOG".length());
java.nio.file.Path[] relocatedDbLogs;
try {
relocatedDbLogs = FileUtils.listDirectory(relocatedDBLogDir.toPath());
while (relocatedDbLogs.length <= 2) {
- // If the default number of log files in rocksdb is not enough, add more logs.
+ // If the default number of log files in ForSt is not enough, add more logs.
try (FlushOptions flushOptions = new FlushOptions()) {
keyedBackend.db.put(RandomUtils.nextBytes(32), RandomUtils.nextBytes(512));
keyedBackend.db.flush(flushOptions);
@@ -290,11 +290,11 @@ public class ForStStateBackendConfigTest {
}
// ------------------------------------------------------------------------
- // RocksDB local file automatic from temp directories
+ // ForSt local file automatic from temp directories
// ------------------------------------------------------------------------
/**
- * This tests whether the RocksDB backends uses the temp directories that are provided from the
+ * This tests whether the ForSt backends uses the temp directories that are provided from the
* {@link Environment} when no db storage path is set.
*/
@Test
@@ -326,7 +326,7 @@ public class ForStStateBackendConfigTest {
cancelStreamRegistry));
try {
- File instanceBasePath = keyedBackend.getInstanceBasePath();
+ File instanceBasePath = keyedBackend.getLocalBasePath();
assertThat(instanceBasePath.getAbsolutePath(), startsWith(dir1.getAbsolutePath()));
} finally {
keyedBackend.dispose();
@@ -335,7 +335,7 @@ public class ForStStateBackendConfigTest {
}
// ------------------------------------------------------------------------
- // RocksDB local file directory initialization
+ // ForSt local file directory initialization
// ------------------------------------------------------------------------
@Test
@@ -429,7 +429,7 @@ public class ForStStateBackendConfigTest {
}
// ------------------------------------------------------------------------
- // RocksDB Options
+ // ForSt Options
// ------------------------------------------------------------------------
@Test
public void testConfigurableOptionsFromConfig() throws Exception {
@@ -465,7 +465,7 @@ public class ForStStateBackendConfigTest {
// verify legal configuration
{
configuration.setString(ForStConfigurableOptions.LOG_LEVEL.key(), "DEBUG_LEVEL");
- configuration.setString(ForStConfigurableOptions.LOG_DIR.key(), "/tmp/rocksdb-logs/");
+ configuration.setString(ForStConfigurableOptions.LOG_DIR.key(), "/tmp/ForSt-logs/");
configuration.setString(ForStConfigurableOptions.LOG_FILE_NUM.key(), "10");
configuration.setString(ForStConfigurableOptions.LOG_MAX_FILE_SIZE.key(), "2MB");
configuration.setString(ForStConfigurableOptions.COMPACTION_STYLE.key(), "level");
@@ -486,12 +486,12 @@ public class ForStStateBackendConfigTest {
configuration.setString(ForStConfigurableOptions.USE_BLOOM_FILTER.key(), "TRUE");
try (ForStResourceContainer optionsContainer =
- new ForStResourceContainer(configuration, null, null, null, false)) {
+ new ForStResourceContainer(configuration, null, null, null, null, false)) {
DBOptions dbOptions = optionsContainer.getDbOptions();
assertEquals(-1, dbOptions.maxOpenFiles());
assertEquals(InfoLogLevel.DEBUG_LEVEL, dbOptions.infoLogLevel());
- assertEquals("/tmp/rocksdb-logs/", dbOptions.dbLogDir());
+ assertEquals("/tmp/ForSt-logs/", dbOptions.dbLogDir());
assertEquals(10, dbOptions.keepLogFileNum());
assertEquals(2 * SizeUnit.MB, dbOptions.maxLogFileSize());
@@ -569,7 +569,7 @@ public class ForStStateBackendConfigTest {
Configuration configuration = new Configuration();
configuration.set(ForStConfigurableOptions.COMPACTION_STYLE, CompactionStyle.UNIVERSAL);
try (final ForStResourceContainer optionsContainer =
- new ForStResourceContainer(configuration, null, null, null, false)) {
+ new ForStResourceContainer(configuration, null, null, null, null, false)) {
final ColumnFamilyOptions columnFamilyOptions = optionsContainer.getColumnOptions();
assertNotNull(columnFamilyOptions);
@@ -577,7 +577,7 @@ public class ForStStateBackendConfigTest {
}
try (final ForStResourceContainer optionsContainer =
- new ForStResourceContainer(new Configuration(), null, null, null, false)) {
+ new ForStResourceContainer(new Configuration(), null, null, null, null, false)) {
final ColumnFamilyOptions columnFamilyOptions = optionsContainer.getColumnOptions();
assertNotNull(columnFamilyOptions);
@@ -638,7 +638,7 @@ public class ForStStateBackendConfigTest {
}
// ------------------------------------------------------------------------
- // RocksDB Memory Control
+ // ForSt Memory Control
// ------------------------------------------------------------------------
@Test
@@ -705,6 +705,32 @@ public class ForStStateBackendConfigTest {
}
}
+ @Test
+ public void testRemoteDirectory() throws Exception {
+ FileSystem.initialize(new Configuration(), null);
+ Configuration configuration = new Configuration();
+ configuration.set(ForStOptions.REMOTE_DIRECTORY, tempFolder.newFolder().toURI().toString());
+ ForStStateBackend forStStateBackend =
+ new ForStStateBackend().configure(configuration, null);
+ ForStKeyedStateBackend<Integer> keyedBackend = null;
+ try {
+ keyedBackend =
+ createKeyedStateBackend(
+ forStStateBackend,
+ getMockEnvironment(tempFolder.newFolder()),
+ IntSerializer.INSTANCE);
+ assertTrue(
+ keyedBackend
+ .getRemoteBasePath()
+ .toString()
+ .startsWith(configuration.get(ForStOptions.REMOTE_DIRECTORY)));
+ } finally {
+ if (keyedBackend != null) {
+ keyedBackend.dispose();
+ }
+ }
+ }
+
private void verifySetParameter(Runnable setter) {
try {
setter.run();
@@ -742,7 +768,7 @@ public class ForStStateBackendConfigTest {
/** An implementation of options factory for testing. */
public static class TestOptionsFactory implements ConfigurableForStOptionsFactory {
public static final ConfigOption<Integer> BACKGROUND_JOBS_OPTION =
- ConfigOptions.key("my.custom.rocksdb.backgroundJobs").intType().defaultValue(2);
+ ConfigOptions.key("my.custom.forst.backgroundJobs").intType().defaultValue(2);
private int backgroundJobs = BACKGROUND_JOBS_OPTION.defaultValue();
(flink) 05/07: [FLINK-35047][state] Support restoring and building ForSt
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hangxiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 625d8a5853f835fcd45f2ff256e1309e16feac61
Author: Hangxiang Yu <ma...@gmail.com>
AuthorDate: Mon Apr 15 23:32:17 2024 +0800
[FLINK-35047][state] Support restoring and building ForSt
---
.../flink/state/forst/ForStOperationUtils.java | 138 +++++++++++++++++++++
.../flink/state/forst/restore/ForStHandle.java | 116 +++++++++++++++++
.../forst/restore/ForStNoneRestoreOperation.java | 62 +++++++++
.../state/forst/restore/ForStRestoreOperation.java | 29 +++++
.../state/forst/restore/ForStRestoreResult.java | 55 ++++++++
.../state/forst/ForStOperationsUtilsTest.java | 89 +++++++++++++
6 files changed, 489 insertions(+)
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStOperationUtils.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStOperationUtils.java
new file mode 100644
index 00000000000..2f3d7dbf4fe
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStOperationUtils.java
@@ -0,0 +1,138 @@
+/*
+ * 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.flink.state.forst;
+
+import org.apache.flink.runtime.execution.Environment;
+import org.apache.flink.runtime.memory.OpaqueMemoryResource;
+import org.apache.flink.util.IOUtils;
+import org.apache.flink.util.OperatingSystem;
+import org.apache.flink.util.Preconditions;
+
+import org.rocksdb.ColumnFamilyDescriptor;
+import org.rocksdb.ColumnFamilyHandle;
+import org.rocksdb.ColumnFamilyOptions;
+import org.rocksdb.DBOptions;
+import org.rocksdb.RocksDB;
+import org.rocksdb.RocksDBException;
+import org.slf4j.Logger;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.Function;
+
+/** Utils for ForSt Operations. */
+public class ForStOperationUtils {
+
+ /**
+ * The name of the merge operator in ForSt. Do not change except you know exactly what you do.
+ */
+ public static final String MERGE_OPERATOR_NAME = "stringappendtest";
+
+ public static RocksDB openDB(
+ String path,
+ List<ColumnFamilyDescriptor> stateColumnFamilyDescriptors,
+ List<ColumnFamilyHandle> stateColumnFamilyHandles,
+ ColumnFamilyOptions columnFamilyOptions,
+ DBOptions dbOptions)
+ throws IOException {
+ List<ColumnFamilyDescriptor> columnFamilyDescriptors =
+ new ArrayList<>(1 + stateColumnFamilyDescriptors.size());
+
+ // we add the required descriptor for the default CF in FIRST position, see
+ // https://github.com/facebook/rocksdb/wiki/RocksJava-Basics#opening-a-database-with-column-families
+ columnFamilyDescriptors.add(
+ new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY, columnFamilyOptions));
+ columnFamilyDescriptors.addAll(stateColumnFamilyDescriptors);
+
+ RocksDB dbRef;
+
+ try {
+ dbRef =
+ RocksDB.open(
+ Preconditions.checkNotNull(dbOptions),
+ Preconditions.checkNotNull(path),
+ columnFamilyDescriptors,
+ stateColumnFamilyHandles);
+ } catch (RocksDBException e) {
+ IOUtils.closeQuietly(columnFamilyOptions);
+ columnFamilyDescriptors.forEach((cfd) -> IOUtils.closeQuietly(cfd.getOptions()));
+
+ // improve error reporting on Windows
+ throwExceptionIfPathLengthExceededOnWindows(path, e);
+
+ throw new IOException("Error while opening ForSt instance.", e);
+ }
+
+ // requested + default CF
+ Preconditions.checkState(
+ 1 + stateColumnFamilyDescriptors.size() == stateColumnFamilyHandles.size(),
+ "Not all requested column family handles have been created");
+ return dbRef;
+ }
+
+ public static ColumnFamilyOptions createColumnFamilyOptions(
+ Function<String, ColumnFamilyOptions> columnFamilyOptionsFactory, String stateName) {
+
+ // ensure that we use the right merge operator, because other code relies on this
+ return columnFamilyOptionsFactory
+ .apply(stateName)
+ .setMergeOperatorName(MERGE_OPERATOR_NAME);
+ }
+
+ @Nullable
+ public static OpaqueMemoryResource<ForStSharedResources> allocateSharedCachesIfConfigured(
+ ForStMemoryConfiguration jobMemoryConfig,
+ Environment env,
+ double memoryFraction,
+ Logger logger,
+ ForStMemoryControllerUtils.ForStMemoryFactory forStMemoryFactory)
+ throws IOException {
+
+ try {
+ ForStSharedResourcesFactory factory =
+ ForStSharedResourcesFactory.from(jobMemoryConfig, env);
+ if (factory == null) {
+ return null;
+ }
+
+ return factory.create(jobMemoryConfig, env, memoryFraction, logger, forStMemoryFactory);
+
+ } catch (Exception e) {
+ throw new IOException("Failed to acquire shared cache resource for ForSt", e);
+ }
+ }
+
+ private static void throwExceptionIfPathLengthExceededOnWindows(String path, Exception cause)
+ throws IOException {
+ // max directory path length on Windows is 247.
+ // the maximum path length is 260, subtracting one file name length (12 chars) and one NULL
+ // terminator.
+ final int maxWinDirPathLen = 247;
+
+ if (path.length() > maxWinDirPathLen && OperatingSystem.isWindows()) {
+ throw new IOException(
+ String.format(
+ "The directory path length (%d) is longer than the directory path length limit for Windows (%d): %s",
+ path.length(), maxWinDirPathLen, path),
+ cause);
+ }
+ }
+}
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/restore/ForStHandle.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/restore/ForStHandle.java
new file mode 100644
index 00000000000..167de2a774d
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/restore/ForStHandle.java
@@ -0,0 +1,116 @@
+/*
+ * 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.flink.state.forst.restore;
+
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.state.forst.ForStNativeMetricMonitor;
+import org.apache.flink.state.forst.ForStNativeMetricOptions;
+import org.apache.flink.state.forst.ForStOperationUtils;
+import org.apache.flink.util.IOUtils;
+
+import org.rocksdb.ColumnFamilyDescriptor;
+import org.rocksdb.ColumnFamilyHandle;
+import org.rocksdb.ColumnFamilyOptions;
+import org.rocksdb.DBOptions;
+import org.rocksdb.RocksDB;
+
+import javax.annotation.Nullable;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.function.Function;
+
+/** Utility for creating a ForSt instance either from scratch or from restored local state. */
+class ForStHandle implements AutoCloseable {
+
+ private final Function<String, ColumnFamilyOptions> columnFamilyOptionsFactory;
+ private final DBOptions dbOptions;
+ private final String dbPath;
+ private final List<ColumnFamilyHandle> columnFamilyHandles;
+ private final List<ColumnFamilyDescriptor> columnFamilyDescriptors;
+ private final ForStNativeMetricOptions nativeMetricOptions;
+ private final MetricGroup metricGroup;
+
+ private RocksDB db;
+ private ColumnFamilyHandle defaultColumnFamilyHandle;
+ @Nullable private ForStNativeMetricMonitor nativeMetricMonitor;
+
+ protected ForStHandle(
+ File instanceRocksDBPath,
+ DBOptions dbOptions,
+ Function<String, ColumnFamilyOptions> columnFamilyOptionsFactory,
+ ForStNativeMetricOptions nativeMetricOptions,
+ MetricGroup metricGroup) {
+ this.dbPath = instanceRocksDBPath.getAbsolutePath();
+ this.dbOptions = dbOptions;
+ this.columnFamilyOptionsFactory = columnFamilyOptionsFactory;
+ this.nativeMetricOptions = nativeMetricOptions;
+ this.metricGroup = metricGroup;
+ this.columnFamilyHandles = new ArrayList<>(1);
+ this.columnFamilyDescriptors = Collections.emptyList();
+ }
+
+ void openDB() throws IOException {
+ loadDb();
+ }
+
+ private void loadDb() throws IOException {
+ db =
+ ForStOperationUtils.openDB(
+ dbPath,
+ columnFamilyDescriptors,
+ columnFamilyHandles,
+ ForStOperationUtils.createColumnFamilyOptions(
+ columnFamilyOptionsFactory, "default"),
+ dbOptions);
+ // remove the default column family which is located at the first index
+ defaultColumnFamilyHandle = columnFamilyHandles.remove(0);
+ // init native metrics monitor if configured
+ nativeMetricMonitor =
+ nativeMetricOptions.isEnabled()
+ ? new ForStNativeMetricMonitor(
+ nativeMetricOptions, metricGroup, db, dbOptions.statistics())
+ : null;
+ }
+
+ public RocksDB getDb() {
+ return db;
+ }
+
+ @Nullable
+ public ForStNativeMetricMonitor getNativeMetricMonitor() {
+ return nativeMetricMonitor;
+ }
+
+ public ColumnFamilyHandle getDefaultColumnFamilyHandle() {
+ return defaultColumnFamilyHandle;
+ }
+
+ @Override
+ public void close() {
+ IOUtils.closeQuietly(defaultColumnFamilyHandle);
+ IOUtils.closeQuietly(nativeMetricMonitor);
+ IOUtils.closeQuietly(db);
+ // Making sure the already created column family options will be closed
+ columnFamilyDescriptors.forEach((cfd) -> IOUtils.closeQuietly(cfd.getOptions()));
+ }
+}
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/restore/ForStNoneRestoreOperation.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/restore/ForStNoneRestoreOperation.java
new file mode 100644
index 00000000000..acf4f893440
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/restore/ForStNoneRestoreOperation.java
@@ -0,0 +1,62 @@
+/*
+ * 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.flink.state.forst.restore;
+
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.state.forst.ForStNativeMetricOptions;
+
+import org.rocksdb.ColumnFamilyOptions;
+import org.rocksdb.DBOptions;
+
+import java.io.File;
+import java.util.function.Function;
+
+/** Encapsulates the process of initiating a ForSt instance without restore. */
+public class ForStNoneRestoreOperation implements ForStRestoreOperation {
+ private final ForStHandle rocksHandle;
+
+ public ForStNoneRestoreOperation(
+ File instanceRocksDBPath,
+ DBOptions dbOptions,
+ Function<String, ColumnFamilyOptions> columnFamilyOptionsFactory,
+ ForStNativeMetricOptions nativeMetricOptions,
+ MetricGroup metricGroup) {
+ this.rocksHandle =
+ new ForStHandle(
+ instanceRocksDBPath,
+ dbOptions,
+ columnFamilyOptionsFactory,
+ nativeMetricOptions,
+ metricGroup);
+ }
+
+ @Override
+ public ForStRestoreResult restore() throws Exception {
+ this.rocksHandle.openDB();
+ return new ForStRestoreResult(
+ this.rocksHandle.getDb(),
+ this.rocksHandle.getDefaultColumnFamilyHandle(),
+ this.rocksHandle.getNativeMetricMonitor());
+ }
+
+ @Override
+ public void close() {
+ this.rocksHandle.close();
+ }
+}
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/restore/ForStRestoreOperation.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/restore/ForStRestoreOperation.java
new file mode 100644
index 00000000000..4aa0ffb1265
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/restore/ForStRestoreOperation.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.flink.state.forst.restore;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.state.RestoreOperation;
+
+/** Interface for ForSt restore. */
+@Internal
+public interface ForStRestoreOperation extends RestoreOperation<ForStRestoreResult>, AutoCloseable {
+ /** Restores state that was previously snapshot-ed from the provided state handles. */
+ ForStRestoreResult restore() throws Exception;
+}
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/restore/ForStRestoreResult.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/restore/ForStRestoreResult.java
new file mode 100644
index 00000000000..92a89f02ab5
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/restore/ForStRestoreResult.java
@@ -0,0 +1,55 @@
+/*
+ * 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.flink.state.forst.restore;
+
+import org.apache.flink.state.forst.ForStNativeMetricMonitor;
+
+import org.rocksdb.ColumnFamilyHandle;
+import org.rocksdb.RocksDB;
+
+import javax.annotation.Nullable;
+
+/** Entity holding result of ForSt instance restore. */
+public class ForStRestoreResult {
+ private final RocksDB db;
+ private final ColumnFamilyHandle defaultColumnFamilyHandle;
+ @Nullable private final ForStNativeMetricMonitor nativeMetricMonitor;
+
+ public ForStRestoreResult(
+ RocksDB db,
+ ColumnFamilyHandle defaultColumnFamilyHandle,
+ ForStNativeMetricMonitor nativeMetricMonitor) {
+ this.db = db;
+ this.defaultColumnFamilyHandle = defaultColumnFamilyHandle;
+ this.nativeMetricMonitor = nativeMetricMonitor;
+ }
+
+ public RocksDB getDb() {
+ return db;
+ }
+
+ public ColumnFamilyHandle getDefaultColumnFamilyHandle() {
+ return defaultColumnFamilyHandle;
+ }
+
+ @Nullable
+ public ForStNativeMetricMonitor getNativeMetricMonitor() {
+ return nativeMetricMonitor;
+ }
+}
diff --git a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStOperationsUtilsTest.java b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStOperationsUtilsTest.java
new file mode 100644
index 00000000000..e20382aff9d
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStOperationsUtilsTest.java
@@ -0,0 +1,89 @@
+/*
+ * 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.flink.state.forst;
+
+import org.apache.flink.util.OperatingSystem;
+
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.rocksdb.ColumnFamilyOptions;
+import org.rocksdb.DBOptions;
+import org.rocksdb.NativeLibraryLoader;
+import org.rocksdb.RocksDB;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.util.Collections;
+
+import static org.hamcrest.Matchers.containsString;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assume.assumeTrue;
+
+/** Tests for the {@link ForStOperationUtils}. */
+public class ForStOperationsUtilsTest {
+
+ @ClassRule public static final TemporaryFolder TMP_DIR = new TemporaryFolder();
+
+ @BeforeClass
+ public static void loadRocksLibrary() throws Exception {
+ NativeLibraryLoader.getInstance().loadLibrary(TMP_DIR.newFolder().getAbsolutePath());
+ }
+
+ @Test
+ public void testPathExceptionOnWindows() throws Exception {
+ assumeTrue(OperatingSystem.isWindows());
+
+ final File folder = TMP_DIR.newFolder();
+ final File rocksDir =
+ new File(folder, getLongString(247 - folder.getAbsolutePath().length()));
+
+ Files.createDirectories(rocksDir.toPath());
+
+ try (DBOptions dbOptions = new DBOptions().setCreateIfMissing(true);
+ ColumnFamilyOptions colOptions = new ColumnFamilyOptions()) {
+
+ RocksDB rocks =
+ ForStOperationUtils.openDB(
+ rocksDir.getAbsolutePath(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ colOptions,
+ dbOptions);
+ rocks.close();
+
+ // do not provoke a test failure if this passes, because some setups may actually
+ // support long paths, in which case: great!
+ } catch (IOException e) {
+ assertThat(
+ e.getMessage(),
+ containsString("longer than the directory path length limit for Windows"));
+ }
+ }
+
+ private static String getLongString(int numChars) {
+ final StringBuilder builder = new StringBuilder();
+ for (int i = numChars; i > 0; --i) {
+ builder.append('a');
+ }
+ return builder.toString();
+ }
+}
(flink) 06/07: [FLINK-35047][state] Introduce ForStStateBackend and ForStKeyedStateBackend
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hangxiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 417484e6029c879a38b698434b9a2f8ba94413fb
Author: Hangxiang Yu <ma...@gmail.com>
AuthorDate: Tue Apr 16 23:34:36 2024 +0800
[FLINK-35047][state] Introduce ForStStateBackend and ForStKeyedStateBackend
---
.../flink/state/forst/ForStKeyedStateBackend.java | 134 ++++
.../state/forst/ForStKeyedStateBackendBuilder.java | 187 +++++
.../flink/state/forst/ForStStateBackend.java | 655 ++++++++++++++++++
.../state/forst/ForStStateBackendFactory.java | 34 +
.../apache/flink/state/forst/ForStInitITCase.java | 64 ++
.../state/forst/ForStMultiClassLoaderTest.java | 88 +++
.../state/forst/ForStStateBackendConfigTest.java | 767 +++++++++++++++++++++
.../state/forst/ForStStateBackendFactoryTest.java | 132 ++++
.../apache/flink/state/forst/ForStTestUtils.java | 54 ++
9 files changed, 2115 insertions(+)
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStKeyedStateBackend.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStKeyedStateBackend.java
new file mode 100644
index 00000000000..8c7dfd0ba0f
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStKeyedStateBackend.java
@@ -0,0 +1,134 @@
+/*
+ * 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.flink.state.forst;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.util.FileUtils;
+import org.apache.flink.util.IOUtils;
+import org.apache.flink.util.Preconditions;
+
+import org.rocksdb.ColumnFamilyHandle;
+import org.rocksdb.RocksDB;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+
+/**
+ * A KeyedStateBackend that stores its state in {@code ForSt}. This state backend can store very
+ * large state that exceeds memory even disk to remote storage. TODO: Support to implement the new
+ * interface of KeyedStateBackend
+ */
+public class ForStKeyedStateBackend<K> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ForStKeyedStateBackend.class);
+
+ /** The key serializer. */
+ protected final TypeSerializer<K> keySerializer;
+
+ /** The container of ForSt options. */
+ private final ForStResourceContainer optionsContainer;
+
+ /** Path where this configured instance stores its local data directory. */
+ private final File localBasePath;
+
+ /**
+ * We are not using the default column family for Flink state ops, but we still need to remember
+ * this handle so that we can close it properly when the backend is closed. Note that the one
+ * returned by {@link RocksDB#open(String)} is different from that by {@link
+ * RocksDB#getDefaultColumnFamily()}, probably it's a bug of RocksDB java API.
+ */
+ private final ColumnFamilyHandle defaultColumnFamily;
+
+ /** The native metrics monitor. */
+ private final ForStNativeMetricMonitor nativeMetricMonitor;
+
+ /**
+ * Our ForSt database. The different k/v states that we have don't each have their own ForSt
+ * instance. They all write to this instance but to their own column family.
+ */
+ protected final RocksDB db;
+
+ // mark whether this backend is already disposed and prevent duplicate disposing
+ private boolean disposed = false;
+
+ public ForStKeyedStateBackend(
+ File localBasePath,
+ ForStResourceContainer optionsContainer,
+ TypeSerializer<K> keySerializer,
+ RocksDB db,
+ ColumnFamilyHandle defaultColumnFamilyHandle,
+ ForStNativeMetricMonitor nativeMetricMonitor) {
+
+ this.localBasePath = Preconditions.checkNotNull(localBasePath);
+ this.optionsContainer = Preconditions.checkNotNull(optionsContainer);
+ this.keySerializer = keySerializer;
+ this.db = db;
+ this.defaultColumnFamily = defaultColumnFamilyHandle;
+ this.nativeMetricMonitor = nativeMetricMonitor;
+ }
+
+ /** Should only be called by one thread, and only after all accesses to the DB happened. */
+ public void dispose() {
+ if (this.disposed) {
+ return;
+ }
+
+ // IMPORTANT: null reference to signal potential async checkpoint workers that the db was
+ // disposed, as
+ // working on the disposed object results in SEGFAULTS.
+ if (db != null) {
+
+ // Metric collection occurs on a background thread. When this method returns
+ // it is guaranteed that thr ForSt reference has been invalidated
+ // and no more metric collection will be attempted against the database.
+ if (nativeMetricMonitor != null) {
+ nativeMetricMonitor.close();
+ }
+
+ IOUtils.closeQuietly(defaultColumnFamily);
+
+ // ... and finally close the DB instance ...
+ IOUtils.closeQuietly(db);
+
+ IOUtils.closeQuietly(optionsContainer);
+
+ cleanLocalBasePath();
+ }
+ this.disposed = true;
+ }
+
+ private void cleanLocalBasePath() {
+ LOG.info(
+ "Closed ForSt State Backend. Cleaning up ForSt local working directory {}.",
+ localBasePath);
+
+ try {
+ FileUtils.deleteDirectory(localBasePath);
+ } catch (IOException ex) {
+ LOG.warn("Could not delete ForSt local working directory: {}", localBasePath, ex);
+ }
+ }
+
+ @VisibleForTesting
+ File getInstanceBasePath() {
+ return localBasePath;
+ }
+}
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStKeyedStateBackendBuilder.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStKeyedStateBackendBuilder.java
new file mode 100644
index 00000000000..1ff68f413a4
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStKeyedStateBackendBuilder.java
@@ -0,0 +1,187 @@
+/*
+ * 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.flink.state.forst;
+
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.runtime.state.BackendBuildingException;
+import org.apache.flink.runtime.state.KeyedStateHandle;
+import org.apache.flink.runtime.state.StateBackendBuilder;
+import org.apache.flink.runtime.state.StateSerializerProvider;
+import org.apache.flink.state.forst.restore.ForStNoneRestoreOperation;
+import org.apache.flink.state.forst.restore.ForStRestoreOperation;
+import org.apache.flink.state.forst.restore.ForStRestoreResult;
+import org.apache.flink.util.CollectionUtil;
+import org.apache.flink.util.FileUtils;
+import org.apache.flink.util.IOUtils;
+import org.apache.flink.util.Preconditions;
+
+import org.rocksdb.ColumnFamilyHandle;
+import org.rocksdb.ColumnFamilyOptions;
+import org.rocksdb.DBOptions;
+import org.rocksdb.RocksDB;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nonnull;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Collection;
+import java.util.function.Function;
+
+/**
+ * Builder class for {@link ForStKeyedStateBackend} which handles all necessary initializations and
+ * cleanups.
+ *
+ * @param <K> The data type that the key serializer serializes.
+ */
+public class ForStKeyedStateBackendBuilder<K>
+ implements StateBackendBuilder<ForStKeyedStateBackend<K>, BackendBuildingException> {
+
+ protected final Logger logger = LoggerFactory.getLogger(getClass());
+
+ private final StateSerializerProvider<K> keySerializerProvider;
+ private final Collection<KeyedStateHandle> restoreStateHandles;
+
+ /** Factory function to create column family options from state name. */
+ private final Function<String, ColumnFamilyOptions> columnFamilyOptionsFactory;
+
+ /** The container of ForSt option factory and predefined options. */
+ private final ForStResourceContainer optionsContainer;
+
+ /** Path where this configured instance stores its data directory. */
+ private final File localBasePath;
+
+ /** Path where this configured instance stores its ForSt database. */
+ private final File localForStPath;
+
+ private final MetricGroup metricGroup;
+
+ /** ForSt property-based and statistics-based native metrics options. */
+ private ForStNativeMetricOptions nativeMetricOptions;
+
+ public ForStKeyedStateBackendBuilder(
+ ForStResourceContainer optionsContainer,
+ Function<String, ColumnFamilyOptions> columnFamilyOptionsFactory,
+ TypeSerializer<K> keySerializer,
+ MetricGroup metricGroup,
+ @Nonnull Collection<KeyedStateHandle> stateHandles) {
+ this.optionsContainer = optionsContainer;
+ this.localBasePath = optionsContainer.getLocalBasePath();
+ this.localForStPath = optionsContainer.getLocalForStPath();
+ this.columnFamilyOptionsFactory = Preconditions.checkNotNull(columnFamilyOptionsFactory);
+ this.keySerializerProvider =
+ StateSerializerProvider.fromNewRegisteredSerializer(keySerializer);
+ this.metricGroup = metricGroup;
+ this.restoreStateHandles = stateHandles;
+ this.nativeMetricOptions = new ForStNativeMetricOptions();
+ }
+
+ ForStKeyedStateBackendBuilder<K> setNativeMetricOptions(
+ ForStNativeMetricOptions nativeMetricOptions) {
+ this.nativeMetricOptions = nativeMetricOptions;
+ return this;
+ }
+
+ @Override
+ public ForStKeyedStateBackend<K> build() throws BackendBuildingException {
+ ColumnFamilyHandle defaultColumnFamilyHandle = null;
+ ForStNativeMetricMonitor nativeMetricMonitor = null;
+ RocksDB db = null;
+ ForStRestoreOperation restoreOperation = null;
+
+ try {
+ prepareLocalDirectories();
+ restoreOperation = getForStRestoreOperation();
+ ForStRestoreResult restoreResult = restoreOperation.restore();
+ db = restoreResult.getDb();
+ defaultColumnFamilyHandle = restoreResult.getDefaultColumnFamilyHandle();
+ nativeMetricMonitor = restoreResult.getNativeMetricMonitor();
+
+ // TODO: Support to init snapshot strategy
+
+ } catch (Throwable e) {
+ // Do clean up
+ IOUtils.closeQuietly(defaultColumnFamilyHandle);
+ IOUtils.closeQuietly(nativeMetricMonitor);
+ IOUtils.closeQuietly(db);
+ // it's possible that db has been initialized but later restore steps failed
+ IOUtils.closeQuietly(restoreOperation);
+ IOUtils.closeQuietly(optionsContainer);
+ try {
+ FileUtils.deleteDirectory(localBasePath);
+ } catch (Exception ex) {
+ logger.warn("Failed to delete local base path for ForSt: " + localBasePath, ex);
+ }
+ // Log and rethrow
+ if (e instanceof BackendBuildingException) {
+ throw (BackendBuildingException) e;
+ } else {
+ String errMsg = "Caught unexpected exception.";
+ logger.error(errMsg, e);
+ throw new BackendBuildingException(errMsg, e);
+ }
+ }
+ logger.info(
+ "Finished building ForSt keyed state-backend at local base path: {}.",
+ localBasePath);
+ return new ForStKeyedStateBackend<>(
+ this.localBasePath,
+ this.optionsContainer,
+ this.keySerializerProvider.currentSchemaSerializer(),
+ db,
+ defaultColumnFamilyHandle,
+ nativeMetricMonitor);
+ }
+
+ private ForStRestoreOperation getForStRestoreOperation() {
+ DBOptions dbOptions = optionsContainer.getDbOptions();
+ if (CollectionUtil.isEmptyOrAllElementsNull(restoreStateHandles)) {
+ return new ForStNoneRestoreOperation(
+ localForStPath,
+ dbOptions,
+ columnFamilyOptionsFactory,
+ nativeMetricOptions,
+ metricGroup);
+ }
+ // TODO: Support Restoring
+ throw new UnsupportedOperationException("Not support restoring yet for ForStStateBackend");
+ }
+
+ private void prepareLocalDirectories() throws IOException {
+ checkAndCreateDirectory(localBasePath);
+ if (localForStPath.exists()) {
+ // Clear the base directory when the backend is created
+ // in case something crashed and the backend never reached dispose()
+ FileUtils.deleteDirectory(localBasePath);
+ }
+ }
+
+ private static void checkAndCreateDirectory(File directory) throws IOException {
+ if (directory.exists()) {
+ if (!directory.isDirectory()) {
+ throw new IOException("Not a directory: " + directory);
+ }
+ } else if (!directory.mkdirs()) {
+ throw new IOException(
+ String.format("Could not create ForSt data directory at %s.", directory));
+ }
+ }
+}
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateBackend.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateBackend.java
new file mode 100644
index 00000000000..be21b54db18
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateBackend.java
@@ -0,0 +1,655 @@
+/*
+ * 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.flink.state.forst;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.IllegalConfigurationException;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.execution.Environment;
+import org.apache.flink.runtime.memory.OpaqueMemoryResource;
+import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
+import org.apache.flink.runtime.state.AbstractManagedMemoryStateBackend;
+import org.apache.flink.runtime.state.ConfigurableStateBackend;
+import org.apache.flink.runtime.state.DefaultOperatorStateBackendBuilder;
+import org.apache.flink.runtime.state.OperatorStateBackend;
+import org.apache.flink.state.forst.ForStMemoryControllerUtils.ForStMemoryFactory;
+import org.apache.flink.util.AbstractID;
+import org.apache.flink.util.DynamicCodeLoadingException;
+import org.apache.flink.util.FileUtils;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import org.rocksdb.NativeLibraryLoader;
+import org.rocksdb.RocksDB;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.File;
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Optional;
+import java.util.Random;
+import java.util.UUID;
+import java.util.function.Supplier;
+
+/**
+ * A {@link org.apache.flink.runtime.state.StateBackend} that stores its state in a ForSt instance.
+ * This state backend can store very large state that exceeds memory even disk and spills to remote
+ * storage.
+ *
+ * <p>The behavior of the ForSt instances can be parametrized by setting ForSt Options using the
+ * methods {@link #setForStOptions(ForStOptionsFactory)}.
+ */
+@Experimental
+public class ForStStateBackend extends AbstractManagedMemoryStateBackend
+ implements ConfigurableStateBackend {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final Logger LOG = LoggerFactory.getLogger(ForStStateBackend.class);
+
+ /** The number of (re)tries for loading the ForSt JNI library. */
+ private static final int FORST_LIB_LOADING_ATTEMPTS = 3;
+
+ /** Flag whether the native library has been loaded. */
+ private static boolean forStInitialized = false;
+
+ // ------------------------------------------------------------------------
+
+ // -- configuration values, set in the application / configuration
+
+ /**
+ * Base paths for ForSt directory, as configured. Null if not yet set, in which case the
+ * configuration values will be used. The configuration defaults to the TaskManager's temp
+ * directories.
+ */
+ @Nullable private File[] localForStDirectories;
+
+ /** The configurable options. */
+ @Nullable private ReadableConfig configurableOptions;
+
+ /** The options factory to create the ForSt options in the cluster. */
+ @Nullable private ForStOptionsFactory forStOptionsFactory;
+
+ /** The configuration for memory settings (pool sizes, etc.). */
+ private final ForStMemoryConfiguration memoryConfiguration;
+
+ /** The default ForSt property-based metrics options. */
+ private final ForStNativeMetricOptions nativeMetricOptions;
+
+ // -- runtime values, set on TaskManager when initializing / using the backend
+
+ /** Base paths for ForSt directory, as initialized. */
+ private transient File[] initializedDbBasePaths;
+
+ /** JobID for uniquifying backup paths. */
+ private transient JobID jobId;
+
+ /** The index of the next directory to be used from {@link #initializedDbBasePaths}. */
+ private transient int nextDirectory;
+
+ /** Whether we already lazily initialized our local storage directories. */
+ private transient boolean isInitialized;
+
+ /** Factory for Write Buffer Manager and Block Cache. */
+ private final ForStMemoryFactory forStMemoryFactory;
+ // ------------------------------------------------------------------------
+
+ /** Creates a new {@code ForStStateBackend} for storing state. */
+ public ForStStateBackend() {
+ this.nativeMetricOptions = new ForStNativeMetricOptions();
+ this.memoryConfiguration = new ForStMemoryConfiguration();
+ this.forStMemoryFactory = ForStMemoryFactory.DEFAULT;
+ }
+
+ /**
+ * Private constructor that creates a re-configured copy of the state backend.
+ *
+ * @param original The state backend to re-configure.
+ * @param config The configuration.
+ * @param classLoader The class loader.
+ */
+ private ForStStateBackend(
+ ForStStateBackend original, ReadableConfig config, ClassLoader classLoader) {
+ this.memoryConfiguration =
+ ForStMemoryConfiguration.fromOtherAndConfiguration(
+ original.memoryConfiguration, config);
+ this.memoryConfiguration.validate();
+
+ // configure local directories
+ if (original.localForStDirectories != null) {
+ this.localForStDirectories = original.localForStDirectories;
+ } else {
+ final String forStLocalPaths = config.get(ForStOptions.LOCAL_DIRECTORIES);
+ if (forStLocalPaths != null) {
+ String[] directories = forStLocalPaths.split(",|" + File.pathSeparator);
+
+ try {
+ setLocalDbStoragePaths(directories);
+ } catch (IllegalArgumentException e) {
+ throw new IllegalConfigurationException(
+ "Invalid configuration for ForSt state "
+ + "backend's local storage directories: "
+ + e.getMessage(),
+ e);
+ }
+ }
+ }
+
+ // configure metric options
+ this.nativeMetricOptions = ForStNativeMetricOptions.fromConfig(config);
+
+ // configurable options
+ this.configurableOptions = mergeConfigurableOptions(original.configurableOptions, config);
+
+ // configure ForSt options factory
+ try {
+ forStOptionsFactory =
+ configureOptionsFactory(
+ original.forStOptionsFactory,
+ config.get(ForStOptions.OPTIONS_FACTORY),
+ config,
+ classLoader);
+ } catch (DynamicCodeLoadingException e) {
+ throw new FlinkRuntimeException(e);
+ }
+
+ // configure latency tracking
+ latencyTrackingConfigBuilder = original.latencyTrackingConfigBuilder.configure(config);
+
+ this.forStMemoryFactory = original.forStMemoryFactory;
+ }
+
+ // ------------------------------------------------------------------------
+ // Reconfiguration
+ // ------------------------------------------------------------------------
+
+ /**
+ * Creates a copy of this state backend that uses the values defined in the configuration for
+ * fields where that were not yet specified in this state backend.
+ *
+ * @param config The configuration.
+ * @param classLoader The class loader.
+ * @return The re-configured variant of the state backend
+ */
+ @Override
+ public ForStStateBackend configure(ReadableConfig config, ClassLoader classLoader) {
+ return new ForStStateBackend(this, config, classLoader);
+ }
+
+ // ------------------------------------------------------------------------
+ // State backend methods
+ // ------------------------------------------------------------------------
+
+ private void lazyInitializeForJob(
+ Environment env, @SuppressWarnings("unused") String operatorIdentifier)
+ throws IOException {
+
+ if (isInitialized) {
+ return;
+ }
+
+ this.jobId = env.getJobID();
+
+ // initialize the paths where the local ForSt files should be stored
+ if (localForStDirectories == null) {
+ initializedDbBasePaths = new File[] {env.getTaskManagerInfo().getTmpWorkingDirectory()};
+ } else {
+ List<File> dirs = new ArrayList<>(localForStDirectories.length);
+ StringBuilder errorMessage = new StringBuilder();
+
+ for (File f : localForStDirectories) {
+ File testDir = new File(f, UUID.randomUUID().toString());
+ if (!testDir.mkdirs()) {
+ String msg =
+ "Local DB files directory '"
+ + f
+ + "' does not exist and cannot be created. ";
+ LOG.error(msg);
+ errorMessage.append(msg);
+ } else {
+ dirs.add(f);
+ }
+ //noinspection ResultOfMethodCallIgnored
+ testDir.delete();
+ }
+
+ if (dirs.isEmpty()) {
+ throw new IOException("No local storage directories available. " + errorMessage);
+ } else {
+ initializedDbBasePaths = dirs.toArray(new File[0]);
+ }
+ }
+
+ nextDirectory = new Random().nextInt(initializedDbBasePaths.length);
+
+ isInitialized = true;
+ }
+
+ private File getNextStoragePath() {
+ int ni = nextDirectory + 1;
+ ni = ni >= initializedDbBasePaths.length ? 0 : ni;
+ nextDirectory = ni;
+
+ return initializedDbBasePaths[ni];
+ }
+
+ // ------------------------------------------------------------------------
+ // State holding data structures
+ // ------------------------------------------------------------------------
+
+ public <K> ForStKeyedStateBackend<K> createForStKeyedStateBackend(
+ KeyedStateBackendParameters<K> parameters) throws IOException {
+ Environment env = parameters.getEnv();
+
+ // first, make sure that the ForSt JNI library is loaded
+ // we do this explicitly here to have better error handling
+ String tempDir = env.getTaskManagerInfo().getTmpWorkingDirectory().getAbsolutePath();
+ ensureForStIsLoaded(tempDir);
+
+ // replace all characters that are not legal for filenames with underscore
+ String fileCompatibleIdentifier =
+ parameters.getOperatorIdentifier().replaceAll("[^a-zA-Z0-9\\-]", "_");
+
+ lazyInitializeForJob(env, fileCompatibleIdentifier);
+
+ File instanceBasePath =
+ new File(
+ getNextStoragePath(),
+ "job_"
+ + jobId
+ + "_op_"
+ + fileCompatibleIdentifier
+ + "_uuid_"
+ + UUID.randomUUID());
+
+ final OpaqueMemoryResource<ForStSharedResources> sharedResources =
+ ForStOperationUtils.allocateSharedCachesIfConfigured(
+ memoryConfiguration,
+ env,
+ parameters.getManagedMemoryFraction(),
+ LOG,
+ forStMemoryFactory);
+ if (sharedResources != null) {
+ LOG.info("Obtained shared ForSt cache of size {} bytes", sharedResources.getSize());
+ }
+ final ForStResourceContainer resourceContainer =
+ createOptionsAndResourceContainer(
+ sharedResources,
+ instanceBasePath,
+ nativeMetricOptions.isStatisticsEnabled());
+
+ ForStKeyedStateBackendBuilder<K> builder =
+ new ForStKeyedStateBackendBuilder<>(
+ resourceContainer,
+ stateName -> resourceContainer.getColumnOptions(),
+ parameters.getKeySerializer(),
+ parameters.getMetricGroup(),
+ parameters.getStateHandles())
+ .setNativeMetricOptions(
+ resourceContainer.getMemoryWatcherOptions(nativeMetricOptions));
+ return builder.build();
+ }
+
+ @Override
+ public <K> AbstractKeyedStateBackend<K> createKeyedStateBackend(
+ KeyedStateBackendParameters<K> parameters) {
+ throw new UnsupportedOperationException("Don't support createKeyedStateBackend yet");
+ }
+
+ @Override
+ public OperatorStateBackend createOperatorStateBackend(
+ OperatorStateBackendParameters parameters) throws Exception {
+ // the default for ForSt; eventually there can be a operator state backend based on
+ // ForSt, too.
+ final boolean asyncSnapshots = true;
+ return new DefaultOperatorStateBackendBuilder(
+ parameters.getEnv().getUserCodeClassLoader().asClassLoader(),
+ parameters.getEnv().getExecutionConfig(),
+ asyncSnapshots,
+ parameters.getStateHandles(),
+ parameters.getCancelStreamRegistry())
+ .build();
+ }
+
+ private ForStOptionsFactory configureOptionsFactory(
+ @Nullable ForStOptionsFactory originalOptionsFactory,
+ @Nullable String factoryClassName,
+ ReadableConfig config,
+ ClassLoader classLoader)
+ throws DynamicCodeLoadingException {
+
+ ForStOptionsFactory optionsFactory = null;
+
+ if (originalOptionsFactory != null) {
+ if (originalOptionsFactory instanceof ConfigurableForStOptionsFactory) {
+ originalOptionsFactory =
+ ((ConfigurableForStOptionsFactory) originalOptionsFactory)
+ .configure(config);
+ }
+ LOG.info("Using application-defined options factory: {}.", originalOptionsFactory);
+
+ optionsFactory = originalOptionsFactory;
+ } else if (factoryClassName != null) {
+ try {
+ Class<? extends ForStOptionsFactory> clazz =
+ Class.forName(factoryClassName, false, classLoader)
+ .asSubclass(ForStOptionsFactory.class);
+
+ optionsFactory = clazz.newInstance();
+ if (optionsFactory instanceof ConfigurableForStOptionsFactory) {
+ optionsFactory =
+ ((ConfigurableForStOptionsFactory) optionsFactory).configure(config);
+ }
+ LOG.info("Using configured options factory: {}.", optionsFactory);
+
+ } catch (ClassNotFoundException e) {
+ throw new DynamicCodeLoadingException(
+ "Cannot find configured options factory class: " + factoryClassName, e);
+ } catch (ClassCastException | InstantiationException | IllegalAccessException e) {
+ throw new DynamicCodeLoadingException(
+ "The class configured under '"
+ + ForStOptions.OPTIONS_FACTORY.key()
+ + "' is not a valid options factory ("
+ + factoryClassName
+ + ')',
+ e);
+ }
+ }
+
+ return optionsFactory;
+ }
+
+ // ------------------------------------------------------------------------
+ // Parameters
+ // ------------------------------------------------------------------------
+
+ /**
+ * Sets the path where the ForSt local files should be stored on the local file system. Setting
+ * this path overrides the default behavior, where the files are stored across the configured
+ * temp directories.
+ *
+ * <p>Passing {@code null} to this function restores the default behavior, where the configured
+ * temp directories will be used.
+ *
+ * @param path The path where the local ForSt database files are stored.
+ */
+ public void setLocalDbStoragePath(String path) {
+ setLocalDbStoragePaths(path == null ? null : new String[] {path});
+ }
+
+ /**
+ * Sets the local directories in which the ForSt database puts some files (like metadata files).
+ * These directories do not need to be persistent, they can be ephemeral, meaning that they are
+ * lost on a machine failure, because state in ForSt is persisted in checkpoints.
+ *
+ * <p>If nothing is configured, these directories default to the TaskManager's local temporary
+ * file directories.
+ *
+ * <p>Each distinct state will be stored in one path, but when the state backend creates
+ * multiple states, they will store their files on different paths.
+ *
+ * <p>Passing {@code null} to this function restores the default behavior, where the configured
+ * temp directories will be used.
+ *
+ * @param paths The paths across which the local ForSt database files will be spread.
+ */
+ public void setLocalDbStoragePaths(String... paths) {
+ if (paths == null) {
+ localForStDirectories = null;
+ } else if (paths.length == 0) {
+ throw new IllegalArgumentException("empty paths");
+ } else {
+ File[] pp = new File[paths.length];
+
+ for (int i = 0; i < paths.length; i++) {
+ final String rawPath = paths[i];
+ final String path;
+
+ if (rawPath == null) {
+ throw new IllegalArgumentException("null path");
+ } else {
+ // we need this for backwards compatibility, to allow URIs like 'file:///'...
+ URI uri = null;
+ try {
+ uri = new Path(rawPath).toUri();
+ } catch (Exception e) {
+ // cannot parse as a path
+ }
+
+ if (uri != null && uri.getScheme() != null) {
+ if ("file".equalsIgnoreCase(uri.getScheme())) {
+ path = uri.getPath();
+ } else {
+ throw new IllegalArgumentException(
+ "Path " + rawPath + " has a non-local scheme");
+ }
+ } else {
+ path = rawPath;
+ }
+ }
+
+ pp[i] = new File(path);
+ if (!pp[i].isAbsolute()) {
+ throw new IllegalArgumentException("Relative paths are not supported");
+ }
+ }
+
+ localForStDirectories = pp;
+ }
+ }
+
+ /**
+ * Gets the configured local DB storage paths, or null, if none were configured.
+ *
+ * <p>Under these directories on the TaskManager, ForSt stores some metadata files. These
+ * directories do not need to be persistent, they can be ephermeral, meaning that they are lost
+ * on a machine failure, because state in ForSt is persisted in checkpoints.
+ *
+ * <p>If nothing is configured, these directories default to the TaskManager's local temporary
+ * file directories.
+ */
+ public String[] getLocalDbStoragePaths() {
+ if (localForStDirectories == null) {
+ return null;
+ } else {
+ String[] paths = new String[localForStDirectories.length];
+ for (int i = 0; i < paths.length; i++) {
+ paths[i] = localForStDirectories[i].toString();
+ }
+ return paths;
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // Parametrize with ForSt Options
+ // ------------------------------------------------------------------------
+
+ /**
+ * Sets {@link org.rocksdb.Options} for the ForSt instances. Because the options are not
+ * serializable and hold native code references, they must be specified through a factory.
+ *
+ * <p>The options created by the factory here are applied on top of user-configured options from
+ * configuration set by {@link #configure(ReadableConfig, ClassLoader)} with keys in {@link
+ * ForStConfigurableOptions}.
+ *
+ * @param optionsFactory The options factory that lazily creates the ForSt options.
+ */
+ public void setForStOptions(ForStOptionsFactory optionsFactory) {
+ this.forStOptionsFactory = optionsFactory;
+ }
+
+ /** Gets {@link org.rocksdb.Options} for the ForSt instances. */
+ @Nullable
+ public ForStOptionsFactory getForStOptions() {
+ return forStOptionsFactory;
+ }
+
+ // ------------------------------------------------------------------------
+ // utilities
+ // ------------------------------------------------------------------------
+
+ private ReadableConfig mergeConfigurableOptions(ReadableConfig base, ReadableConfig onTop) {
+ if (base == null) {
+ base = new Configuration();
+ }
+ Configuration configuration = new Configuration();
+ for (ConfigOption<?> option : ForStConfigurableOptions.CANDIDATE_CONFIGS) {
+ Optional<?> baseValue = base.getOptional(option);
+ Optional<?> topValue = onTop.getOptional(option);
+
+ if (topValue.isPresent() || baseValue.isPresent()) {
+ Object validValue = topValue.isPresent() ? topValue.get() : baseValue.get();
+ ForStConfigurableOptions.checkArgumentValid(option, validValue);
+ configuration.setString(option.key(), validValue.toString());
+ }
+ }
+ return configuration;
+ }
+
+ @VisibleForTesting
+ ForStResourceContainer createOptionsAndResourceContainer(@Nullable File instanceBasePath) {
+ return createOptionsAndResourceContainer(null, instanceBasePath, false);
+ }
+
+ @VisibleForTesting
+ private ForStResourceContainer createOptionsAndResourceContainer(
+ @Nullable OpaqueMemoryResource<ForStSharedResources> sharedResources,
+ @Nullable File instanceBasePath,
+ boolean enableStatistics) {
+
+ return new ForStResourceContainer(
+ configurableOptions != null ? configurableOptions : new Configuration(),
+ forStOptionsFactory,
+ sharedResources,
+ instanceBasePath,
+ enableStatistics);
+ }
+
+ @Override
+ public String toString() {
+ return "ForStStateBackend{"
+ + ", localForStDirectories="
+ + Arrays.toString(localForStDirectories)
+ + '}';
+ }
+
+ // ------------------------------------------------------------------------
+ // static library loading utilities
+ // ------------------------------------------------------------------------
+
+ @VisibleForTesting
+ static void ensureForStIsLoaded(String tempDirectory) throws IOException {
+ ensureForStIsLoaded(tempDirectory, NativeLibraryLoader::getInstance);
+ }
+
+ @VisibleForTesting
+ static void setForStInitialized(boolean initialized) {
+ forStInitialized = initialized;
+ }
+
+ @VisibleForTesting
+ static void ensureForStIsLoaded(
+ String tempDirectory, Supplier<NativeLibraryLoader> nativeLibraryLoaderSupplier)
+ throws IOException {
+ synchronized (ForStStateBackend.class) {
+ if (!forStInitialized) {
+
+ final File tempDirParent = new File(tempDirectory).getAbsoluteFile();
+ LOG.info(
+ "Attempting to load ForSt native library and store it under '{}'",
+ tempDirParent);
+
+ Throwable lastException = null;
+ for (int attempt = 1; attempt <= FORST_LIB_LOADING_ATTEMPTS; attempt++) {
+ File rocksLibFolder = null;
+ try {
+ // when multiple instances of this class and ForSt exist in different
+ // class loaders, then we can see the following exception:
+ // "java.lang.UnsatisfiedLinkError: Native Library
+ // /path/to/temp/dir/librocksdbjni-linux64.so
+ // already loaded in another class loader"
+
+ // to avoid that, we need to add a random element to the library file path
+ // (I know, seems like an unnecessary hack, since the JVM obviously can
+ // handle multiple
+ // instances of the same JNI library being loaded in different class
+ // loaders, but
+ // apparently not when coming from the same file path, so there we go)
+
+ rocksLibFolder = new File(tempDirParent, "rocksdb-lib-" + new AbstractID());
+
+ // make sure the temp path exists
+ LOG.debug(
+ "Attempting to create ForSt native library folder {}",
+ rocksLibFolder);
+ // noinspection ResultOfMethodCallIgnored
+ rocksLibFolder.mkdirs();
+
+ // explicitly load the JNI dependency if it has not been loaded before
+ nativeLibraryLoaderSupplier
+ .get()
+ .loadLibrary(rocksLibFolder.getAbsolutePath());
+
+ // this initialization here should validate that the loading succeeded
+ RocksDB.loadLibrary();
+
+ // seems to have worked
+ LOG.info("Successfully loaded ForSt native library");
+ forStInitialized = true;
+ return;
+ } catch (Throwable t) {
+ lastException = t;
+ LOG.debug("ForSt JNI library loading attempt {} failed", attempt, t);
+
+ // try to force ForSt to attempt reloading the library
+ try {
+ resetForStLoadedFlag();
+ } catch (Throwable tt) {
+ LOG.debug(
+ "Failed to reset 'initialized' flag in ForSt native code loader",
+ tt);
+ }
+
+ FileUtils.deleteDirectoryQuietly(rocksLibFolder);
+ }
+ }
+
+ throw new IOException("Could not load the native ForSt library", lastException);
+ }
+ }
+ }
+
+ @VisibleForTesting
+ static void resetForStLoadedFlag() throws Exception {
+ final Field initField = NativeLibraryLoader.class.getDeclaredField("initialized");
+ initField.setAccessible(true);
+ initField.setBoolean(null, false);
+ }
+}
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateBackendFactory.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateBackendFactory.java
new file mode 100644
index 00000000000..4a9b5199f29
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateBackendFactory.java
@@ -0,0 +1,34 @@
+/*
+ * 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.flink.state.forst;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.configuration.IllegalConfigurationException;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.runtime.state.StateBackendFactory;
+
+/** A factory that creates an {@link ForStStateBackend} from a configuration. */
+@Experimental
+public class ForStStateBackendFactory implements StateBackendFactory<ForStStateBackend> {
+ @Override
+ public ForStStateBackend createFromConfig(ReadableConfig config, ClassLoader classLoader)
+ throws IllegalConfigurationException {
+ return new ForStStateBackend().configure(config, classLoader);
+ }
+}
diff --git a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStInitITCase.java b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStInitITCase.java
new file mode 100644
index 00000000000..37868ffa2ac
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStInitITCase.java
@@ -0,0 +1,64 @@
+/*
+ * 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.flink.state.forst;
+
+import org.apache.flink.runtime.operators.testutils.ExpectedTestException;
+
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.io.IOException;
+
+import static org.junit.Assert.fail;
+
+/** Tests for {@link ForStStateBackend} on initialization. */
+public class ForStInitITCase {
+
+ @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ /**
+ * This test checks that the ForSt native code loader still responds to resetting the init flag.
+ */
+ @Test
+ public void testResetInitFlag() throws Exception {
+ ForStStateBackend.resetForStLoadedFlag();
+ }
+
+ @Test
+ public void testTempLibFolderDeletedOnFail() throws Exception {
+ ForStStateBackend.setForStInitialized(false);
+ File tempFolder = temporaryFolder.newFolder();
+ try {
+ ForStStateBackend.ensureForStIsLoaded(
+ tempFolder.getAbsolutePath(),
+ () -> {
+ throw new ExpectedTestException();
+ });
+ fail("Not throwing expected exception.");
+ } catch (IOException ignored) {
+ // ignored
+ }
+ File[] files = tempFolder.listFiles();
+ Assert.assertNotNull(files);
+ Assert.assertEquals(0, files.length);
+ }
+}
diff --git a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStMultiClassLoaderTest.java b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStMultiClassLoaderTest.java
new file mode 100644
index 00000000000..9465cc0e3f3
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStMultiClassLoaderTest.java
@@ -0,0 +1,88 @@
+/*
+ * 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.flink.state.forst;
+
+import org.apache.flink.util.FlinkUserCodeClassLoaders;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.rocksdb.RocksDB;
+
+import java.lang.reflect.Method;
+import java.net.URL;
+
+import static org.apache.flink.util.FlinkUserCodeClassLoader.NOOP_EXCEPTION_HANDLER;
+import static org.junit.Assert.assertNotEquals;
+
+/**
+ * This test validates that the ForSt JNI library loading works properly in the presence of the
+ * ForSt code being loaded dynamically via reflection. That can happen when ForSt is in the user
+ * code JAR, or in certain test setups. TODO: test working with both ForSt and RocksDB
+ */
+public class ForStMultiClassLoaderTest {
+
+ @Rule public final TemporaryFolder tmp = new TemporaryFolder();
+
+ @Test
+ public void testTwoSeparateClassLoaders() throws Exception {
+ // collect the libraries / class folders with ForSt related code: the state backend and
+ // ForSt itself
+ final URL codePath1 =
+ ForStStateBackend.class.getProtectionDomain().getCodeSource().getLocation();
+ final URL codePath2 = RocksDB.class.getProtectionDomain().getCodeSource().getLocation();
+
+ final ClassLoader parent = getClass().getClassLoader();
+ final ClassLoader loader1 =
+ FlinkUserCodeClassLoaders.childFirst(
+ new URL[] {codePath1, codePath2},
+ parent,
+ new String[0],
+ NOOP_EXCEPTION_HANDLER,
+ true);
+ final ClassLoader loader2 =
+ FlinkUserCodeClassLoaders.childFirst(
+ new URL[] {codePath1, codePath2},
+ parent,
+ new String[0],
+ NOOP_EXCEPTION_HANDLER,
+ true);
+
+ final String className = ForStStateBackend.class.getName();
+
+ final Class<?> clazz1 = Class.forName(className, false, loader1);
+ final Class<?> clazz2 = Class.forName(className, false, loader2);
+ assertNotEquals(
+ "Test broken - the two reflectively loaded classes are equal", clazz1, clazz2);
+
+ final Object instance1 = clazz1.getConstructor().newInstance();
+ final Object instance2 = clazz2.getConstructor().newInstance();
+
+ final String tempDir = tmp.newFolder().getAbsolutePath();
+
+ final Method meth1 = clazz1.getDeclaredMethod("ensureForStIsLoaded", String.class);
+ final Method meth2 = clazz2.getDeclaredMethod("ensureForStIsLoaded", String.class);
+ meth1.setAccessible(true);
+ meth2.setAccessible(true);
+
+ // if all is well, these methods can both complete successfully
+ meth1.invoke(instance1, tempDir);
+ meth2.invoke(instance2, tempDir);
+ }
+}
diff --git a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateBackendConfigTest.java b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateBackendConfigTest.java
new file mode 100644
index 00000000000..11733f0ceca
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateBackendConfigTest.java
@@ -0,0 +1,767 @@
+/*
+ * 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.flink.state.forst;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.typeutils.base.IntSerializer;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.configuration.MemorySize;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.core.fs.CloseableRegistry;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
+import org.apache.flink.runtime.execution.Environment;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.operators.testutils.MockEnvironment;
+import org.apache.flink.runtime.query.KvStateRegistry;
+import org.apache.flink.runtime.query.TaskKvStateRegistry;
+import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.KeyedStateBackendParametersImpl;
+import org.apache.flink.runtime.state.ttl.TtlTimeProvider;
+import org.apache.flink.runtime.util.TestingTaskManagerRuntimeInfo;
+import org.apache.flink.testutils.junit.FailsInGHAContainerWithRootUser;
+import org.apache.flink.util.FileUtils;
+
+import org.apache.commons.lang3.RandomUtils;
+import org.junit.Assume;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.jupiter.api.Timeout;
+import org.junit.rules.TemporaryFolder;
+import org.rocksdb.BlockBasedTableConfig;
+import org.rocksdb.BloomFilter;
+import org.rocksdb.ColumnFamilyOptions;
+import org.rocksdb.CompactionStyle;
+import org.rocksdb.CompressionType;
+import org.rocksdb.DBOptions;
+import org.rocksdb.FlushOptions;
+import org.rocksdb.InfoLogLevel;
+import org.rocksdb.util.SizeUnit;
+
+import java.io.File;
+import java.nio.file.Files;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+
+import static org.apache.flink.state.forst.ForStTestUtils.createKeyedStateBackend;
+import static org.hamcrest.CoreMatchers.anyOf;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.startsWith;
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+/** Tests for configuring the ForSt State Backend. */
+public class ForStStateBackendConfigTest {
+
+ @Rule public final TemporaryFolder tempFolder = new TemporaryFolder();
+
+ // ------------------------------------------------------------------------
+ // default values
+ // ------------------------------------------------------------------------
+
+ @Test
+ public void testDefaultDbLogDir() throws Exception {
+ final ForStStateBackend backend = new ForStStateBackend();
+ final File logFile = File.createTempFile(getClass().getSimpleName() + "-", ".log");
+ // set the environment variable 'log.file' with the Flink log file location
+ System.setProperty("log.file", logFile.getPath());
+ try (ForStResourceContainer container = backend.createOptionsAndResourceContainer(null)) {
+ assertEquals(
+ ForStConfigurableOptions.LOG_LEVEL.defaultValue(),
+ container.getDbOptions().infoLogLevel());
+ assertEquals(logFile.getParent(), container.getDbOptions().dbLogDir());
+ } finally {
+ logFile.delete();
+ }
+
+ StringBuilder longInstanceBasePath =
+ new StringBuilder(tempFolder.newFolder().getAbsolutePath());
+ while (longInstanceBasePath.length() < 255) {
+ longInstanceBasePath.append("/append-for-long-path");
+ }
+ try (ForStResourceContainer container =
+ backend.createOptionsAndResourceContainer(
+ new File(longInstanceBasePath.toString()))) {
+ assertTrue(container.getDbOptions().dbLogDir().isEmpty());
+ } finally {
+ logFile.delete();
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // ForSt local file directory
+ // ------------------------------------------------------------------------
+
+ /** This test checks the behavior for basic setting of local DB directories. */
+ @Test
+ public void testSetDbPath() throws Exception {
+ final ForStStateBackend forStStateBackend = new ForStStateBackend();
+
+ final String testDir1 = tempFolder.newFolder().getAbsolutePath();
+ final String testDir2 = tempFolder.newFolder().getAbsolutePath();
+
+ assertNull(forStStateBackend.getLocalDbStoragePaths());
+
+ forStStateBackend.setLocalDbStoragePath(testDir1);
+ assertArrayEquals(new String[] {testDir1}, forStStateBackend.getLocalDbStoragePaths());
+
+ forStStateBackend.setLocalDbStoragePath(null);
+ assertNull(forStStateBackend.getLocalDbStoragePaths());
+
+ forStStateBackend.setLocalDbStoragePaths(testDir1, testDir2);
+ assertArrayEquals(
+ new String[] {testDir1, testDir2}, forStStateBackend.getLocalDbStoragePaths());
+
+ final MockEnvironment env = getMockEnvironment(tempFolder.newFolder());
+ final ForStKeyedStateBackend<Integer> keyedBackend =
+ createKeyedStateBackend(forStStateBackend, env, IntSerializer.INSTANCE);
+
+ try {
+ File instanceBasePath = keyedBackend.getInstanceBasePath();
+ assertThat(
+ instanceBasePath.getAbsolutePath(),
+ anyOf(startsWith(testDir1), startsWith(testDir2)));
+
+ //noinspection NullArgumentToVariableArgMethod
+ forStStateBackend.setLocalDbStoragePaths(null);
+ assertNull(forStStateBackend.getLocalDbStoragePaths());
+ } finally {
+ keyedBackend.dispose();
+ env.close();
+ }
+ }
+
+ @Test
+ public void testStoragePathWithFilePrefix() throws Exception {
+ final File folder = tempFolder.newFolder();
+ final String dbStoragePath = new Path(folder.toURI().toString()).toString();
+
+ assertTrue(dbStoragePath.startsWith("file:"));
+
+ testLocalDbPaths(dbStoragePath, folder);
+ }
+
+ @Test
+ public void testWithDefaultFsSchemeNoStoragePath() throws Exception {
+ try {
+ // set the default file system scheme
+ Configuration config = new Configuration();
+ config.set(CoreOptions.DEFAULT_FILESYSTEM_SCHEME, "file:///mydomain.com:8020/flink");
+ FileSystem.initialize(config);
+ testLocalDbPaths(null, tempFolder.getRoot());
+ } finally {
+ FileSystem.initialize(new Configuration());
+ }
+ }
+
+ @Test
+ public void testWithDefaultFsSchemeAbsoluteStoragePath() throws Exception {
+ final File folder = tempFolder.newFolder();
+ final String dbStoragePath = folder.getAbsolutePath();
+
+ try {
+ // set the default file system scheme
+ Configuration config = new Configuration();
+ config.set(CoreOptions.DEFAULT_FILESYSTEM_SCHEME, "file:///mydomain.com:8020/flink");
+ FileSystem.initialize(config);
+
+ testLocalDbPaths(dbStoragePath, folder);
+ } finally {
+ FileSystem.initialize(new Configuration());
+ }
+ }
+
+ private void testLocalDbPaths(String configuredPath, File expectedPath) throws Exception {
+ final ForStStateBackend forStBackend = new ForStStateBackend();
+ forStBackend.setLocalDbStoragePath(configuredPath);
+
+ final MockEnvironment env = getMockEnvironment(tempFolder.newFolder());
+ ForStKeyedStateBackend<Integer> keyedBackend =
+ createKeyedStateBackend(forStBackend, env, IntSerializer.INSTANCE);
+
+ try {
+ File instanceBasePath = keyedBackend.getInstanceBasePath();
+ assertThat(
+ instanceBasePath.getAbsolutePath(), startsWith(expectedPath.getAbsolutePath()));
+
+ //noinspection NullArgumentToVariableArgMethod
+ forStBackend.setLocalDbStoragePaths(null);
+ assertNull(forStBackend.getLocalDbStoragePaths());
+ } finally {
+ keyedBackend.dispose();
+ env.close();
+ }
+ }
+
+ /** Validates that empty arguments for the local DB path are invalid. */
+ @Test(expected = IllegalArgumentException.class)
+ public void testSetEmptyPaths() throws Exception {
+ ForStStateBackend forStStateBackend = new ForStStateBackend();
+ forStStateBackend.setLocalDbStoragePaths();
+ }
+
+ /** Validates that schemes other than 'file:/' are not allowed. */
+ @Test(expected = IllegalArgumentException.class)
+ public void testNonFileSchemePath() throws Exception {
+ ForStStateBackend forStStateBackend = new ForStStateBackend();
+ forStStateBackend.setLocalDbStoragePath("hdfs:///some/path/to/perdition");
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testDbPathRelativePaths() throws Exception {
+ ForStStateBackend forStStateBackend = new ForStStateBackend();
+ forStStateBackend.setLocalDbStoragePath("relative/path");
+ }
+
+ @Test
+ @Timeout(value = 60)
+ public void testCleanRelocatedDbLogs() throws Exception {
+ final File folder = tempFolder.newFolder();
+ final File relocatedDBLogDir = tempFolder.newFolder("db_logs");
+ final File logFile = new File(relocatedDBLogDir, "taskManager.log");
+ Files.createFile(logFile.toPath());
+ System.setProperty("log.file", logFile.getAbsolutePath());
+
+ Configuration conf = new Configuration();
+ conf.set(ForStConfigurableOptions.LOG_LEVEL, InfoLogLevel.DEBUG_LEVEL);
+ conf.set(ForStConfigurableOptions.LOG_FILE_NUM, 4);
+ conf.set(ForStConfigurableOptions.LOG_MAX_FILE_SIZE, MemorySize.parse("1kb"));
+ final ForStStateBackend rocksDbBackend =
+ new ForStStateBackend().configure(conf, getClass().getClassLoader());
+ final String dbStoragePath = new Path(folder.toURI().toString()).toString();
+ rocksDbBackend.setLocalDbStoragePath(dbStoragePath);
+
+ final MockEnvironment env = getMockEnvironment(tempFolder.newFolder());
+ ForStKeyedStateBackend<Integer> keyedBackend =
+ createKeyedStateBackend(rocksDbBackend, env, IntSerializer.INSTANCE);
+
+ File instanceBasePath = keyedBackend.getInstanceBasePath();
+ File instanceRocksDBPath = new File(instanceBasePath, "db");
+
+ // avoid tests without relocate.
+ Assume.assumeTrue(instanceRocksDBPath.getAbsolutePath().length() <= 255 - "_LOG".length());
+
+ java.nio.file.Path[] relocatedDbLogs;
+ try {
+ relocatedDbLogs = FileUtils.listDirectory(relocatedDBLogDir.toPath());
+ while (relocatedDbLogs.length <= 2) {
+ // If the default number of log files in rocksdb is not enough, add more logs.
+ try (FlushOptions flushOptions = new FlushOptions()) {
+ keyedBackend.db.put(RandomUtils.nextBytes(32), RandomUtils.nextBytes(512));
+ keyedBackend.db.flush(flushOptions);
+ }
+ relocatedDbLogs = FileUtils.listDirectory(relocatedDBLogDir.toPath());
+ }
+ } finally {
+ keyedBackend.dispose();
+ env.close();
+ }
+
+ relocatedDbLogs = FileUtils.listDirectory(relocatedDBLogDir.toPath());
+ assertEquals(1, relocatedDbLogs.length);
+ assertEquals("taskManager.log", relocatedDbLogs[0].toFile().getName());
+ }
+
+ // ------------------------------------------------------------------------
+ // RocksDB local file automatic from temp directories
+ // ------------------------------------------------------------------------
+
+ /**
+ * This tests whether the RocksDB backends uses the temp directories that are provided from the
+ * {@link Environment} when no db storage path is set.
+ */
+ @Test
+ public void testUseTempDirectories() throws Exception {
+ ForStStateBackend forStStateBackend = new ForStStateBackend();
+
+ File dir1 = tempFolder.newFolder();
+
+ assertNull(forStStateBackend.getLocalDbStoragePaths());
+
+ final MockEnvironment env = getMockEnvironment(dir1);
+ JobID jobID = env.getJobID();
+ KeyGroupRange keyGroupRange = new KeyGroupRange(0, 0);
+ TaskKvStateRegistry kvStateRegistry = env.getTaskKvStateRegistry();
+ CloseableRegistry cancelStreamRegistry = new CloseableRegistry();
+ ForStKeyedStateBackend<Integer> keyedBackend =
+ forStStateBackend.createForStKeyedStateBackend(
+ new KeyedStateBackendParametersImpl<>(
+ env,
+ jobID,
+ "test_op",
+ IntSerializer.INSTANCE,
+ 1,
+ keyGroupRange,
+ kvStateRegistry,
+ TtlTimeProvider.DEFAULT,
+ new UnregisteredMetricsGroup(),
+ Collections.emptyList(),
+ cancelStreamRegistry));
+
+ try {
+ File instanceBasePath = keyedBackend.getInstanceBasePath();
+ assertThat(instanceBasePath.getAbsolutePath(), startsWith(dir1.getAbsolutePath()));
+ } finally {
+ keyedBackend.dispose();
+ env.close();
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // RocksDB local file directory initialization
+ // ------------------------------------------------------------------------
+
+ @Test
+ @Category(FailsInGHAContainerWithRootUser.class)
+ public void testFailWhenNoLocalStorageDir() throws Exception {
+ final File targetDir = tempFolder.newFolder();
+ Assume.assumeTrue(
+ "Cannot mark directory non-writable", targetDir.setWritable(false, false));
+
+ ForStStateBackend forStStateBackend = new ForStStateBackend();
+
+ try (MockEnvironment env = getMockEnvironment(tempFolder.newFolder())) {
+ forStStateBackend.setLocalDbStoragePath(targetDir.getAbsolutePath());
+
+ boolean hasFailure = false;
+ try {
+ JobID jobID = env.getJobID();
+ KeyGroupRange keyGroupRange = new KeyGroupRange(0, 0);
+ TaskKvStateRegistry kvStateRegistry =
+ new KvStateRegistry().createTaskRegistry(env.getJobID(), new JobVertexID());
+ CloseableRegistry cancelStreamRegistry = new CloseableRegistry();
+ forStStateBackend.createForStKeyedStateBackend(
+ new KeyedStateBackendParametersImpl<>(
+ env,
+ jobID,
+ "foobar",
+ IntSerializer.INSTANCE,
+ 1,
+ keyGroupRange,
+ kvStateRegistry,
+ TtlTimeProvider.DEFAULT,
+ new UnregisteredMetricsGroup(),
+ Collections.emptyList(),
+ cancelStreamRegistry));
+ } catch (Exception e) {
+ assertTrue(e.getMessage().contains("No local storage directories available"));
+ assertTrue(e.getMessage().contains(targetDir.getAbsolutePath()));
+ hasFailure = true;
+ }
+ assertTrue(
+ "We must see a failure because no storaged directory is feasible.", hasFailure);
+ } finally {
+ //noinspection ResultOfMethodCallIgnored
+ targetDir.setWritable(true, false);
+ }
+ }
+
+ @Test
+ public void testContinueOnSomeDbDirectoriesMissing() throws Exception {
+ final File targetDir1 = tempFolder.newFolder();
+ final File targetDir2 = tempFolder.newFolder();
+ Assume.assumeTrue(
+ "Cannot mark directory non-writable", targetDir1.setWritable(false, false));
+
+ ForStStateBackend forStStateBackend = new ForStStateBackend();
+
+ try (MockEnvironment env = getMockEnvironment(tempFolder.newFolder())) {
+ forStStateBackend.setLocalDbStoragePaths(
+ targetDir1.getAbsolutePath(), targetDir2.getAbsolutePath());
+
+ try {
+ JobID jobID = env.getJobID();
+ KeyGroupRange keyGroupRange = new KeyGroupRange(0, 0);
+ TaskKvStateRegistry kvStateRegistry =
+ new KvStateRegistry().createTaskRegistry(env.getJobID(), new JobVertexID());
+ CloseableRegistry cancelStreamRegistry = new CloseableRegistry();
+ ForStKeyedStateBackend<Integer> keyedStateBackend =
+ forStStateBackend.createForStKeyedStateBackend(
+ new KeyedStateBackendParametersImpl<>(
+ env,
+ jobID,
+ "foobar",
+ IntSerializer.INSTANCE,
+ 1,
+ keyGroupRange,
+ kvStateRegistry,
+ TtlTimeProvider.DEFAULT,
+ new UnregisteredMetricsGroup(),
+ Collections.emptyList(),
+ cancelStreamRegistry));
+
+ keyedStateBackend.dispose();
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail("Backend initialization failed even though some paths were available");
+ }
+ } finally {
+ //noinspection ResultOfMethodCallIgnored
+ targetDir1.setWritable(true, false);
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // RocksDB Options
+ // ------------------------------------------------------------------------
+ @Test
+ public void testConfigurableOptionsFromConfig() throws Exception {
+ Configuration configuration = new Configuration();
+
+ // verify illegal configuration
+ {
+ verifyIllegalArgument(ForStConfigurableOptions.MAX_BACKGROUND_THREADS, "-1");
+ verifyIllegalArgument(ForStConfigurableOptions.LOG_LEVEL, "DEBUG");
+ verifyIllegalArgument(ForStConfigurableOptions.LOG_DIR, "tmp/rocksdb-logs/");
+ verifyIllegalArgument(ForStConfigurableOptions.LOG_DIR, "");
+ verifyIllegalArgument(ForStConfigurableOptions.LOG_FILE_NUM, "0");
+ verifyIllegalArgument(ForStConfigurableOptions.LOG_FILE_NUM, "-1");
+ verifyIllegalArgument(ForStConfigurableOptions.LOG_MAX_FILE_SIZE, "-1KB");
+ verifyIllegalArgument(ForStConfigurableOptions.MAX_WRITE_BUFFER_NUMBER, "-1");
+ verifyIllegalArgument(ForStConfigurableOptions.MIN_WRITE_BUFFER_NUMBER_TO_MERGE, "-1");
+
+ verifyIllegalArgument(ForStConfigurableOptions.TARGET_FILE_SIZE_BASE, "0KB");
+ verifyIllegalArgument(ForStConfigurableOptions.MAX_SIZE_LEVEL_BASE, "1BB");
+ verifyIllegalArgument(ForStConfigurableOptions.WRITE_BUFFER_SIZE, "-1KB");
+ verifyIllegalArgument(ForStConfigurableOptions.BLOCK_SIZE, "0MB");
+ verifyIllegalArgument(ForStConfigurableOptions.METADATA_BLOCK_SIZE, "0MB");
+ verifyIllegalArgument(ForStConfigurableOptions.BLOCK_CACHE_SIZE, "0");
+
+ verifyIllegalArgument(ForStConfigurableOptions.USE_DYNAMIC_LEVEL_SIZE, "1");
+
+ verifyIllegalArgument(ForStConfigurableOptions.COMPACTION_STYLE, "LEV");
+ verifyIllegalArgument(ForStConfigurableOptions.COMPRESSION_PER_LEVEL, "SNAP");
+ verifyIllegalArgument(ForStConfigurableOptions.USE_BLOOM_FILTER, "NO");
+ verifyIllegalArgument(ForStConfigurableOptions.BLOOM_FILTER_BLOCK_BASED_MODE, "YES");
+ }
+
+ // verify legal configuration
+ {
+ configuration.setString(ForStConfigurableOptions.LOG_LEVEL.key(), "DEBUG_LEVEL");
+ configuration.setString(ForStConfigurableOptions.LOG_DIR.key(), "/tmp/rocksdb-logs/");
+ configuration.setString(ForStConfigurableOptions.LOG_FILE_NUM.key(), "10");
+ configuration.setString(ForStConfigurableOptions.LOG_MAX_FILE_SIZE.key(), "2MB");
+ configuration.setString(ForStConfigurableOptions.COMPACTION_STYLE.key(), "level");
+ configuration.setString(
+ ForStConfigurableOptions.COMPRESSION_PER_LEVEL.key(),
+ "no_compression;snappy_compression;lz4_compression");
+ configuration.setString(ForStConfigurableOptions.USE_DYNAMIC_LEVEL_SIZE.key(), "TRUE");
+ configuration.setString(ForStConfigurableOptions.TARGET_FILE_SIZE_BASE.key(), "8 mb");
+ configuration.setString(ForStConfigurableOptions.MAX_SIZE_LEVEL_BASE.key(), "128MB");
+ configuration.setString(ForStConfigurableOptions.MAX_BACKGROUND_THREADS.key(), "4");
+ configuration.setString(ForStConfigurableOptions.MAX_WRITE_BUFFER_NUMBER.key(), "4");
+ configuration.setString(
+ ForStConfigurableOptions.MIN_WRITE_BUFFER_NUMBER_TO_MERGE.key(), "2");
+ configuration.setString(ForStConfigurableOptions.WRITE_BUFFER_SIZE.key(), "64 MB");
+ configuration.setString(ForStConfigurableOptions.BLOCK_SIZE.key(), "4 kb");
+ configuration.setString(ForStConfigurableOptions.METADATA_BLOCK_SIZE.key(), "8 kb");
+ configuration.setString(ForStConfigurableOptions.BLOCK_CACHE_SIZE.key(), "512 mb");
+ configuration.setString(ForStConfigurableOptions.USE_BLOOM_FILTER.key(), "TRUE");
+
+ try (ForStResourceContainer optionsContainer =
+ new ForStResourceContainer(configuration, null, null, null, false)) {
+
+ DBOptions dbOptions = optionsContainer.getDbOptions();
+ assertEquals(-1, dbOptions.maxOpenFiles());
+ assertEquals(InfoLogLevel.DEBUG_LEVEL, dbOptions.infoLogLevel());
+ assertEquals("/tmp/rocksdb-logs/", dbOptions.dbLogDir());
+ assertEquals(10, dbOptions.keepLogFileNum());
+ assertEquals(2 * SizeUnit.MB, dbOptions.maxLogFileSize());
+
+ ColumnFamilyOptions columnOptions = optionsContainer.getColumnOptions();
+ assertEquals(CompactionStyle.LEVEL, columnOptions.compactionStyle());
+ assertTrue(columnOptions.levelCompactionDynamicLevelBytes());
+ assertEquals(8 * SizeUnit.MB, columnOptions.targetFileSizeBase());
+ assertEquals(128 * SizeUnit.MB, columnOptions.maxBytesForLevelBase());
+ assertEquals(4, columnOptions.maxWriteBufferNumber());
+ assertEquals(2, columnOptions.minWriteBufferNumberToMerge());
+ assertEquals(64 * SizeUnit.MB, columnOptions.writeBufferSize());
+ assertEquals(
+ Arrays.asList(
+ CompressionType.NO_COMPRESSION,
+ CompressionType.SNAPPY_COMPRESSION,
+ CompressionType.LZ4_COMPRESSION),
+ columnOptions.compressionPerLevel());
+
+ BlockBasedTableConfig tableConfig =
+ (BlockBasedTableConfig) columnOptions.tableFormatConfig();
+ assertEquals(4 * SizeUnit.KB, tableConfig.blockSize());
+ assertEquals(8 * SizeUnit.KB, tableConfig.metadataBlockSize());
+ assertEquals(512 * SizeUnit.MB, tableConfig.blockCacheSize());
+ assertTrue(tableConfig.filterPolicy() instanceof BloomFilter);
+ }
+ }
+ }
+
+ @Test
+ public void testOptionsFactory() throws Exception {
+ ForStStateBackend forStStateBackend = new ForStStateBackend();
+
+ // verify that user-defined options factory could be configured via config.yaml
+ Configuration config = new Configuration();
+ config.setString(ForStOptions.OPTIONS_FACTORY.key(), TestOptionsFactory.class.getName());
+ config.setString(TestOptionsFactory.BACKGROUND_JOBS_OPTION.key(), "4");
+
+ forStStateBackend = forStStateBackend.configure(config, getClass().getClassLoader());
+
+ assertTrue(forStStateBackend.getForStOptions() instanceof TestOptionsFactory);
+
+ try (ForStResourceContainer optionsContainer =
+ forStStateBackend.createOptionsAndResourceContainer(null)) {
+ DBOptions dbOptions = optionsContainer.getDbOptions();
+ assertEquals(4, dbOptions.maxBackgroundJobs());
+ }
+
+ // verify that user-defined options factory could be set programmatically and override
+ // pre-configured one.
+ forStStateBackend.setForStOptions(
+ new ForStOptionsFactory() {
+ @Override
+ public DBOptions createDBOptions(
+ DBOptions currentOptions, Collection<AutoCloseable> handlesToClose) {
+ return currentOptions;
+ }
+
+ @Override
+ public ColumnFamilyOptions createColumnOptions(
+ ColumnFamilyOptions currentOptions,
+ Collection<AutoCloseable> handlesToClose) {
+ return currentOptions.setCompactionStyle(CompactionStyle.FIFO);
+ }
+ });
+
+ try (ForStResourceContainer optionsContainer =
+ forStStateBackend.createOptionsAndResourceContainer(null)) {
+ ColumnFamilyOptions colCreated = optionsContainer.getColumnOptions();
+ assertEquals(CompactionStyle.FIFO, colCreated.compactionStyle());
+ }
+ }
+
+ @Test
+ public void testConfigurableOptions() throws Exception {
+ Configuration configuration = new Configuration();
+ configuration.set(ForStConfigurableOptions.COMPACTION_STYLE, CompactionStyle.UNIVERSAL);
+ try (final ForStResourceContainer optionsContainer =
+ new ForStResourceContainer(configuration, null, null, null, false)) {
+
+ final ColumnFamilyOptions columnFamilyOptions = optionsContainer.getColumnOptions();
+ assertNotNull(columnFamilyOptions);
+ assertEquals(CompactionStyle.UNIVERSAL, columnFamilyOptions.compactionStyle());
+ }
+
+ try (final ForStResourceContainer optionsContainer =
+ new ForStResourceContainer(new Configuration(), null, null, null, false)) {
+
+ final ColumnFamilyOptions columnFamilyOptions = optionsContainer.getColumnOptions();
+ assertNotNull(columnFamilyOptions);
+ assertEquals(CompactionStyle.LEVEL, columnFamilyOptions.compactionStyle());
+ }
+ }
+
+ @Test
+ public void testPredefinedAndOptionsFactory() throws Exception {
+ final ForStOptionsFactory optionsFactory =
+ new ForStOptionsFactory() {
+ @Override
+ public DBOptions createDBOptions(
+ DBOptions currentOptions, Collection<AutoCloseable> handlesToClose) {
+ return currentOptions;
+ }
+
+ @Override
+ public ColumnFamilyOptions createColumnOptions(
+ ColumnFamilyOptions currentOptions,
+ Collection<AutoCloseable> handlesToClose) {
+ return currentOptions.setCompactionStyle(CompactionStyle.UNIVERSAL);
+ }
+ };
+
+ try (final ForStResourceContainer optionsContainer =
+ new ForStResourceContainer(optionsFactory)) {
+
+ final ColumnFamilyOptions columnFamilyOptions = optionsContainer.getColumnOptions();
+ assertNotNull(columnFamilyOptions);
+ assertEquals(CompactionStyle.UNIVERSAL, columnFamilyOptions.compactionStyle());
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // Reconfiguration
+ // ------------------------------------------------------------------------
+
+ @Test
+ public void testForStReconfigurationCopiesExistingValues() throws Exception {
+ final ForStStateBackend original = new ForStStateBackend();
+ final ForStOptionsFactory optionsFactory = new TestOptionsFactory();
+ original.setForStOptions(optionsFactory);
+
+ final String[] localDirs =
+ new String[] {
+ tempFolder.newFolder().getAbsolutePath(),
+ tempFolder.newFolder().getAbsolutePath()
+ };
+ original.setLocalDbStoragePaths(localDirs);
+
+ ForStStateBackend copy =
+ original.configure(
+ new Configuration(), Thread.currentThread().getContextClassLoader());
+
+ assertArrayEquals(original.getLocalDbStoragePaths(), copy.getLocalDbStoragePaths());
+ assertEquals(original.getForStOptions(), copy.getForStOptions());
+ }
+
+ // ------------------------------------------------------------------------
+ // RocksDB Memory Control
+ // ------------------------------------------------------------------------
+
+ @Test
+ public void testDefaultMemoryControlParameters() {
+ ForStMemoryConfiguration memSettings = new ForStMemoryConfiguration();
+ assertTrue(memSettings.isUsingManagedMemory());
+ assertFalse(memSettings.isUsingFixedMemoryPerSlot());
+ assertEquals(
+ ForStOptions.HIGH_PRIORITY_POOL_RATIO.defaultValue(),
+ memSettings.getHighPriorityPoolRatio(),
+ 0.0);
+ assertEquals(
+ ForStOptions.WRITE_BUFFER_RATIO.defaultValue(),
+ memSettings.getWriteBufferRatio(),
+ 0.0);
+
+ ForStMemoryConfiguration configured =
+ ForStMemoryConfiguration.fromOtherAndConfiguration(
+ memSettings, new Configuration());
+ assertTrue(configured.isUsingManagedMemory());
+ assertFalse(configured.isUsingFixedMemoryPerSlot());
+ assertEquals(
+ ForStOptions.HIGH_PRIORITY_POOL_RATIO.defaultValue(),
+ configured.getHighPriorityPoolRatio(),
+ 0.0);
+ assertEquals(
+ ForStOptions.WRITE_BUFFER_RATIO.defaultValue(),
+ configured.getWriteBufferRatio(),
+ 0.0);
+ }
+
+ @Test
+ public void testConfigureManagedMemory() {
+ final Configuration config = new Configuration();
+ config.set(ForStOptions.USE_MANAGED_MEMORY, true);
+
+ final ForStMemoryConfiguration memSettings =
+ ForStMemoryConfiguration.fromOtherAndConfiguration(
+ new ForStMemoryConfiguration(), config);
+
+ assertTrue(memSettings.isUsingManagedMemory());
+ }
+
+ @Test
+ public void testConfigureIllegalMemoryControlParameters() {
+ ForStMemoryConfiguration memSettings = new ForStMemoryConfiguration();
+
+ verifySetParameter(() -> memSettings.setFixedMemoryPerSlot("-1B"));
+ verifySetParameter(() -> memSettings.setHighPriorityPoolRatio(-0.1));
+ verifySetParameter(() -> memSettings.setHighPriorityPoolRatio(1.1));
+ verifySetParameter(() -> memSettings.setWriteBufferRatio(-0.1));
+ verifySetParameter(() -> memSettings.setWriteBufferRatio(1.1));
+
+ memSettings.setFixedMemoryPerSlot("128MB");
+ memSettings.setWriteBufferRatio(0.6);
+ memSettings.setHighPriorityPoolRatio(0.6);
+
+ try {
+ // sum of writeBufferRatio and highPriPoolRatio larger than 1.0
+ memSettings.validate();
+ fail("Expected an IllegalArgumentException.");
+ } catch (IllegalArgumentException expected) {
+ // expected exception
+ }
+ }
+
+ private void verifySetParameter(Runnable setter) {
+ try {
+ setter.run();
+ fail("No expected IllegalArgumentException.");
+ } catch (IllegalArgumentException expected) {
+ // expected exception
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // Utilities
+ // ------------------------------------------------------------------------
+
+ static MockEnvironment getMockEnvironment(File tempDir) {
+ return MockEnvironment.builder()
+ .setUserCodeClassLoader(ForStStateBackendConfigTest.class.getClassLoader())
+ .setTaskManagerRuntimeInfo(
+ new TestingTaskManagerRuntimeInfo(new Configuration(), tempDir))
+ .build();
+ }
+
+ private void verifyIllegalArgument(ConfigOption<?> configOption, String configValue) {
+ Configuration configuration = new Configuration();
+ configuration.setString(configOption.key(), configValue);
+
+ ForStStateBackend stateBackend = new ForStStateBackend();
+ try {
+ stateBackend.configure(configuration, null);
+ fail("Not throwing expected IllegalArgumentException.");
+ } catch (IllegalArgumentException e) {
+ // ignored
+ }
+ }
+
+ /** An implementation of options factory for testing. */
+ public static class TestOptionsFactory implements ConfigurableForStOptionsFactory {
+ public static final ConfigOption<Integer> BACKGROUND_JOBS_OPTION =
+ ConfigOptions.key("my.custom.rocksdb.backgroundJobs").intType().defaultValue(2);
+
+ private int backgroundJobs = BACKGROUND_JOBS_OPTION.defaultValue();
+
+ @Override
+ public DBOptions createDBOptions(
+ DBOptions currentOptions, Collection<AutoCloseable> handlesToClose) {
+ return currentOptions.setMaxBackgroundJobs(backgroundJobs);
+ }
+
+ @Override
+ public ColumnFamilyOptions createColumnOptions(
+ ColumnFamilyOptions currentOptions, Collection<AutoCloseable> handlesToClose) {
+ return currentOptions.setCompactionStyle(CompactionStyle.UNIVERSAL);
+ }
+
+ @Override
+ public ForStOptionsFactory configure(ReadableConfig configuration) {
+ this.backgroundJobs = configuration.get(BACKGROUND_JOBS_OPTION);
+ return this;
+ }
+ }
+}
diff --git a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateBackendFactoryTest.java b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateBackendFactoryTest.java
new file mode 100644
index 00000000000..effcd2be793
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateBackendFactoryTest.java
@@ -0,0 +1,132 @@
+/*
+ * 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.flink.state.forst;
+
+import org.apache.flink.configuration.CheckpointingOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.StateBackendOptions;
+import org.apache.flink.runtime.state.StateBackend;
+import org.apache.flink.runtime.state.StateBackendLoader;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.util.Arrays;
+import java.util.HashSet;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+/** Tests for the ForStStateBackendFactory. */
+public class ForStStateBackendFactoryTest {
+
+ @Rule public final TemporaryFolder tmp = new TemporaryFolder();
+
+ private final ClassLoader cl = getClass().getClassLoader();
+
+ private final String backendKey = StateBackendOptions.STATE_BACKEND.key();
+
+ // ------------------------------------------------------------------------
+
+ @Test
+ public void testEmbeddedFactoryName() {
+ // construct the name such that it will not be automatically adjusted on refactorings
+ String factoryName = "org.apache.flink.state.forst.For";
+ factoryName += "StStateBackendFactory";
+
+ // !!! if this fails, the code in StateBackendLoader must be adjusted
+ assertEquals(factoryName, ForStStateBackendFactory.class.getName());
+ }
+
+ /**
+ * Validates loading a file system state backend with additional parameters from the cluster
+ * configuration.
+ */
+ @Test
+ public void testLoadForStStateBackend() throws Exception {
+ final String localDir1 = tmp.newFolder().getAbsolutePath();
+ final String localDir2 = tmp.newFolder().getAbsolutePath();
+ final String localDirs = localDir1 + File.pathSeparator + localDir2;
+ final boolean incremental = !CheckpointingOptions.INCREMENTAL_CHECKPOINTS.defaultValue();
+
+ // TODO: Support short name of backendKey
+
+ final Configuration config1 = new Configuration();
+ config1.setString(backendKey, ForStStateBackendFactory.class.getName());
+ config1.set(ForStOptions.LOCAL_DIRECTORIES, localDirs);
+ config1.set(CheckpointingOptions.INCREMENTAL_CHECKPOINTS, incremental);
+
+ StateBackend backend1 = StateBackendLoader.loadStateBackendFromConfig(config1, cl, null);
+
+ assertTrue(backend1 instanceof ForStStateBackend);
+
+ ForStStateBackend fs1 = (ForStStateBackend) backend1;
+
+ checkPaths(fs1.getLocalDbStoragePaths(), localDir1, localDir2);
+ }
+
+ /**
+ * Validates taking the application-defined ForSt state backend and adding with additional
+ * parameters from the cluster configuration, but giving precedence to application-defined
+ * parameters over configuration-defined parameters.
+ */
+ @Test
+ public void testLoadForStStateBackendMixed() throws Exception {
+ final String localDir1 = tmp.newFolder().getAbsolutePath();
+ final String localDir2 = tmp.newFolder().getAbsolutePath();
+ final String localDir3 = tmp.newFolder().getAbsolutePath();
+ final String localDir4 = tmp.newFolder().getAbsolutePath();
+
+ final ForStStateBackend backend = new ForStStateBackend();
+ backend.setLocalDbStoragePaths(localDir1, localDir2);
+
+ final Configuration config = new Configuration();
+ config.setString(backendKey, "hashmap"); // this should not be picked up
+ config.set(
+ ForStOptions.LOCAL_DIRECTORIES,
+ localDir3 + ":" + localDir4); // this should not be picked up
+
+ final StateBackend loadedBackend =
+ StateBackendLoader.fromApplicationOrConfigOrDefault(
+ backend, new Configuration(), config, cl, null);
+ assertTrue(loadedBackend instanceof ForStStateBackend);
+
+ final ForStStateBackend loadedRocks = (ForStStateBackend) loadedBackend;
+
+ checkPaths(loadedRocks.getLocalDbStoragePaths(), localDir1, localDir2);
+ }
+
+ // ------------------------------------------------------------------------
+
+ private static void checkPaths(String[] pathsArray, String... paths) {
+ assertNotNull(pathsArray);
+ assertNotNull(paths);
+
+ assertEquals(pathsArray.length, paths.length);
+
+ HashSet<String> pathsSet = new HashSet<>(Arrays.asList(pathsArray));
+
+ for (String path : paths) {
+ assertTrue(pathsSet.contains(path));
+ }
+ }
+}
diff --git a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStTestUtils.java b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStTestUtils.java
new file mode 100644
index 00000000000..e5cb4c91872
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStTestUtils.java
@@ -0,0 +1,54 @@
+/*
+ * 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.flink.state.forst;
+
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.fs.CloseableRegistry;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
+import org.apache.flink.runtime.execution.Environment;
+import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.KeyedStateBackendParametersImpl;
+import org.apache.flink.runtime.state.ttl.TtlTimeProvider;
+
+import java.io.IOException;
+import java.util.Collections;
+
+/** Test utils for the ForSt state backend. */
+public final class ForStTestUtils {
+ public static <K> ForStKeyedStateBackend<K> createKeyedStateBackend(
+ ForStStateBackend forStStateBackend, Environment env, TypeSerializer<K> keySerializer)
+ throws IOException {
+
+ return forStStateBackend.createForStKeyedStateBackend(
+ new KeyedStateBackendParametersImpl<>(
+ env,
+ env.getJobID(),
+ "test_op",
+ keySerializer,
+ 1,
+ new KeyGroupRange(0, 0),
+ env.getTaskKvStateRegistry(),
+ TtlTimeProvider.DEFAULT,
+ new UnregisteredMetricsGroup(),
+ (name, value) -> {},
+ Collections.emptyList(),
+ new CloseableRegistry(),
+ 1.0));
+ }
+}
(flink) 02/07: [FLINK-35047][state] Introduce ForSt related options
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hangxiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 92d7f6538607a0189f9f7f2bff0470f8c8cad408
Author: Hangxiang Yu <ma...@gmail.com>
AuthorDate: Mon Apr 15 21:05:07 2024 +0800
[FLINK-35047][state] Introduce ForSt related options
---
.../forst/ConfigurableForStOptionsFactory.java | 39 +++
.../state/forst/ForStConfigurableOptions.java | 377 +++++++++++++++++++++
.../org/apache/flink/state/forst/ForStOptions.java | 123 +++++++
.../flink/state/forst/ForStOptionsFactory.java | 106 ++++++
4 files changed, 645 insertions(+)
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ConfigurableForStOptionsFactory.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ConfigurableForStOptionsFactory.java
new file mode 100644
index 00000000000..181d60ba603
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ConfigurableForStOptionsFactory.java
@@ -0,0 +1,39 @@
+/*
+ * 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.flink.state.forst;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.configuration.ReadableConfig;
+
+/** An interface for options factory that pick up additional parameters from a configuration. */
+@Experimental
+public interface ConfigurableForStOptionsFactory extends ForStOptionsFactory {
+
+ /**
+ * Creates a variant of the options factory that applies additional configuration parameters.
+ *
+ * <p>If no configuration is applied, or if the method directly applies configuration values to
+ * the (mutable) options factory object, this method may return the original options factory
+ * object. Otherwise it typically returns a modified copy.
+ *
+ * @param configuration The configuration to pick the values from.
+ * @return A reconfigured options factory.
+ */
+ ForStOptionsFactory configure(ReadableConfig configuration);
+}
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStConfigurableOptions.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStConfigurableOptions.java
new file mode 100644
index 00000000000..d811199e5e0
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStConfigurableOptions.java
@@ -0,0 +1,377 @@
+/*
+ * 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.flink.state.forst;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.MemorySize;
+import org.apache.flink.configuration.description.Description;
+import org.apache.flink.util.Preconditions;
+
+import org.rocksdb.CompactionStyle;
+import org.rocksdb.CompressionType;
+import org.rocksdb.InfoLogLevel;
+
+import java.io.File;
+import java.io.Serializable;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import static org.apache.flink.configuration.ConfigOptions.key;
+import static org.apache.flink.configuration.description.LinkElement.link;
+import static org.apache.flink.configuration.description.TextElement.code;
+import static org.rocksdb.CompactionStyle.FIFO;
+import static org.rocksdb.CompactionStyle.LEVEL;
+import static org.rocksdb.CompactionStyle.NONE;
+import static org.rocksdb.CompactionStyle.UNIVERSAL;
+import static org.rocksdb.CompressionType.LZ4_COMPRESSION;
+import static org.rocksdb.CompressionType.NO_COMPRESSION;
+import static org.rocksdb.CompressionType.SNAPPY_COMPRESSION;
+import static org.rocksdb.InfoLogLevel.INFO_LEVEL;
+
+/**
+ * This class contains the configuration options for the ForStStateBackend.
+ *
+ * <p>Currently, Options of ForSt would be configured by values here, and a user-defined {@link
+ * ForStOptionsFactory} may override the configurations here.
+ */
+@Experimental
+public class ForStConfigurableOptions implements Serializable {
+
+ // --------------------------------------------------------------------------
+ // Provided configurable DBOptions within Flink
+ // --------------------------------------------------------------------------
+
+ public static final ConfigOption<Integer> MAX_BACKGROUND_THREADS =
+ key("state.backend.forst.thread.num")
+ .intType()
+ .defaultValue(2)
+ .withDescription(
+ "The maximum number of concurrent background flush and compaction jobs (per stateful operator). "
+ + "The default value is '2'.");
+
+ public static final ConfigOption<Integer> MAX_OPEN_FILES =
+ key("state.backend.forst.files.open")
+ .intType()
+ .defaultValue(-1)
+ .withDescription(
+ "The maximum number of open files (per stateful operator) that can be used by the DB, '-1' means no limit. "
+ + "The default value is '-1'.");
+
+ public static final ConfigOption<MemorySize> LOG_MAX_FILE_SIZE =
+ key("state.backend.forst.log.max-file-size")
+ .memoryType()
+ .defaultValue(MemorySize.parse("25mb"))
+ .withDescription(
+ "The maximum size of ForSt's file used for information logging. "
+ + "If the log files becomes larger than this, a new file will be created. "
+ + "If 0, all logs will be written to one log file. "
+ + "The default maximum file size is '25MB'. ");
+
+ public static final ConfigOption<Integer> LOG_FILE_NUM =
+ key("state.backend.forst.log.file-num")
+ .intType()
+ .defaultValue(4)
+ .withDescription(
+ "The maximum number of files ForSt should keep for information logging (Default setting: 4).");
+
+ public static final ConfigOption<String> LOG_DIR =
+ key("state.backend.forst.log.dir")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "The directory for ForSt's information logging files. "
+ + "If empty (Flink default setting), log files will be in the same directory as the Flink log. "
+ + "If non-empty, this directory will be used and the data directory's absolute path will be used as the prefix of the log file name. "
+ + "If setting this option as a non-existing location, e.g '/dev/null', ForSt will then create the log under its own database folder as before.");
+
+ public static final ConfigOption<InfoLogLevel> LOG_LEVEL =
+ key("state.backend.forst.log.level")
+ .enumType(InfoLogLevel.class)
+ .defaultValue(INFO_LEVEL)
+ .withDescription(
+ Description.builder()
+ .text(
+ "The specified information logging level for ForSt. "
+ + "If unset, Flink will use %s.",
+ code(INFO_LEVEL.name()))
+ .linebreak()
+ .text(
+ "Note: ForSt info logs will not be written to the TaskManager logs and there "
+ + "is no rolling strategy, unless you configure %s, %s, and %s accordingly. "
+ + "Without a rolling strategy, long-running tasks may lead to uncontrolled "
+ + "disk space usage if configured with increased log levels!",
+ code(LOG_DIR.key()),
+ code(LOG_MAX_FILE_SIZE.key()),
+ code(LOG_FILE_NUM.key()))
+ .linebreak()
+ .text(
+ "There is no need to modify the ForSt log level, unless for troubleshooting ForSt.")
+ .build());
+
+ // --------------------------------------------------------------------------
+ // Provided configurable ColumnFamilyOptions within Flink
+ // --------------------------------------------------------------------------
+
+ public static final ConfigOption<CompactionStyle> COMPACTION_STYLE =
+ key("state.backend.forst.compaction.style")
+ .enumType(CompactionStyle.class)
+ .defaultValue(LEVEL)
+ .withDescription(
+ String.format(
+ "The specified compaction style for DB. Candidate compaction style is %s, %s, %s or %s, "
+ + "and Flink chooses '%s' as default style.",
+ LEVEL.name(),
+ FIFO.name(),
+ UNIVERSAL.name(),
+ NONE.name(),
+ LEVEL.name()));
+
+ public static final ConfigOption<Boolean> USE_DYNAMIC_LEVEL_SIZE =
+ key("state.backend.forst.compaction.level.use-dynamic-size")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ Description.builder()
+ .text(
+ "If true, ForSt will pick target size of each level dynamically. From an empty DB, ")
+ .text(
+ "ForSt would make last level the base level, which means merging L0 data into the last level, ")
+ .text(
+ "until it exceeds max_bytes_for_level_base. And then repeat this process for second last level and so on. ")
+ .text("The default value is 'false'. ")
+ .text(
+ "For more information, please refer to %s",
+ link(
+ "https://github.com/facebook/rocksdb/wiki/Leveled-Compaction#level_compaction_dynamic_level_bytes-is-true",
+ "RocksDB's doc."))
+ .build());
+
+ public static final ConfigOption<List<CompressionType>> COMPRESSION_PER_LEVEL =
+ key("state.backend.forst.compression.per.level")
+ .enumType(CompressionType.class)
+ .asList()
+ .defaultValues(SNAPPY_COMPRESSION)
+ .withDescription(
+ Description.builder()
+ .text(
+ "A semicolon-separated list of Compression Type. Different levels can have different "
+ + "compression policies. In many cases, lower levels use fast compression algorithms,"
+ + " while higher levels with more data use slower but more effective compression algorithms. "
+ + "The N th element in the List corresponds to the compression type of the level N-1"
+ + "When %s is true, compression_per_level[0] still determines L0, but other "
+ + "elements are based on the base level and may not match the level seen in the info log",
+ code(USE_DYNAMIC_LEVEL_SIZE.key()))
+ .linebreak()
+ .text(
+ "Note: If the List size is smaller than the level number, the undefined lower level uses the last Compression Type in the List")
+ .linebreak()
+ .text(
+ "Some commonly used compression algorithms for candidates include %s ,%s and %s",
+ code(NO_COMPRESSION.name()),
+ code(SNAPPY_COMPRESSION.name()),
+ code(LZ4_COMPRESSION.name()))
+ .linebreak()
+ .text(
+ "The default value is %s, which means that all data uses the Snappy compression algorithm.",
+ code(SNAPPY_COMPRESSION.name()))
+ .text(
+ "Likewise, if set to %s , means that all data is not compressed, which will achieve faster speed but will bring some space amplification.",
+ code(NO_COMPRESSION.name()))
+ .text(
+ "In addition, if we need to consider both spatial amplification and performance, we can also set it to '%s;%s;%s', which means that L0 and L1 data will not be compressed, and other data will be compressed using LZ4.",
+ code(NO_COMPRESSION.name()),
+ code(NO_COMPRESSION.name()),
+ code(LZ4_COMPRESSION.name()))
+ .build());
+
+ public static final ConfigOption<MemorySize> TARGET_FILE_SIZE_BASE =
+ key("state.backend.forst.compaction.level.target-file-size-base")
+ .memoryType()
+ .defaultValue(MemorySize.parse("64mb"))
+ .withDescription(
+ "The target file size for compaction, which determines a level-1 file size. "
+ + "The default value is '64MB'.");
+
+ public static final ConfigOption<MemorySize> MAX_SIZE_LEVEL_BASE =
+ key("state.backend.forst.compaction.level.max-size-level-base")
+ .memoryType()
+ .defaultValue(MemorySize.parse("256mb"))
+ .withDescription(
+ "The upper-bound of the total size of level base files in bytes. "
+ + "The default value is '256MB'.");
+
+ public static final ConfigOption<MemorySize> WRITE_BUFFER_SIZE =
+ key("state.backend.forst.writebuffer.size")
+ .memoryType()
+ .defaultValue(MemorySize.parse("64mb"))
+ .withDescription(
+ "The amount of data built up in memory (backed by an unsorted log on disk) "
+ + "before converting to a sorted on-disk files. The default writebuffer size is '64MB'.");
+
+ public static final ConfigOption<Integer> MAX_WRITE_BUFFER_NUMBER =
+ key("state.backend.forst.writebuffer.count")
+ .intType()
+ .defaultValue(2)
+ .withDescription(
+ "The maximum number of write buffers that are built up in memory. "
+ + "The default value is '2'.");
+
+ public static final ConfigOption<Integer> MIN_WRITE_BUFFER_NUMBER_TO_MERGE =
+ key("state.backend.forst.writebuffer.number-to-merge")
+ .intType()
+ .defaultValue(1)
+ .withDescription(
+ "The minimum number of write buffers that will be merged together before writing to storage. "
+ + "The default value is '1'.");
+
+ public static final ConfigOption<MemorySize> BLOCK_SIZE =
+ key("state.backend.forst.block.blocksize")
+ .memoryType()
+ .defaultValue(MemorySize.parse("4kb"))
+ .withDescription(
+ "The approximate size (in bytes) of user data packed per block. "
+ + "The default blocksize is '4KB'.");
+
+ public static final ConfigOption<MemorySize> METADATA_BLOCK_SIZE =
+ key("state.backend.forst.block.metadata-blocksize")
+ .memoryType()
+ .defaultValue(MemorySize.parse("4kb"))
+ .withDescription(
+ "Approximate size of partitioned metadata packed per block. "
+ + "Currently applied to indexes block when partitioned index/filters option is enabled. "
+ + "The default blocksize is '4KB'.");
+
+ public static final ConfigOption<MemorySize> BLOCK_CACHE_SIZE =
+ key("state.backend.forst.block.cache-size")
+ .memoryType()
+ .defaultValue(MemorySize.parse("8mb"))
+ .withDescription(
+ "The amount of the cache for data blocks in ForSt. "
+ + "The default block-cache size is '8MB'.");
+
+ public static final ConfigOption<MemorySize> WRITE_BATCH_SIZE =
+ key("state.backend.forst.write-batch-size")
+ .memoryType()
+ .defaultValue(MemorySize.parse("2mb"))
+ .withDescription(
+ "The max size of the consumed memory for ForSt batch write, "
+ + "will flush just based on item count if this config set to 0.");
+
+ public static final ConfigOption<Boolean> USE_BLOOM_FILTER =
+ key("state.backend.forst.use-bloom-filter")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "If true, every newly created SST file will contain a Bloom filter. "
+ + "It is disabled by default.");
+
+ public static final ConfigOption<Double> BLOOM_FILTER_BITS_PER_KEY =
+ key("state.backend.forst.bloom-filter.bits-per-key")
+ .doubleType()
+ .defaultValue(10.0)
+ .withDescription(
+ "Bits per key that bloom filter will use, this only take effect when bloom filter is used. "
+ + "The default value is 10.0.");
+
+ public static final ConfigOption<Boolean> BLOOM_FILTER_BLOCK_BASED_MODE =
+ key("state.backend.forst.bloom-filter.block-based-mode")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "If true, ForSt will use block-based filter instead of full filter, this only take effect when bloom filter is used. "
+ + "The default value is 'false'.");
+
+ static final ConfigOption<?>[] CANDIDATE_CONFIGS =
+ new ConfigOption<?>[] {
+ // configurable DBOptions
+ MAX_BACKGROUND_THREADS,
+ MAX_OPEN_FILES,
+ LOG_LEVEL,
+ LOG_MAX_FILE_SIZE,
+ LOG_FILE_NUM,
+ LOG_DIR,
+
+ // configurable ColumnFamilyOptions
+ COMPACTION_STYLE,
+ COMPRESSION_PER_LEVEL,
+ USE_DYNAMIC_LEVEL_SIZE,
+ TARGET_FILE_SIZE_BASE,
+ MAX_SIZE_LEVEL_BASE,
+ WRITE_BUFFER_SIZE,
+ MAX_WRITE_BUFFER_NUMBER,
+ MIN_WRITE_BUFFER_NUMBER_TO_MERGE,
+ BLOCK_SIZE,
+ METADATA_BLOCK_SIZE,
+ BLOCK_CACHE_SIZE,
+ USE_BLOOM_FILTER,
+ BLOOM_FILTER_BITS_PER_KEY,
+ BLOOM_FILTER_BLOCK_BASED_MODE,
+ };
+
+ private static final Set<ConfigOption<?>> POSITIVE_INT_CONFIG_SET =
+ new HashSet<>(
+ Arrays.asList(
+ MAX_BACKGROUND_THREADS,
+ LOG_FILE_NUM,
+ MAX_WRITE_BUFFER_NUMBER,
+ MIN_WRITE_BUFFER_NUMBER_TO_MERGE));
+
+ private static final Set<ConfigOption<?>> SIZE_CONFIG_SET =
+ new HashSet<>(
+ Arrays.asList(
+ TARGET_FILE_SIZE_BASE,
+ MAX_SIZE_LEVEL_BASE,
+ WRITE_BUFFER_SIZE,
+ BLOCK_SIZE,
+ METADATA_BLOCK_SIZE,
+ BLOCK_CACHE_SIZE));
+
+ /**
+ * Helper method to check whether the (key,value) is valid through given configuration and
+ * returns the formatted value.
+ *
+ * @param option The configuration key which is configurable in {@link
+ * ForStConfigurableOptions}.
+ * @param value The value within given configuration.
+ */
+ static void checkArgumentValid(ConfigOption<?> option, Object value) {
+ final String key = option.key();
+
+ if (POSITIVE_INT_CONFIG_SET.contains(option)) {
+ Preconditions.checkArgument(
+ (Integer) value > 0,
+ "Configured value for key: " + key + " must be larger than 0.");
+ } else if (SIZE_CONFIG_SET.contains(option)) {
+ Preconditions.checkArgument(
+ ((MemorySize) value).getBytes() > 0,
+ "Configured size for key" + key + " must be larger than 0.");
+ } else if (LOG_MAX_FILE_SIZE.equals(option)) {
+ Preconditions.checkArgument(
+ ((MemorySize) value).getBytes() >= 0,
+ "Configured size for key " + key + " must be larger than or equal to 0.");
+ } else if (LOG_DIR.equals(option)) {
+ Preconditions.checkArgument(
+ new File((String) value).isAbsolute(),
+ "Configured path for key " + key + " is not absolute.");
+ }
+ }
+}
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStOptions.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStOptions.java
new file mode 100644
index 00000000000..ba8ce6b2c62
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStOptions.java
@@ -0,0 +1,123 @@
+/*
+ * 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.flink.state.forst;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.configuration.ClusterOptions;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.MemorySize;
+import org.apache.flink.configuration.description.Description;
+import org.apache.flink.configuration.description.TextElement;
+
+/** Configuration options for the ForStStateBackend. */
+@Experimental
+public class ForStOptions {
+
+ /** The local directory (on the TaskManager) where ForSt puts its some files. */
+ public static final ConfigOption<String> LOCAL_DIRECTORIES =
+ ConfigOptions.key("state.backend.forst.localdir")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "The local directory (on the TaskManager) where ForSt puts its some files. Per default, it will be <WORKING_DIR>/tmp. See %s for more details.",
+ TextElement.code(
+ ClusterOptions
+ .TASK_MANAGER_PROCESS_WORKING_DIR_BASE
+ .key()))
+ .build());
+
+ /** The options factory class for ForSt to create DBOptions and ColumnFamilyOptions. */
+ public static final ConfigOption<String> OPTIONS_FACTORY =
+ ConfigOptions.key("state.backend.forst.options-factory")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "The options factory class for users to add customized options in DBOptions and ColumnFamilyOptions for ForSt. "
+ + "If set, the ForSt state backend will load the class and apply configs to DBOptions and ColumnFamilyOptions "
+ + "after loading ones from 'ForStConfigurableOptions' and pre-defined options.");
+
+ public static final ConfigOption<Boolean> USE_MANAGED_MEMORY =
+ ConfigOptions.key("state.backend.forst.memory.managed")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "If set, the ForSt state backend will automatically configure itself to use the "
+ + "managed memory budget of the task slot, and divide the memory over write buffers, indexes, "
+ + "block caches, etc. That way, the three major uses of memory of ForSt will be capped.");
+
+ public static final ConfigOption<MemorySize> FIX_PER_SLOT_MEMORY_SIZE =
+ ConfigOptions.key("state.backend.forst.memory.fixed-per-slot")
+ .memoryType()
+ .noDefaultValue()
+ .withDescription(
+ String.format(
+ "The fixed total amount of memory, shared among all ForSt instances per slot. "
+ + "This option overrides the '%s' option when configured.",
+ USE_MANAGED_MEMORY.key()));
+
+ public static final ConfigOption<MemorySize> FIX_PER_TM_MEMORY_SIZE =
+ ConfigOptions.key("state.backend.forst.memory.fixed-per-tm")
+ .memoryType()
+ .noDefaultValue()
+ .withDescription(
+ String.format(
+ "The fixed total amount of memory, shared among all ForSt instances per Task Manager (cluster-level option). "
+ + "This option only takes effect if neither '%s' nor '%s' are not configured. If none is configured "
+ + "then each ForSt column family state has its own memory caches (as controlled by the column "
+ + "family options). "
+ + "The relevant options for the shared resources (e.g. write-buffer-ratio) can be set on the same level (config.yaml)."
+ + "Note, that this feature breaks resource isolation between the slots",
+ USE_MANAGED_MEMORY.key(), FIX_PER_SLOT_MEMORY_SIZE.key()));
+
+ public static final ConfigOption<Double> WRITE_BUFFER_RATIO =
+ ConfigOptions.key("state.backend.forst.memory.write-buffer-ratio")
+ .doubleType()
+ .defaultValue(0.5)
+ .withDescription(
+ String.format(
+ "The maximum amount of memory that write buffers may take, as a fraction of the total shared memory. "
+ + "This option only has an effect when '%s' or '%s' are configured.",
+ USE_MANAGED_MEMORY.key(), FIX_PER_SLOT_MEMORY_SIZE.key()));
+
+ public static final ConfigOption<Double> HIGH_PRIORITY_POOL_RATIO =
+ ConfigOptions.key("state.backend.forst.memory.high-prio-pool-ratio")
+ .doubleType()
+ .defaultValue(0.1)
+ .withDescription(
+ String.format(
+ "The fraction of cache memory that is reserved for high-priority data like index, filter, and "
+ + "compression dictionary blocks. This option only has an effect when '%s' or '%s' are configured.",
+ USE_MANAGED_MEMORY.key(), FIX_PER_SLOT_MEMORY_SIZE.key()));
+
+ public static final ConfigOption<Boolean> USE_PARTITIONED_INDEX_FILTERS =
+ ConfigOptions.key("state.backend.forst.memory.partitioned-index-filters")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ String.format(
+ "With partitioning, the index/filter block of an SST file is partitioned into smaller blocks with "
+ + "an additional top-level index on them. When reading an index/filter, only top-level index is loaded into memory. "
+ + "The partitioned index/filter then uses the top-level index to load on demand into the block cache "
+ + "the partitions that are required to perform the index/filter query. "
+ + "This option only has an effect when '%s' or '%s' are configured.",
+ USE_MANAGED_MEMORY.key(), FIX_PER_SLOT_MEMORY_SIZE.key()));
+}
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStOptionsFactory.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStOptionsFactory.java
new file mode 100644
index 00000000000..7fbc432e659
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStOptionsFactory.java
@@ -0,0 +1,106 @@
+/*
+ * 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.flink.state.forst;
+
+import org.apache.flink.annotation.Experimental;
+
+import org.rocksdb.ColumnFamilyOptions;
+import org.rocksdb.DBOptions;
+import org.rocksdb.ReadOptions;
+import org.rocksdb.WriteOptions;
+
+import java.util.Collection;
+
+/**
+ * A factory for {@link DBOptions} and {@link ColumnFamilyOptions} to be passed to the
+ * ForStStateBackend. Options have to be created lazily by this factory, because the {@code Options}
+ * class is not serializable and holds pointers to native code.
+ */
+@Experimental
+public interface ForStOptionsFactory extends java.io.Serializable {
+
+ /**
+ * This method should set the additional options on top of the current options object. The
+ * current options object may contain pre-defined options based on flags that have been
+ * configured on the state backend.
+ *
+ * <p>It is important to set the options on the current object and return the result from the
+ * setter methods, otherwise the pre-defined options may get lost.
+ *
+ * @param currentOptions The options object with the pre-defined options.
+ * @param handlesToClose The collection to register newly created {@link
+ * org.rocksdb.RocksObject}s.
+ * @return The options object on which the additional options are set.
+ */
+ DBOptions createDBOptions(DBOptions currentOptions, Collection<AutoCloseable> handlesToClose);
+
+ /**
+ * This method should set the additional options on top of the current options object. The
+ * current options object may contain pre-defined options based on flags that have been
+ * configured on the state backend.
+ *
+ * <p>It is important to set the options on the current object and return the result from the
+ * setter methods, otherwise the pre-defined options may get lost.
+ *
+ * @param currentOptions The options object with the pre-defined options.
+ * @param handlesToClose The collection to register newly created {@link
+ * org.rocksdb.RocksObject}s.
+ * @return The options object on which the additional options are set.
+ */
+ ColumnFamilyOptions createColumnOptions(
+ ColumnFamilyOptions currentOptions, Collection<AutoCloseable> handlesToClose);
+
+ // TODO: Support Metrics Options
+
+ /**
+ * This method should set the additional options on top of the current options object. The
+ * current options object may contain pre-defined options based on flags that have been
+ * configured on the state backend.
+ *
+ * <p>It is important to set the options on the current object and return the result from the
+ * setter methods, otherwise the pre-defined options may get lost.
+ *
+ * @param currentOptions The options object with the pre-defined options.
+ * @param handlesToClose The collection to register newly created {@link
+ * org.rocksdb.RocksObject}s.
+ * @return The options object on which the additional options are set.
+ */
+ default WriteOptions createWriteOptions(
+ WriteOptions currentOptions, Collection<AutoCloseable> handlesToClose) {
+ return currentOptions;
+ }
+
+ /**
+ * This method should set the additional options on top of the current options object. The
+ * current options object may contain pre-defined options based on flags that have been
+ * configured on the state backend.
+ *
+ * <p>It is important to set the options on the current object and return the result from the
+ * setter methods, otherwise the pre-defined options may get lost.
+ *
+ * @param currentOptions The options object with the pre-defined options.
+ * @param handlesToClose The collection to register newly created {@link
+ * org.rocksdb.RocksObject}s.
+ * @return The options object on which the additional options are set.
+ */
+ default ReadOptions createReadOptions(
+ ReadOptions currentOptions, Collection<AutoCloseable> handlesToClose) {
+ return currentOptions;
+ }
+}
(flink) 04/07: [FLINK-35047][state] Introduce ResourceContainer for ForStStateBackend
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hangxiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 5f5d5022c03b59f943e357c0c8028325bf12e8c0
Author: Hangxiang Yu <ma...@gmail.com>
AuthorDate: Mon Apr 15 23:16:43 2024 +0800
[FLINK-35047][state] Introduce ResourceContainer for ForStStateBackend
---
.../state/forst/ForStMemoryConfiguration.java | 248 +++++++++++
.../state/forst/ForStMemoryControllerUtils.java | 195 +++++++++
.../flink/state/forst/ForStResourceContainer.java | 468 +++++++++++++++++++++
.../flink/state/forst/ForStSharedResources.java | 73 ++++
.../state/forst/ForStSharedResourcesFactory.java | 185 ++++++++
.../forst/ForStMemoryControllerUtilsTest.java | 129 ++++++
.../state/forst/ForStResourceContainerTest.java | 297 +++++++++++++
7 files changed, 1595 insertions(+)
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStMemoryConfiguration.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStMemoryConfiguration.java
new file mode 100644
index 00000000000..558f48cb0a1
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStMemoryConfiguration.java
@@ -0,0 +1,248 @@
+/*
+ * 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.flink.state.forst;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.MemorySize;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.util.Preconditions;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+/** The settings regarding ForSt memory usage. */
+public final class ForStMemoryConfiguration implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ /** Flag whether to use the managed memory budget for ForSt. Null is not set. */
+ @Nullable private Boolean useManagedMemory;
+
+ /** The total memory for all ForSt instances at this slot. Null is not set. */
+ @Nullable private MemorySize fixedMemoryPerSlot;
+
+ /**
+ * The maximum fraction of the total shared memory consumed by the write buffers. Null if not
+ * set.
+ */
+ @Nullable private Double writeBufferRatio;
+
+ /**
+ * The high priority pool ratio in the shared cache, used for index & filter blocks. Null if not
+ * set.
+ */
+ @Nullable private Double highPriorityPoolRatio;
+
+ /** Flag whether to use partition index/filters. Null if not set. */
+ @Nullable private Boolean usePartitionedIndexFilters;
+
+ // ------------------------------------------------------------------------
+
+ /**
+ * Configures ForSt to use the managed memory of a slot. See {@link
+ * ForStOptions#USE_MANAGED_MEMORY} for details.
+ */
+ public void setUseManagedMemory(boolean useManagedMemory) {
+ this.useManagedMemory = useManagedMemory;
+ }
+
+ /**
+ * Configures ForSt to use a fixed amount of memory shared between all instances (operators) in
+ * a slot. See {@link ForStOptions#FIX_PER_SLOT_MEMORY_SIZE} for details.
+ */
+ public void setFixedMemoryPerSlot(MemorySize fixedMemoryPerSlot) {
+ checkArgument(
+ fixedMemoryPerSlot == null || fixedMemoryPerSlot.getBytes() > 0,
+ "Total memory per slot must be > 0");
+
+ this.fixedMemoryPerSlot = fixedMemoryPerSlot;
+ }
+
+ /**
+ * Configures ForSt to use a fixed amount of memory shared between all instances (operators) in
+ * a slot. See {@link #setFixedMemoryPerSlot(MemorySize)} for details.
+ */
+ public void setFixedMemoryPerSlot(String totalMemoryPerSlotStr) {
+ setFixedMemoryPerSlot(MemorySize.parse(totalMemoryPerSlotStr));
+ }
+
+ /**
+ * Sets the fraction of the total memory to be used for write buffers. This only has an effect
+ * is either {@link #setUseManagedMemory(boolean)} or {@link #setFixedMemoryPerSlot(MemorySize)}
+ * are set.
+ *
+ * <p>See {@link ForStOptions#WRITE_BUFFER_RATIO} for details.
+ */
+ public void setWriteBufferRatio(double writeBufferRatio) {
+ Preconditions.checkArgument(
+ writeBufferRatio > 0 && writeBufferRatio < 1.0,
+ "Write Buffer ratio %s must be in (0, 1)",
+ writeBufferRatio);
+ this.writeBufferRatio = writeBufferRatio;
+ }
+
+ /**
+ * Sets the fraction of the total memory to be used for high priority blocks like indexes,
+ * dictionaries, etc. This only has an effect is either {@link #setUseManagedMemory(boolean)} or
+ * {@link #setFixedMemoryPerSlot(MemorySize)} are set.
+ *
+ * <p>See {@link ForStOptions#HIGH_PRIORITY_POOL_RATIO} for details.
+ */
+ public void setHighPriorityPoolRatio(double highPriorityPoolRatio) {
+ Preconditions.checkArgument(
+ highPriorityPoolRatio > 0 && highPriorityPoolRatio < 1.0,
+ "High priority pool ratio %s must be in (0, 1)",
+ highPriorityPoolRatio);
+ this.highPriorityPoolRatio = highPriorityPoolRatio;
+ }
+
+ /**
+ * Gets whether the state backend is configured to use the managed memory of a slot for ForSt.
+ * See {@link ForStOptions#USE_MANAGED_MEMORY} for details.
+ */
+ public boolean isUsingManagedMemory() {
+ return useManagedMemory != null
+ ? useManagedMemory
+ : ForStOptions.USE_MANAGED_MEMORY.defaultValue();
+ }
+
+ /**
+ * Gets whether the state backend is configured to use a fixed amount of memory shared between
+ * all ForSt instances (in all tasks and operators) of a slot. See {@link
+ * ForStOptions#FIX_PER_SLOT_MEMORY_SIZE} for details.
+ */
+ public boolean isUsingFixedMemoryPerSlot() {
+ return fixedMemoryPerSlot != null;
+ }
+
+ /**
+ * Gets the fixed amount of memory to be shared between all RocksDB instances (in all tasks and
+ * operators) of a slot. Null is not configured. See {@link ForStOptions#USE_MANAGED_MEMORY} for
+ * details.
+ */
+ @Nullable
+ public MemorySize getFixedMemoryPerSlot() {
+ return fixedMemoryPerSlot;
+ }
+
+ /**
+ * Gets the fraction of the total memory to be used for write buffers. This only has an effect
+ * is either {@link #setUseManagedMemory(boolean)} or {@link #setFixedMemoryPerSlot(MemorySize)}
+ * are set.
+ *
+ * <p>See {@link ForStOptions#WRITE_BUFFER_RATIO} for details.
+ */
+ public double getWriteBufferRatio() {
+ return writeBufferRatio != null
+ ? writeBufferRatio
+ : ForStOptions.WRITE_BUFFER_RATIO.defaultValue();
+ }
+
+ /**
+ * Gets the fraction of the total memory to be used for high priority blocks like indexes,
+ * dictionaries, etc. This only has an effect is either {@link #setUseManagedMemory(boolean)} or
+ * {@link #setFixedMemoryPerSlot(MemorySize)} are set.
+ *
+ * <p>See {@link ForStOptions#HIGH_PRIORITY_POOL_RATIO} for details.
+ */
+ public double getHighPriorityPoolRatio() {
+ return highPriorityPoolRatio != null
+ ? highPriorityPoolRatio
+ : ForStOptions.HIGH_PRIORITY_POOL_RATIO.defaultValue();
+ }
+
+ /**
+ * Gets whether the state backend is configured to use partitioned index/filters for ForSt.
+ *
+ * <p>See {@link ForStOptions#USE_PARTITIONED_INDEX_FILTERS} for details.
+ */
+ public Boolean isUsingPartitionedIndexFilters() {
+ return usePartitionedIndexFilters != null
+ ? usePartitionedIndexFilters
+ : ForStOptions.USE_PARTITIONED_INDEX_FILTERS.defaultValue();
+ }
+
+ // ------------------------------------------------------------------------
+
+ /** Validates if the configured options are valid with respect to one another. */
+ public void validate() {
+ // As FLINK-15512 introduce a new mechanism to calculate the cache capacity,
+ // the relationship of write_buffer_manager_capacity and cache_capacity has changed to:
+ // write_buffer_manager_capacity / cache_capacity = 2 * writeBufferRatio / (3 -
+ // writeBufferRatio)
+ // we should ensure the sum of write buffer manager capacity and high priority pool less
+ // than cache capacity.
+ // TODO change the formula once FLINK-15532 resolved.
+ if (writeBufferRatio != null
+ && highPriorityPoolRatio != null
+ && 2 * writeBufferRatio / (3 - writeBufferRatio) + highPriorityPoolRatio >= 1.0) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Invalid configuration: writeBufferRatio %s with highPriPoolRatio %s",
+ writeBufferRatio, highPriorityPoolRatio));
+ }
+ }
+
+ // ------------------------------------------------------------------------
+
+ /**
+ * Derives a ForStMemoryConfiguration from another object and a configuration. The values set on
+ * the other object take precedence, and the values from the configuration are used if no values
+ * are set on the other config object.
+ */
+ public static ForStMemoryConfiguration fromOtherAndConfiguration(
+ ForStMemoryConfiguration other, ReadableConfig config) {
+
+ final ForStMemoryConfiguration newConfig = new ForStMemoryConfiguration();
+
+ newConfig.useManagedMemory =
+ other.useManagedMemory != null
+ ? other.useManagedMemory
+ : config.get(ForStOptions.USE_MANAGED_MEMORY);
+
+ newConfig.fixedMemoryPerSlot =
+ other.fixedMemoryPerSlot != null
+ ? other.fixedMemoryPerSlot
+ : config.get(ForStOptions.FIX_PER_SLOT_MEMORY_SIZE);
+
+ newConfig.writeBufferRatio =
+ other.writeBufferRatio != null
+ ? other.writeBufferRatio
+ : config.get(ForStOptions.WRITE_BUFFER_RATIO);
+
+ newConfig.highPriorityPoolRatio =
+ other.highPriorityPoolRatio != null
+ ? other.highPriorityPoolRatio
+ : config.get(ForStOptions.HIGH_PRIORITY_POOL_RATIO);
+
+ newConfig.usePartitionedIndexFilters =
+ other.usePartitionedIndexFilters != null
+ ? other.usePartitionedIndexFilters
+ : config.get(ForStOptions.USE_PARTITIONED_INDEX_FILTERS);
+
+ return newConfig;
+ }
+
+ public static ForStMemoryConfiguration fromConfiguration(Configuration configuration) {
+ return fromOtherAndConfiguration(new ForStMemoryConfiguration(), configuration);
+ }
+}
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStMemoryControllerUtils.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStMemoryControllerUtils.java
new file mode 100644
index 00000000000..670c991cd60
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStMemoryControllerUtils.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.flink.state.forst;
+
+import org.apache.flink.annotation.VisibleForTesting;
+
+import org.rocksdb.Cache;
+import org.rocksdb.LRUCache;
+import org.rocksdb.WriteBufferManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Serializable;
+
+/**
+ * Utils to create {@link Cache} and {@link WriteBufferManager} which are used to control total
+ * memory usage of ForSt.
+ */
+public class ForStMemoryControllerUtils {
+ private static final Logger LOG = LoggerFactory.getLogger(ForStMemoryControllerUtils.class);
+
+ /**
+ * Allocate memory controllable ForSt shared resources.
+ *
+ * @param totalMemorySize The total memory limit size.
+ * @param writeBufferRatio The ratio of total memory which is occupied by write buffer manager.
+ * @param highPriorityPoolRatio The high priority pool ratio of cache.
+ * @param factory creates Write Buffer Manager and Bock Cache
+ * @return memory controllable RocksDB shared resources.
+ */
+ public static ForStSharedResources allocateForStSharedResources(
+ long totalMemorySize,
+ double writeBufferRatio,
+ double highPriorityPoolRatio,
+ boolean usingPartitionedIndexFilters,
+ ForStMemoryFactory factory) {
+
+ long calculatedCacheCapacity =
+ ForStMemoryControllerUtils.calculateActualCacheCapacity(
+ totalMemorySize, writeBufferRatio);
+ final Cache cache = factory.createCache(calculatedCacheCapacity, highPriorityPoolRatio);
+
+ long writeBufferManagerCapacity =
+ ForStMemoryControllerUtils.calculateWriteBufferManagerCapacity(
+ totalMemorySize, writeBufferRatio);
+ final WriteBufferManager wbm =
+ factory.createWriteBufferManager(writeBufferManagerCapacity, cache);
+
+ LOG.debug(
+ "Allocated ForSt shared resources, calculatedCacheCapacity: {}, highPriorityPoolRatio: {}, writeBufferManagerCapacity: {}, usingPartitionedIndexFilters: {}",
+ calculatedCacheCapacity,
+ highPriorityPoolRatio,
+ writeBufferManagerCapacity,
+ usingPartitionedIndexFilters);
+ return new ForStSharedResources(
+ cache, wbm, writeBufferManagerCapacity, usingPartitionedIndexFilters);
+ }
+
+ /**
+ * Calculate the actual memory capacity of cache, which would be shared among ForSt instance(s).
+ * We introduce this method because: a) We cannot create a strict capacity limit cache util
+ * FLINK-15532 resolved. b) Regardless of the memory usage of blocks pinned by ForSt iterators,
+ * which is difficult to calculate and only happened when we iterator entries in MapState, the
+ * overuse of memory is mainly occupied by at most half of the write buffer usage. (see <a
+ * href="https://github.com/dataArtisans/frocksdb/blob/958f191d3f7276ae59b270f9db8390034d549ee0/include/rocksdb/write_buffer_manager.h#L51">the
+ * flush implementation of write buffer manager</a>). Thus, we have four equations below:
+ * write_buffer_manager_memory = 1.5 * write_buffer_manager_capacity write_buffer_manager_memory
+ * = total_memory_size * write_buffer_ratio write_buffer_manager_memory + other_part =
+ * total_memory_size write_buffer_manager_capacity + other_part = cache_capacity And we would
+ * deduce the formula: cache_capacity = (3 - write_buffer_ratio) * total_memory_size / 3
+ * write_buffer_manager_capacity = 2 * total_memory_size * write_buffer_ratio / 3
+ *
+ * @param totalMemorySize Total off-heap memory size reserved for ForSt instance(s).
+ * @param writeBufferRatio The ratio of total memory size which would be reserved for write
+ * buffer manager and its over-capacity part.
+ * @return The actual calculated cache capacity.
+ */
+ @VisibleForTesting
+ public static long calculateActualCacheCapacity(long totalMemorySize, double writeBufferRatio) {
+ return (long) ((3 - writeBufferRatio) * totalMemorySize / 3);
+ }
+
+ /**
+ * Calculate the actual memory capacity of write buffer manager, which would be shared among
+ * ForSt instance(s). The formula to use here could refer to the doc of {@link
+ * #calculateActualCacheCapacity(long, double)}.
+ *
+ * @param totalMemorySize Total off-heap memory size reserved for ForSt instance(s).
+ * @param writeBufferRatio The ratio of total memory size which would be reserved for write
+ * buffer manager and its over-capacity part.
+ * @return The actual calculated write buffer manager capacity.
+ */
+ @VisibleForTesting
+ static long calculateWriteBufferManagerCapacity(long totalMemorySize, double writeBufferRatio) {
+ return (long) (2 * totalMemorySize * writeBufferRatio / 3);
+ }
+
+ @VisibleForTesting
+ static Cache createCache(long cacheCapacity, double highPriorityPoolRatio) {
+ // TODO use strict capacity limit until FLINK-15532 resolved
+ return new LRUCache(cacheCapacity, -1, false, highPriorityPoolRatio);
+ }
+
+ @VisibleForTesting
+ static WriteBufferManager createWriteBufferManager(
+ long writeBufferManagerCapacity, Cache cache) {
+ return new WriteBufferManager(writeBufferManagerCapacity, cache);
+ }
+
+ /**
+ * Calculate the default arena block size as ForSt calculates it in <a
+ * href="https://github.com/dataArtisans/frocksdb/blob/49bc897d5d768026f1eb816d960c1f2383396ef4/db/column_family.cc#L196-L201">
+ * here</a>.
+ *
+ * @return the default arena block size
+ * @param writeBufferSize the write buffer size (bytes)
+ */
+ static long calculateForStDefaultArenaBlockSize(long writeBufferSize) {
+ long arenaBlockSize = writeBufferSize / 8;
+
+ // Align up to 4k
+ final long align = 4 * 1024;
+ return ((arenaBlockSize + align - 1) / align) * align;
+ }
+
+ /**
+ * Calculate {@code mutable_limit_} as ForSt calculates it in <a
+ * href="https://github.com/dataArtisans/frocksdb/blob/FRocksDB-5.17.2/memtable/write_buffer_manager.cc#L54">
+ * here</a>.
+ *
+ * @param bufferSize write buffer size
+ * @return mutableLimit
+ */
+ static long calculateForStMutableLimit(long bufferSize) {
+ return bufferSize * 7 / 8;
+ }
+
+ /**
+ * ForSt starts flushing the active memtable constantly in the case when the arena block size is
+ * greater than mutable limit (as calculated in {@link #calculateForStMutableLimit(long)}).
+ *
+ * <p>This happens because in such a case the check <a
+ * href="https://github.com/dataArtisans/frocksdb/blob/958f191d3f7276ae59b270f9db8390034d549ee0/include/rocksdb/write_buffer_manager.h#L47">
+ * here</a> is always true.
+ *
+ * <p>This method checks that arena block size is smaller than mutable limit.
+ *
+ * @param arenaBlockSize Arena block size
+ * @param mutableLimit mutable limit
+ * @return whether arena block size is sensible
+ */
+ @VisibleForTesting
+ static boolean validateArenaBlockSize(long arenaBlockSize, long mutableLimit) {
+ return arenaBlockSize <= mutableLimit;
+ }
+
+ /** Factory for Write Buffer Manager and Bock Cache. */
+ public interface ForStMemoryFactory extends Serializable {
+ Cache createCache(long cacheCapacity, double highPriorityPoolRatio);
+
+ WriteBufferManager createWriteBufferManager(long writeBufferManagerCapacity, Cache cache);
+
+ ForStMemoryFactory DEFAULT =
+ new ForStMemoryFactory() {
+ @Override
+ public Cache createCache(long cacheCapacity, double highPriorityPoolRatio) {
+ return ForStMemoryControllerUtils.createCache(
+ cacheCapacity, highPriorityPoolRatio);
+ }
+
+ @Override
+ public WriteBufferManager createWriteBufferManager(
+ long writeBufferManagerCapacity, Cache cache) {
+ return ForStMemoryControllerUtils.createWriteBufferManager(
+ writeBufferManagerCapacity, cache);
+ }
+ };
+ }
+}
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java
new file mode 100644
index 00000000000..2f54522d1c7
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java
@@ -0,0 +1,468 @@
+/*
+ * 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.flink.state.forst;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.runtime.memory.OpaqueMemoryResource;
+import org.apache.flink.util.FileUtils;
+import org.apache.flink.util.IOUtils;
+import org.apache.flink.util.Preconditions;
+
+import org.rocksdb.BlockBasedTableConfig;
+import org.rocksdb.BloomFilter;
+import org.rocksdb.Cache;
+import org.rocksdb.ColumnFamilyOptions;
+import org.rocksdb.DBOptions;
+import org.rocksdb.Filter;
+import org.rocksdb.IndexType;
+import org.rocksdb.PlainTableConfig;
+import org.rocksdb.ReadOptions;
+import org.rocksdb.Statistics;
+import org.rocksdb.TableFormatConfig;
+import org.rocksdb.WriteOptions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+
+/**
+ * The container for ForSt resources, including option factory and shared resource among instances.
+ *
+ * <p>This should be the only entrance for ForStStateBackend to get ForSt options, and should be
+ * properly (and necessarily) closed to prevent resource leak.
+ */
+public final class ForStResourceContainer implements AutoCloseable {
+ private static final Logger LOG = LoggerFactory.getLogger(ForStResourceContainer.class);
+
+ private static final String FORST_RELOCATE_LOG_SUFFIX = "_LOG";
+
+ // the filename length limit is 255 on most operating systems
+ private static final int INSTANCE_PATH_LENGTH_LIMIT = 255 - FORST_RELOCATE_LOG_SUFFIX.length();
+
+ private static final String LOCAL_DB_DIR_STRING = "db";
+
+ @Nullable private final File localBasePath;
+
+ @Nullable private final File localForStPath;
+
+ /** The configurations from file. */
+ private final ReadableConfig configuration;
+
+ /** The options factory to create the ForSt options. */
+ @Nullable private final ForStOptionsFactory optionsFactory;
+
+ /**
+ * The shared resource among ForSt instances. This resource is not part of the 'handlesToClose',
+ * because the handles to close are closed quietly, whereas for this one, we want exceptions to
+ * be reported.
+ */
+ @Nullable private final OpaqueMemoryResource<ForStSharedResources> sharedResources;
+
+ private final boolean enableStatistics;
+
+ /** The handles to be closed when the container is closed. */
+ private final ArrayList<AutoCloseable> handlesToClose;
+
+ @Nullable private Path relocatedDbLogBaseDir;
+
+ @VisibleForTesting
+ public ForStResourceContainer() {
+ this(new Configuration(), null, null, null, false);
+ }
+
+ @VisibleForTesting
+ public ForStResourceContainer(@Nullable ForStOptionsFactory optionsFactory) {
+ this(new Configuration(), optionsFactory, null, null, false);
+ }
+
+ @VisibleForTesting
+ public ForStResourceContainer(
+ @Nullable ForStOptionsFactory optionsFactory,
+ @Nullable OpaqueMemoryResource<ForStSharedResources> sharedResources) {
+ this(new Configuration(), optionsFactory, sharedResources, null, false);
+ }
+
+ public ForStResourceContainer(
+ ReadableConfig configuration,
+ @Nullable ForStOptionsFactory optionsFactory,
+ @Nullable OpaqueMemoryResource<ForStSharedResources> sharedResources,
+ @Nullable File localBasePath,
+ boolean enableStatistics) {
+
+ this.configuration = configuration;
+ this.optionsFactory = optionsFactory;
+ this.sharedResources = sharedResources;
+
+ this.localBasePath = localBasePath;
+ this.localForStPath =
+ localBasePath != null ? new File(localBasePath, LOCAL_DB_DIR_STRING) : null;
+
+ this.enableStatistics = enableStatistics;
+ this.handlesToClose = new ArrayList<>();
+ }
+
+ /** Gets the ForSt {@link DBOptions} to be used for ForSt instances. */
+ public DBOptions getDbOptions() {
+ // initial options from common profile
+ DBOptions opt = createBaseCommonDBOptions();
+ handlesToClose.add(opt);
+
+ // load configurable options on top of pre-defined profile
+ setDBOptionsFromConfigurableOptions(opt);
+
+ // add user-defined options factory, if specified
+ if (optionsFactory != null) {
+ opt = optionsFactory.createDBOptions(opt, handlesToClose);
+ }
+
+ // add necessary default options
+ opt = opt.setCreateIfMissing(true).setAvoidFlushDuringShutdown(true);
+
+ // if sharedResources is non-null, use the write buffer manager from it.
+ if (sharedResources != null) {
+ opt.setWriteBufferManager(sharedResources.getResourceHandle().getWriteBufferManager());
+ }
+
+ if (enableStatistics) {
+ Statistics statistics = new Statistics();
+ opt.setStatistics(statistics);
+ handlesToClose.add(statistics);
+ }
+
+ return opt;
+ }
+
+ /** Gets the ForSt {@link ColumnFamilyOptions} to be used for all ForSt instances. */
+ public ColumnFamilyOptions getColumnOptions() {
+ // initial options from common profile
+ ColumnFamilyOptions opt = createBaseCommonColumnOptions();
+ handlesToClose.add(opt);
+
+ // load configurable options on top of pre-defined profile
+ setColumnFamilyOptionsFromConfigurableOptions(opt, handlesToClose);
+
+ // add user-defined options, if specified
+ if (optionsFactory != null) {
+ opt = optionsFactory.createColumnOptions(opt, handlesToClose);
+ }
+
+ // if sharedResources is non-null, use the block cache from it and
+ // set necessary options for performance consideration with memory control
+ if (sharedResources != null) {
+ final ForStSharedResources rocksResources = sharedResources.getResourceHandle();
+ final Cache blockCache = rocksResources.getCache();
+ TableFormatConfig tableFormatConfig = opt.tableFormatConfig();
+ BlockBasedTableConfig blockBasedTableConfig;
+ if (tableFormatConfig == null) {
+ blockBasedTableConfig = new BlockBasedTableConfig();
+ } else {
+ Preconditions.checkArgument(
+ tableFormatConfig instanceof BlockBasedTableConfig,
+ "We currently only support BlockBasedTableConfig When bounding total memory.");
+ blockBasedTableConfig = (BlockBasedTableConfig) tableFormatConfig;
+ }
+ if (rocksResources.isUsingPartitionedIndexFilters()
+ && overwriteFilterIfExist(blockBasedTableConfig)) {
+ blockBasedTableConfig.setIndexType(IndexType.kTwoLevelIndexSearch);
+ blockBasedTableConfig.setPartitionFilters(true);
+ blockBasedTableConfig.setPinTopLevelIndexAndFilter(true);
+ }
+ blockBasedTableConfig.setBlockCache(blockCache);
+ blockBasedTableConfig.setCacheIndexAndFilterBlocks(true);
+ blockBasedTableConfig.setCacheIndexAndFilterBlocksWithHighPriority(true);
+ blockBasedTableConfig.setPinL0FilterAndIndexBlocksInCache(true);
+ opt.setTableFormatConfig(blockBasedTableConfig);
+ }
+
+ return opt;
+ }
+
+ /** Gets the ForSt {@link WriteOptions} to be used for write operations. */
+ public WriteOptions getWriteOptions() {
+ // Disable WAL by default
+ WriteOptions opt = new WriteOptions().setDisableWAL(true);
+ handlesToClose.add(opt);
+
+ // add user-defined options factory, if specified
+ if (optionsFactory != null) {
+ opt = optionsFactory.createWriteOptions(opt, handlesToClose);
+ }
+
+ return opt;
+ }
+
+ /** Gets the ForSt {@link ReadOptions} to be used for read operations. */
+ public ReadOptions getReadOptions() {
+ ReadOptions opt = new ReadOptions();
+ handlesToClose.add(opt);
+
+ // add user-defined options factory, if specified
+ if (optionsFactory != null) {
+ opt = optionsFactory.createReadOptions(opt, handlesToClose);
+ }
+
+ return opt;
+ }
+
+ @Nullable
+ public File getLocalBasePath() {
+ return localBasePath;
+ }
+
+ @Nullable
+ public File getLocalForStPath() {
+ return localForStPath;
+ }
+
+ ForStNativeMetricOptions getMemoryWatcherOptions(
+ ForStNativeMetricOptions defaultMetricOptions) {
+ return optionsFactory == null
+ ? defaultMetricOptions
+ : optionsFactory.createNativeMetricsOptions(defaultMetricOptions);
+ }
+
+ @Override
+ public void close() throws Exception {
+ handlesToClose.forEach(IOUtils::closeQuietly);
+ handlesToClose.clear();
+
+ if (sharedResources != null) {
+ sharedResources.close();
+ }
+ cleanRelocatedDbLogs();
+ }
+
+ /**
+ * Overwrite configured {@link Filter} if enable partitioned filter. Partitioned filter only
+ * worked in full bloom filter, not blocked based.
+ */
+ private boolean overwriteFilterIfExist(BlockBasedTableConfig blockBasedTableConfig) {
+ if (blockBasedTableConfig.filterPolicy() != null) {
+ // TODO Can get filter's config in the future ForSt version, and build new filter use
+ // existing config.
+ BloomFilter newFilter = new BloomFilter(10, false);
+ LOG.info(
+ "Existing filter has been overwritten to full filters since partitioned index filters is enabled.");
+ blockBasedTableConfig.setFilterPolicy(newFilter);
+ handlesToClose.add(newFilter);
+ }
+ return true;
+ }
+
+ /** Create a {@link DBOptions} for ForSt, including some common settings. */
+ DBOptions createBaseCommonDBOptions() {
+ return new DBOptions().setUseFsync(false).setStatsDumpPeriodSec(0);
+ }
+
+ /** Create a {@link ColumnFamilyOptions} for ForSt, including some common settings. */
+ ColumnFamilyOptions createBaseCommonColumnOptions() {
+ return new ColumnFamilyOptions();
+ }
+
+ /**
+ * Get a value for option from pre-defined option and configurable option settings. The priority
+ * relationship is as below.
+ *
+ * <p>Configured value > pre-defined value > default value.
+ *
+ * @param option the wanted option
+ * @param <T> the value type
+ * @return the final value for the option according to the priority above.
+ */
+ @Nullable
+ private <T> T internalGetOption(ConfigOption<T> option) {
+ return configuration.getOptional(option).orElseGet(option::defaultValue);
+ }
+
+ @SuppressWarnings("ConstantConditions")
+ private DBOptions setDBOptionsFromConfigurableOptions(DBOptions currentOptions) {
+
+ currentOptions.setMaxBackgroundJobs(
+ internalGetOption(ForStConfigurableOptions.MAX_BACKGROUND_THREADS));
+
+ currentOptions.setMaxOpenFiles(internalGetOption(ForStConfigurableOptions.MAX_OPEN_FILES));
+
+ currentOptions.setInfoLogLevel(internalGetOption(ForStConfigurableOptions.LOG_LEVEL));
+
+ String logDir = internalGetOption(ForStConfigurableOptions.LOG_DIR);
+ if (logDir == null || logDir.isEmpty()) {
+ if (localForStPath == null
+ || localForStPath.getAbsolutePath().length() <= INSTANCE_PATH_LENGTH_LIMIT) {
+ relocateDefaultDbLogDir(currentOptions);
+ } else {
+ // disable log relocate when instance path length exceeds limit to prevent ForSt
+ // log file creation failure, details in FLINK-31743
+ LOG.warn(
+ "ForSt local path length exceeds limit : {}, disable log relocate.",
+ localForStPath);
+ }
+ } else {
+ currentOptions.setDbLogDir(logDir);
+ }
+
+ currentOptions.setMaxLogFileSize(
+ internalGetOption(ForStConfigurableOptions.LOG_MAX_FILE_SIZE).getBytes());
+
+ currentOptions.setKeepLogFileNum(internalGetOption(ForStConfigurableOptions.LOG_FILE_NUM));
+
+ return currentOptions;
+ }
+
+ @SuppressWarnings("ConstantConditions")
+ private ColumnFamilyOptions setColumnFamilyOptionsFromConfigurableOptions(
+ ColumnFamilyOptions currentOptions, Collection<AutoCloseable> handlesToClose) {
+
+ currentOptions.setCompactionStyle(
+ internalGetOption(ForStConfigurableOptions.COMPACTION_STYLE));
+
+ currentOptions.setCompressionPerLevel(
+ internalGetOption(ForStConfigurableOptions.COMPRESSION_PER_LEVEL));
+
+ currentOptions.setLevelCompactionDynamicLevelBytes(
+ internalGetOption(ForStConfigurableOptions.USE_DYNAMIC_LEVEL_SIZE));
+
+ currentOptions.setTargetFileSizeBase(
+ internalGetOption(ForStConfigurableOptions.TARGET_FILE_SIZE_BASE).getBytes());
+
+ currentOptions.setMaxBytesForLevelBase(
+ internalGetOption(ForStConfigurableOptions.MAX_SIZE_LEVEL_BASE).getBytes());
+
+ currentOptions.setWriteBufferSize(
+ internalGetOption(ForStConfigurableOptions.WRITE_BUFFER_SIZE).getBytes());
+
+ currentOptions.setMaxWriteBufferNumber(
+ internalGetOption(ForStConfigurableOptions.MAX_WRITE_BUFFER_NUMBER));
+
+ currentOptions.setMinWriteBufferNumberToMerge(
+ internalGetOption(ForStConfigurableOptions.MIN_WRITE_BUFFER_NUMBER_TO_MERGE));
+
+ TableFormatConfig tableFormatConfig = currentOptions.tableFormatConfig();
+
+ BlockBasedTableConfig blockBasedTableConfig;
+ if (tableFormatConfig == null) {
+ blockBasedTableConfig = new BlockBasedTableConfig();
+ } else {
+ if (tableFormatConfig instanceof PlainTableConfig) {
+ // if the table format config is PlainTableConfig, we just return current
+ // column-family options
+ return currentOptions;
+ } else {
+ blockBasedTableConfig = (BlockBasedTableConfig) tableFormatConfig;
+ }
+ }
+
+ blockBasedTableConfig.setBlockSize(
+ internalGetOption(ForStConfigurableOptions.BLOCK_SIZE).getBytes());
+
+ blockBasedTableConfig.setMetadataBlockSize(
+ internalGetOption(ForStConfigurableOptions.METADATA_BLOCK_SIZE).getBytes());
+
+ blockBasedTableConfig.setBlockCacheSize(
+ internalGetOption(ForStConfigurableOptions.BLOCK_CACHE_SIZE).getBytes());
+
+ if (internalGetOption(ForStConfigurableOptions.USE_BLOOM_FILTER)) {
+ final double bitsPerKey =
+ internalGetOption(ForStConfigurableOptions.BLOOM_FILTER_BITS_PER_KEY);
+ final boolean blockBasedMode =
+ internalGetOption(ForStConfigurableOptions.BLOOM_FILTER_BLOCK_BASED_MODE);
+ BloomFilter bloomFilter = new BloomFilter(bitsPerKey, blockBasedMode);
+ handlesToClose.add(bloomFilter);
+ blockBasedTableConfig.setFilterPolicy(bloomFilter);
+ }
+
+ return currentOptions.setTableFormatConfig(blockBasedTableConfig);
+ }
+
+ /**
+ * Relocates the default log directory of ForSt with the Flink log directory. Finds the Flink
+ * log directory using log.file Java property that is set during startup.
+ *
+ * @param dbOptions The ForSt {@link DBOptions}.
+ */
+ private void relocateDefaultDbLogDir(DBOptions dbOptions) {
+ String logFilePath = System.getProperty("log.file");
+ if (logFilePath != null) {
+ File logFile = resolveFileLocation(logFilePath);
+ if (logFile != null && resolveFileLocation(logFile.getParent()) != null) {
+ String relocatedDbLogDir = logFile.getParent();
+ this.relocatedDbLogBaseDir = new File(relocatedDbLogDir).toPath();
+ dbOptions.setDbLogDir(relocatedDbLogDir);
+ }
+ }
+ }
+
+ /**
+ * Verify log file location.
+ *
+ * @param logFilePath Path to log file
+ * @return File or null if not a valid log file
+ */
+ private File resolveFileLocation(String logFilePath) {
+ File logFile = new File(logFilePath);
+ return (logFile.exists() && logFile.canRead()) ? logFile : null;
+ }
+
+ /** Clean all relocated ForSt logs. */
+ private void cleanRelocatedDbLogs() {
+ if (localForStPath != null && relocatedDbLogBaseDir != null) {
+ LOG.info("Cleaning up relocated ForSt logs: {}.", relocatedDbLogBaseDir);
+
+ String relocatedDbLogPrefix =
+ resolveRelocatedDbLogPrefix(localForStPath.getAbsolutePath());
+ try {
+ Arrays.stream(FileUtils.listDirectory(relocatedDbLogBaseDir))
+ .filter(
+ path ->
+ !Files.isDirectory(path)
+ && path.toFile()
+ .getName()
+ .startsWith(relocatedDbLogPrefix))
+ .forEach(IOUtils::deleteFileQuietly);
+ } catch (IOException e) {
+ LOG.warn("Could not list relocated ForSt log directory: {}", relocatedDbLogBaseDir);
+ }
+ }
+ }
+
+ /**
+ * Resolve the prefix of ForSt's log file name according to ForSt's log file name rules.
+ *
+ * @param instanceForStAbsolutePath The path where the ForSt directory is located.
+ * @return Resolved ForSt log name prefix.
+ */
+ private String resolveRelocatedDbLogPrefix(String instanceForStAbsolutePath) {
+ if (!instanceForStAbsolutePath.isEmpty()
+ && !instanceForStAbsolutePath.matches("^[a-zA-Z0-9\\-._].*")) {
+ instanceForStAbsolutePath = instanceForStAbsolutePath.substring(1);
+ }
+ return instanceForStAbsolutePath.replaceAll("[^a-zA-Z0-9\\-._]", "_")
+ + FORST_RELOCATE_LOG_SUFFIX;
+ }
+}
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStSharedResources.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStSharedResources.java
new file mode 100644
index 00000000000..e6451fefda6
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStSharedResources.java
@@ -0,0 +1,73 @@
+/*
+ * 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.flink.state.forst;
+
+import org.rocksdb.Cache;
+import org.rocksdb.WriteBufferManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * The set of resources that can be shared by all ForSt instances in a slot. Sharing these resources
+ * helps ForSt a predictable resource footprint.
+ */
+final class ForStSharedResources implements AutoCloseable {
+ private static final Logger LOG = LoggerFactory.getLogger(ForStSharedResources.class);
+
+ private final Cache cache;
+
+ private final WriteBufferManager writeBufferManager;
+ private final long writeBufferManagerCapacity;
+
+ private final boolean usingPartitionedIndexFilters;
+
+ ForStSharedResources(
+ Cache cache,
+ WriteBufferManager writeBufferManager,
+ long writeBufferManagerCapacity,
+ boolean usingPartitionedIndexFilters) {
+ this.cache = cache;
+ this.writeBufferManager = writeBufferManager;
+ this.writeBufferManagerCapacity = writeBufferManagerCapacity;
+ this.usingPartitionedIndexFilters = usingPartitionedIndexFilters;
+ }
+
+ public Cache getCache() {
+ return cache;
+ }
+
+ public WriteBufferManager getWriteBufferManager() {
+ return writeBufferManager;
+ }
+
+ public long getWriteBufferManagerCapacity() {
+ return writeBufferManagerCapacity;
+ }
+
+ public boolean isUsingPartitionedIndexFilters() {
+ return usingPartitionedIndexFilters;
+ }
+
+ @Override
+ public void close() {
+ LOG.debug("Closing ForStSharedResources");
+ writeBufferManager.close();
+ cache.close();
+ }
+}
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStSharedResourcesFactory.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStSharedResourcesFactory.java
new file mode 100644
index 00000000000..beb98e71509
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStSharedResourcesFactory.java
@@ -0,0 +1,185 @@
+/*
+ * 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.flink.state.forst;
+
+import org.apache.flink.configuration.MemorySize;
+import org.apache.flink.runtime.execution.Environment;
+import org.apache.flink.runtime.memory.OpaqueMemoryResource;
+import org.apache.flink.runtime.memory.SharedResources;
+import org.apache.flink.util.function.LongFunctionWithException;
+import org.apache.flink.util.function.ThrowingRunnable;
+
+import org.slf4j.Logger;
+
+import javax.annotation.Nullable;
+
+import static org.apache.flink.state.forst.ForStOptions.FIX_PER_TM_MEMORY_SIZE;
+
+/**
+ * A factory of {@link ForStSharedResources}. Encapsulates memory share scope (e.g. TM, Slot) and
+ * lifecycle (managed/unmanaged).
+ */
+enum ForStSharedResourcesFactory {
+ /** Memory allocated per Slot (shared across slot tasks), managed by Flink. */
+ SLOT_SHARED_MANAGED(false, MemoryShareScope.SLOT) {
+ @Override
+ protected OpaqueMemoryResource<ForStSharedResources> createInternal(
+ ForStMemoryConfiguration jobMemoryConfig,
+ String resourceId,
+ Environment env,
+ double memoryFraction,
+ LongFunctionWithException<ForStSharedResources, Exception> allocator)
+ throws Exception {
+ return env.getMemoryManager()
+ .getSharedMemoryResourceForManagedMemory(resourceId, allocator, memoryFraction);
+ }
+ },
+ /** Memory allocated per Slot (shared across slot tasks), unmanaged. */
+ SLOT_SHARED_UNMANAGED(false, MemoryShareScope.SLOT) {
+ @Override
+ protected OpaqueMemoryResource<ForStSharedResources> createInternal(
+ ForStMemoryConfiguration jobMemoryConfig,
+ String resourceId,
+ Environment env,
+ double memoryFraction,
+ LongFunctionWithException<ForStSharedResources, Exception> allocator)
+ throws Exception {
+ return env.getMemoryManager()
+ .getExternalSharedMemoryResource(
+ resourceId,
+ allocator,
+ jobMemoryConfig.getFixedMemoryPerSlot().getBytes());
+ }
+ },
+ /** Memory allocated per TM (shared across all tasks), unmanaged. */
+ TM_SHARED_UNMANAGED(false, MemoryShareScope.TM) {
+ @Override
+ protected OpaqueMemoryResource<ForStSharedResources> createInternal(
+ ForStMemoryConfiguration jobMemoryConfig,
+ String resourceId,
+ Environment env,
+ double memoryFraction,
+ LongFunctionWithException<ForStSharedResources, Exception> allocator)
+ throws Exception {
+
+ SharedResources sharedResources = env.getSharedResources();
+ Object leaseHolder = new Object();
+ SharedResources.ResourceAndSize<ForStSharedResources> resource =
+ sharedResources.getOrAllocateSharedResource(
+ resourceId, leaseHolder, allocator, getTmSharedMemorySize(env));
+ ThrowingRunnable<Exception> disposer =
+ () -> sharedResources.release(resourceId, leaseHolder, unused -> {});
+
+ return new OpaqueMemoryResource<>(resource.resourceHandle(), resource.size(), disposer);
+ }
+ };
+
+ private final boolean managed;
+ private final MemoryShareScope shareScope;
+
+ ForStSharedResourcesFactory(boolean managed, MemoryShareScope shareScope) {
+ this.managed = managed;
+ this.shareScope = shareScope;
+ }
+
+ @Nullable
+ public static ForStSharedResourcesFactory from(
+ ForStMemoryConfiguration jobMemoryConfig, Environment env) {
+ if (jobMemoryConfig.isUsingFixedMemoryPerSlot()) {
+ return ForStSharedResourcesFactory.SLOT_SHARED_UNMANAGED;
+ } else if (jobMemoryConfig.isUsingManagedMemory()) {
+ return ForStSharedResourcesFactory.SLOT_SHARED_MANAGED;
+ } else if (getTmSharedMemorySize(env) > 0) {
+ return ForStSharedResourcesFactory.TM_SHARED_UNMANAGED;
+ } else {
+ // not shared and not managed - allocate per column family
+ return null;
+ }
+ }
+
+ public final OpaqueMemoryResource<ForStSharedResources> create(
+ ForStMemoryConfiguration jobMemoryConfig,
+ Environment env,
+ double memoryFraction,
+ Logger logger,
+ ForStMemoryControllerUtils.ForStMemoryFactory forStMemoryFactory)
+ throws Exception {
+ logger.info(
+ "Getting shared memory for ForSt: shareScope={}, managed={}", shareScope, managed);
+ return createInternal(
+ jobMemoryConfig,
+ managed ? MANAGED_MEMORY_RESOURCE_ID : UNMANAGED_MEMORY_RESOURCE_ID,
+ env,
+ memoryFraction,
+ createAllocator(
+ shareScope.getConfiguration(jobMemoryConfig, env), forStMemoryFactory));
+ }
+
+ protected abstract OpaqueMemoryResource<ForStSharedResources> createInternal(
+ ForStMemoryConfiguration jobMemoryConfig,
+ String resourceId,
+ Environment env,
+ double memoryFraction,
+ LongFunctionWithException<ForStSharedResources, Exception> allocator)
+ throws Exception;
+
+ private static long getTmSharedMemorySize(Environment env) {
+ return env.getTaskManagerInfo()
+ .getConfiguration()
+ .getOptional(FIX_PER_TM_MEMORY_SIZE)
+ .orElse(MemorySize.ZERO)
+ .getBytes();
+ }
+
+ private static final String MANAGED_MEMORY_RESOURCE_ID = "state-forst-managed-memory";
+
+ private static final String UNMANAGED_MEMORY_RESOURCE_ID = "state-forst-fixed-slot-memory";
+
+ private static LongFunctionWithException<ForStSharedResources, Exception> createAllocator(
+ ForStMemoryConfiguration config,
+ ForStMemoryControllerUtils.ForStMemoryFactory forStMemoryFactory) {
+ return size ->
+ ForStMemoryControllerUtils.allocateForStSharedResources(
+ size,
+ config.getWriteBufferRatio(),
+ config.getHighPriorityPoolRatio(),
+ config.isUsingPartitionedIndexFilters(),
+ forStMemoryFactory);
+ }
+}
+
+enum MemoryShareScope {
+ TM {
+ @Override
+ public ForStMemoryConfiguration getConfiguration(
+ ForStMemoryConfiguration jobMemoryConfig, Environment env) {
+ return ForStMemoryConfiguration.fromConfiguration(
+ env.getTaskManagerInfo().getConfiguration());
+ }
+ },
+ SLOT {
+ @Override
+ public ForStMemoryConfiguration getConfiguration(
+ ForStMemoryConfiguration jobMemoryConfig, Environment env) {
+ return jobMemoryConfig;
+ }
+ };
+
+ public abstract ForStMemoryConfiguration getConfiguration(
+ ForStMemoryConfiguration jobMemoryConfig, Environment env);
+}
diff --git a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStMemoryControllerUtilsTest.java b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStMemoryControllerUtilsTest.java
new file mode 100644
index 00000000000..ae9a1ae6391
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStMemoryControllerUtilsTest.java
@@ -0,0 +1,129 @@
+/*
+ * 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.flink.state.forst;
+
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.rocksdb.Cache;
+import org.rocksdb.NativeLibraryLoader;
+import org.rocksdb.WriteBufferManager;
+
+import java.io.IOException;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+/** Tests to guard {@link ForStMemoryControllerUtils}. */
+public class ForStMemoryControllerUtilsTest {
+
+ @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ @Before
+ public void ensureRocksDbNativeLibraryLoaded() throws IOException {
+ NativeLibraryLoader.getInstance()
+ .loadLibrary(temporaryFolder.newFolder().getAbsolutePath());
+ }
+
+ @Test
+ public void testCreateSharedResourcesWithExpectedCapacity() {
+ long totalMemorySize = 2048L;
+ double writeBufferRatio = 0.5;
+ double highPriPoolRatio = 0.1;
+ TestingForStMemoryFactory factory = new TestingForStMemoryFactory();
+ ForStSharedResources forStSharedResources =
+ ForStMemoryControllerUtils.allocateForStSharedResources(
+ totalMemorySize, writeBufferRatio, highPriPoolRatio, false, factory);
+ long expectedCacheCapacity =
+ ForStMemoryControllerUtils.calculateActualCacheCapacity(
+ totalMemorySize, writeBufferRatio);
+ long expectedWbmCapacity =
+ ForStMemoryControllerUtils.calculateWriteBufferManagerCapacity(
+ totalMemorySize, writeBufferRatio);
+
+ assertThat(factory.actualCacheCapacity, is(expectedCacheCapacity));
+ assertThat(factory.actualWbmCapacity, is(expectedWbmCapacity));
+ assertThat(forStSharedResources.getWriteBufferManagerCapacity(), is(expectedWbmCapacity));
+ }
+
+ @Test
+ public void testCalculateForStDefaultArenaBlockSize() {
+ final long align = 4 * 1024;
+ final long writeBufferSize = 64 * 1024 * 1024;
+ final long expectArenaBlockSize = writeBufferSize / 8;
+
+ // Normal case test
+ assertThat(
+ "Arena block size calculation error for normal case",
+ ForStMemoryControllerUtils.calculateForStDefaultArenaBlockSize(writeBufferSize),
+ is(expectArenaBlockSize));
+
+ // Alignment tests
+ assertThat(
+ "Arena block size calculation error for alignment case",
+ ForStMemoryControllerUtils.calculateForStDefaultArenaBlockSize(writeBufferSize - 1),
+ is(expectArenaBlockSize));
+ assertThat(
+ "Arena block size calculation error for alignment case2",
+ ForStMemoryControllerUtils.calculateForStDefaultArenaBlockSize(writeBufferSize + 8),
+ is(expectArenaBlockSize + align));
+ }
+
+ @Test
+ public void testCalculateForStMutableLimit() {
+ long bufferSize = 64 * 1024 * 1024;
+ long limit = bufferSize * 7 / 8;
+ assertThat(ForStMemoryControllerUtils.calculateForStMutableLimit(bufferSize), is(limit));
+ }
+
+ @Test
+ public void testValidateArenaBlockSize() {
+ long arenaBlockSize = 8 * 1024 * 1024;
+ assertFalse(
+ ForStMemoryControllerUtils.validateArenaBlockSize(
+ arenaBlockSize, (long) (arenaBlockSize * 0.5)));
+ assertTrue(
+ ForStMemoryControllerUtils.validateArenaBlockSize(
+ arenaBlockSize, (long) (arenaBlockSize * 1.5)));
+ }
+
+ private static final class TestingForStMemoryFactory
+ implements ForStMemoryControllerUtils.ForStMemoryFactory {
+ private Long actualCacheCapacity = null;
+ private Long actualWbmCapacity = null;
+
+ @Override
+ public Cache createCache(long cacheCapacity, double highPriorityPoolRatio) {
+ actualCacheCapacity = cacheCapacity;
+ return ForStMemoryControllerUtils.ForStMemoryFactory.DEFAULT.createCache(
+ cacheCapacity, highPriorityPoolRatio);
+ }
+
+ @Override
+ public WriteBufferManager createWriteBufferManager(
+ long writeBufferManagerCapacity, Cache cache) {
+ actualWbmCapacity = writeBufferManagerCapacity;
+ return ForStMemoryControllerUtils.ForStMemoryFactory.DEFAULT.createWriteBufferManager(
+ writeBufferManagerCapacity, cache);
+ }
+ }
+}
diff --git a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStResourceContainerTest.java b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStResourceContainerTest.java
new file mode 100644
index 00000000000..53c0524308b
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStResourceContainerTest.java
@@ -0,0 +1,297 @@
+/*
+ * 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.flink.state.forst;
+
+import org.apache.flink.runtime.memory.OpaqueMemoryResource;
+import org.apache.flink.util.function.ThrowingRunnable;
+
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.rocksdb.BlockBasedTableConfig;
+import org.rocksdb.BloomFilter;
+import org.rocksdb.Cache;
+import org.rocksdb.ColumnFamilyOptions;
+import org.rocksdb.DBOptions;
+import org.rocksdb.IndexType;
+import org.rocksdb.LRUCache;
+import org.rocksdb.NativeLibraryLoader;
+import org.rocksdb.ReadOptions;
+import org.rocksdb.TableFormatConfig;
+import org.rocksdb.WriteBufferManager;
+import org.rocksdb.WriteOptions;
+
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashSet;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.CoreMatchers.not;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.fail;
+
+/** Tests to guard {@link ForStResourceContainer}. */
+public class ForStResourceContainerTest {
+
+ @ClassRule public static final TemporaryFolder TMP_FOLDER = new TemporaryFolder();
+
+ @BeforeClass
+ public static void ensureForStNativeLibraryLoaded() throws IOException {
+ NativeLibraryLoader.getInstance().loadLibrary(TMP_FOLDER.newFolder().getAbsolutePath());
+ }
+
+ // ------------------------------------------------------------------------
+
+ @Test
+ public void testFreeDBOptionsAfterClose() throws Exception {
+ ForStResourceContainer container = new ForStResourceContainer();
+ DBOptions dbOptions = container.getDbOptions();
+ assertThat(dbOptions.isOwningHandle(), is(true));
+ container.close();
+ assertThat(dbOptions.isOwningHandle(), is(false));
+ }
+
+ @Test
+ public void testFreeMultipleDBOptionsAfterClose() throws Exception {
+ ForStResourceContainer container = new ForStResourceContainer();
+ final int optionNumber = 20;
+ ArrayList<DBOptions> dbOptions = new ArrayList<>(optionNumber);
+ for (int i = 0; i < optionNumber; i++) {
+ dbOptions.add(container.getDbOptions());
+ }
+ container.close();
+ for (DBOptions dbOption : dbOptions) {
+ assertThat(dbOption.isOwningHandle(), is(false));
+ }
+ }
+
+ /**
+ * Guard the shared resources will be released after {@link ForStResourceContainer#close()} when
+ * the {@link ForStResourceContainer} instance is initiated with {@link OpaqueMemoryResource}.
+ *
+ * @throws Exception if unexpected error happened.
+ */
+ @Test
+ public void testSharedResourcesAfterClose() throws Exception {
+ OpaqueMemoryResource<ForStSharedResources> sharedResources = getSharedResources();
+ ForStResourceContainer container = new ForStResourceContainer(null, sharedResources);
+ container.close();
+ ForStSharedResources forStSharedResources = sharedResources.getResourceHandle();
+ assertThat(forStSharedResources.getCache().isOwningHandle(), is(false));
+ assertThat(forStSharedResources.getWriteBufferManager().isOwningHandle(), is(false));
+ }
+
+ /**
+ * Guard that {@link ForStResourceContainer#getDbOptions()} shares the same {@link
+ * WriteBufferManager} instance if the {@link ForStResourceContainer} instance is initiated with
+ * {@link OpaqueMemoryResource}.
+ *
+ * @throws Exception if unexpected error happened.
+ */
+ @Test
+ public void testGetDbOptionsWithSharedResources() throws Exception {
+ final int optionNumber = 20;
+ OpaqueMemoryResource<ForStSharedResources> sharedResources = getSharedResources();
+ ForStResourceContainer container = new ForStResourceContainer(null, sharedResources);
+ HashSet<WriteBufferManager> writeBufferManagers = new HashSet<>();
+ for (int i = 0; i < optionNumber; i++) {
+ DBOptions dbOptions = container.getDbOptions();
+ WriteBufferManager writeBufferManager = getWriteBufferManager(dbOptions);
+ writeBufferManagers.add(writeBufferManager);
+ }
+ assertThat(writeBufferManagers.size(), is(1));
+ assertThat(
+ writeBufferManagers.iterator().next(),
+ is(sharedResources.getResourceHandle().getWriteBufferManager()));
+ container.close();
+ }
+
+ /**
+ * Guard that {@link ForStResourceContainer#getColumnOptions()} shares the same {@link Cache}
+ * instance if the {@link ForStResourceContainer} instance is initiated with {@link
+ * OpaqueMemoryResource}.
+ *
+ * @throws Exception if unexpected error happened.
+ */
+ @Test
+ public void testGetColumnFamilyOptionsWithSharedResources() throws Exception {
+ final int optionNumber = 20;
+ OpaqueMemoryResource<ForStSharedResources> sharedResources = getSharedResources();
+ ForStResourceContainer container = new ForStResourceContainer(null, sharedResources);
+ HashSet<Cache> caches = new HashSet<>();
+ for (int i = 0; i < optionNumber; i++) {
+ ColumnFamilyOptions columnOptions = container.getColumnOptions();
+ Cache cache = getBlockCache(columnOptions);
+ caches.add(cache);
+ }
+ assertThat(caches.size(), is(1));
+ assertThat(caches.iterator().next(), is(sharedResources.getResourceHandle().getCache()));
+ container.close();
+ }
+
+ private OpaqueMemoryResource<ForStSharedResources> getSharedResources() {
+ final long cacheSize = 1024L, writeBufferSize = 512L;
+ final LRUCache cache = new LRUCache(cacheSize, -1, false, 0.1);
+ final WriteBufferManager wbm = new WriteBufferManager(writeBufferSize, cache);
+ ForStSharedResources forStSharedResources =
+ new ForStSharedResources(cache, wbm, writeBufferSize, false);
+ return new OpaqueMemoryResource<>(
+ forStSharedResources, cacheSize, forStSharedResources::close);
+ }
+
+ private Cache getBlockCache(ColumnFamilyOptions columnOptions) {
+ BlockBasedTableConfig blockBasedTableConfig = null;
+ try {
+ blockBasedTableConfig = (BlockBasedTableConfig) columnOptions.tableFormatConfig();
+ } catch (ClassCastException e) {
+ fail("Table config got from ColumnFamilyOptions is not BlockBasedTableConfig");
+ }
+ Field cacheField = null;
+ try {
+ cacheField = BlockBasedTableConfig.class.getDeclaredField("blockCache");
+ } catch (NoSuchFieldException e) {
+ fail("blockCache is not defined");
+ }
+ cacheField.setAccessible(true);
+ try {
+ return (Cache) cacheField.get(blockBasedTableConfig);
+ } catch (IllegalAccessException e) {
+ fail("Cannot access blockCache field.");
+ return null;
+ }
+ }
+
+ private WriteBufferManager getWriteBufferManager(DBOptions dbOptions) {
+
+ Field writeBufferManagerField = null;
+ try {
+ writeBufferManagerField = DBOptions.class.getDeclaredField("writeBufferManager_");
+ } catch (NoSuchFieldException e) {
+ fail("writeBufferManager_ is not defined.");
+ }
+ writeBufferManagerField.setAccessible(true);
+ try {
+ return (WriteBufferManager) writeBufferManagerField.get(dbOptions);
+ } catch (IllegalAccessException e) {
+ fail("Cannot access writeBufferManager_ field.");
+ return null;
+ }
+ }
+
+ @Test
+ public void testFreeColumnOptionsAfterClose() throws Exception {
+ ForStResourceContainer container = new ForStResourceContainer();
+ ColumnFamilyOptions columnFamilyOptions = container.getColumnOptions();
+ assertThat(columnFamilyOptions.isOwningHandle(), is(true));
+ container.close();
+ assertThat(columnFamilyOptions.isOwningHandle(), is(false));
+ }
+
+ @Test
+ public void testFreeMultipleColumnOptionsAfterClose() throws Exception {
+ ForStResourceContainer container = new ForStResourceContainer();
+ final int optionNumber = 20;
+ ArrayList<ColumnFamilyOptions> columnFamilyOptions = new ArrayList<>(optionNumber);
+ for (int i = 0; i < optionNumber; i++) {
+ columnFamilyOptions.add(container.getColumnOptions());
+ }
+ container.close();
+ for (ColumnFamilyOptions columnFamilyOption : columnFamilyOptions) {
+ assertThat(columnFamilyOption.isOwningHandle(), is(false));
+ }
+ }
+
+ @Test
+ public void testFreeSharedResourcesAfterClose() throws Exception {
+ LRUCache cache = new LRUCache(1024L);
+ WriteBufferManager wbm = new WriteBufferManager(1024L, cache);
+ ForStSharedResources sharedResources = new ForStSharedResources(cache, wbm, 1024L, false);
+ final ThrowingRunnable<Exception> disposer = sharedResources::close;
+ OpaqueMemoryResource<ForStSharedResources> opaqueResource =
+ new OpaqueMemoryResource<>(sharedResources, 1024L, disposer);
+
+ ForStResourceContainer container = new ForStResourceContainer(null, opaqueResource);
+
+ container.close();
+ assertThat(cache.isOwningHandle(), is(false));
+ assertThat(wbm.isOwningHandle(), is(false));
+ }
+
+ @Test
+ public void testFreeWriteReadOptionsAfterClose() throws Exception {
+ ForStResourceContainer container = new ForStResourceContainer();
+ WriteOptions writeOptions = container.getWriteOptions();
+ ReadOptions readOptions = container.getReadOptions();
+ assertThat(writeOptions.isOwningHandle(), is(true));
+ assertThat(readOptions.isOwningHandle(), is(true));
+ container.close();
+ assertThat(writeOptions.isOwningHandle(), is(false));
+ assertThat(readOptions.isOwningHandle(), is(false));
+ }
+
+ @Test
+ public void testGetColumnFamilyOptionsWithPartitionedIndex() throws Exception {
+ LRUCache cache = new LRUCache(1024L);
+ WriteBufferManager wbm = new WriteBufferManager(1024L, cache);
+ ForStSharedResources sharedResources = new ForStSharedResources(cache, wbm, 1024L, true);
+ final ThrowingRunnable<Exception> disposer = sharedResources::close;
+ OpaqueMemoryResource<ForStSharedResources> opaqueResource =
+ new OpaqueMemoryResource<>(sharedResources, 1024L, disposer);
+ BloomFilter blockBasedFilter = new BloomFilter();
+ ForStOptionsFactory blockBasedBloomFilterOptionFactory =
+ new ForStOptionsFactory() {
+
+ @Override
+ public DBOptions createDBOptions(
+ DBOptions currentOptions, Collection<AutoCloseable> handlesToClose) {
+ return currentOptions;
+ }
+
+ @Override
+ public ColumnFamilyOptions createColumnOptions(
+ ColumnFamilyOptions currentOptions,
+ Collection<AutoCloseable> handlesToClose) {
+ TableFormatConfig tableFormatConfig = currentOptions.tableFormatConfig();
+ BlockBasedTableConfig blockBasedTableConfig =
+ tableFormatConfig == null
+ ? new BlockBasedTableConfig()
+ : (BlockBasedTableConfig) tableFormatConfig;
+ blockBasedTableConfig.setFilter(blockBasedFilter);
+ handlesToClose.add(blockBasedFilter);
+ currentOptions.setTableFormatConfig(blockBasedTableConfig);
+ return currentOptions;
+ }
+ };
+ try (ForStResourceContainer container =
+ new ForStResourceContainer(blockBasedBloomFilterOptionFactory, opaqueResource)) {
+ ColumnFamilyOptions columnOptions = container.getColumnOptions();
+ BlockBasedTableConfig actual =
+ (BlockBasedTableConfig) columnOptions.tableFormatConfig();
+ assertThat(actual.indexType(), is(IndexType.kTwoLevelIndexSearch));
+ assertThat(actual.partitionFilters(), is(true));
+ assertThat(actual.pinTopLevelIndexAndFilter(), is(true));
+ assertThat(actual.filterPolicy(), not(blockBasedFilter));
+ }
+ assertFalse("Block based filter is left unclosed.", blockBasedFilter.isOwningHandle());
+ }
+}
(flink) 03/07: [FLINK-35047][state] Introduce Metrics for ForStStateBackend
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hangxiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 8411367ab0a213ec55c679a496352853bf96af51
Author: Hangxiang Yu <ma...@gmail.com>
AuthorDate: Mon Apr 15 21:25:47 2024 +0800
[FLINK-35047][state] Introduce Metrics for ForStStateBackend
---
.../state/forst/ForStNativeMetricMonitor.java | 233 +++++++
.../state/forst/ForStNativeMetricOptions.java | 700 +++++++++++++++++++++
.../flink/state/forst/ForStOptionsFactory.java | 14 +-
.../apache/flink/state/forst/ForStProperty.java | 115 ++++
.../apache/flink/state/forst/ForStExtension.java | 212 +++++++
.../state/forst/ForStNativeMetricMonitorTest.java | 254 ++++++++
.../state/forst/ForStNativeMetricOptionsTest.java | 53 ++
.../flink/state/forst/ForStPropertyTest.java | 47 ++
8 files changed, 1627 insertions(+), 1 deletion(-)
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStNativeMetricMonitor.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStNativeMetricMonitor.java
new file mode 100644
index 00000000000..71b511853d5
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStNativeMetricMonitor.java
@@ -0,0 +1,233 @@
+/*
+ * 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.flink.state.forst;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.metrics.Gauge;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.View;
+
+import org.rocksdb.ColumnFamilyHandle;
+import org.rocksdb.RocksDB;
+import org.rocksdb.Statistics;
+import org.rocksdb.TickerType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.GuardedBy;
+
+import java.io.Closeable;
+import java.math.BigInteger;
+
+/**
+ * A monitor which pulls {{@link RocksDB}} native metrics and forwards them to Flink's metric group.
+ * All metrics are unsigned longs and are reported at the column family level.
+ */
+@Internal
+public class ForStNativeMetricMonitor implements Closeable {
+ private static final Logger LOG = LoggerFactory.getLogger(ForStNativeMetricMonitor.class);
+
+ private final ForStNativeMetricOptions options;
+
+ private final MetricGroup metricGroup;
+
+ private final Object lock;
+
+ static final String COLUMN_FAMILY_KEY = "column_family";
+
+ @GuardedBy("lock")
+ private RocksDB rocksDB;
+
+ @Nullable
+ @GuardedBy("lock")
+ private Statistics statistics;
+
+ public ForStNativeMetricMonitor(
+ @Nonnull ForStNativeMetricOptions options,
+ @Nonnull MetricGroup metricGroup,
+ @Nonnull RocksDB rocksDB,
+ @Nullable Statistics statistics) {
+ this.options = options;
+ this.metricGroup = metricGroup;
+ this.rocksDB = rocksDB;
+ this.statistics = statistics;
+ this.lock = new Object();
+ registerStatistics();
+ }
+
+ /** Register gauges to pull native metrics for the database. */
+ private void registerStatistics() {
+ if (statistics != null) {
+ for (TickerType tickerType : options.getMonitorTickerTypes()) {
+ metricGroup.gauge(
+ String.format("rocksdb.%s", tickerType.name().toLowerCase()),
+ new ForStNativeStatisticsMetricView(tickerType));
+ }
+ }
+ }
+
+ /**
+ * Register gauges to pull native metrics for the column family.
+ *
+ * @param columnFamilyName group name for the new gauges
+ * @param handle native handle to the column family
+ */
+ void registerColumnFamily(String columnFamilyName, ColumnFamilyHandle handle) {
+
+ boolean columnFamilyAsVariable = options.isColumnFamilyAsVariable();
+ MetricGroup group =
+ columnFamilyAsVariable
+ ? metricGroup.addGroup(COLUMN_FAMILY_KEY, columnFamilyName)
+ : metricGroup.addGroup(columnFamilyName);
+
+ for (ForStProperty property : options.getProperties()) {
+ ForStNativePropertyMetricView gauge =
+ new ForStNativePropertyMetricView(handle, property);
+ group.gauge(property.getForStProperty(), gauge);
+ }
+ }
+
+ /** Updates the value of metricView if the reference is still valid. */
+ private void setProperty(ForStNativePropertyMetricView metricView) {
+ if (metricView.isClosed()) {
+ return;
+ }
+ try {
+ synchronized (lock) {
+ if (rocksDB != null) {
+ long value =
+ metricView.property.getNumericalPropertyValue(
+ rocksDB, metricView.handle);
+ metricView.setValue(value);
+ }
+ }
+ } catch (Exception e) {
+ metricView.close();
+ LOG.warn("Failed to read native metric {} from ForSt.", metricView.property, e);
+ }
+ }
+
+ private void setStatistics(ForStNativeStatisticsMetricView metricView) {
+ if (metricView.isClosed()) {
+ return;
+ }
+ if (statistics != null) {
+ synchronized (lock) {
+ metricView.setValue(statistics.getTickerCount(metricView.tickerType));
+ }
+ }
+ }
+
+ @Override
+ public void close() {
+ synchronized (lock) {
+ rocksDB = null;
+ statistics = null;
+ }
+ }
+
+ abstract static class ForStNativeView implements View {
+ private boolean closed;
+
+ ForStNativeView() {
+ this.closed = false;
+ }
+
+ void close() {
+ closed = true;
+ }
+
+ boolean isClosed() {
+ return closed;
+ }
+ }
+
+ /**
+ * A gauge which periodically pulls a ForSt property-based native metric for the specified
+ * column family / metric pair.
+ *
+ * <p><strong>Note</strong>: As the returned property is of type {@code uint64_t} on C++ side
+ * the returning value can be negative. Because java does not support unsigned long types, this
+ * gauge wraps the result in a {@link BigInteger}.
+ */
+ class ForStNativePropertyMetricView extends ForStNativeView implements Gauge<BigInteger> {
+ private final ForStProperty property;
+
+ private final ColumnFamilyHandle handle;
+
+ private BigInteger bigInteger;
+
+ private ForStNativePropertyMetricView(
+ ColumnFamilyHandle handle, @Nonnull ForStProperty property) {
+ this.handle = handle;
+ this.property = property;
+ this.bigInteger = BigInteger.ZERO;
+ }
+
+ public void setValue(long value) {
+ if (value >= 0L) {
+ bigInteger = BigInteger.valueOf(value);
+ } else {
+ int upper = (int) (value >>> 32);
+ int lower = (int) value;
+
+ bigInteger =
+ BigInteger.valueOf(Integer.toUnsignedLong(upper))
+ .shiftLeft(32)
+ .add(BigInteger.valueOf(Integer.toUnsignedLong(lower)));
+ }
+ }
+
+ @Override
+ public BigInteger getValue() {
+ return bigInteger;
+ }
+
+ @Override
+ public void update() {
+ setProperty(this);
+ }
+ }
+
+ /** A gauge which periodically pulls a ForSt statistics-based native metric for the database. */
+ class ForStNativeStatisticsMetricView extends ForStNativeView implements Gauge<Long> {
+ private final TickerType tickerType;
+ private long value;
+
+ private ForStNativeStatisticsMetricView(TickerType tickerType) {
+ this.tickerType = tickerType;
+ }
+
+ @Override
+ public Long getValue() {
+ return value;
+ }
+
+ void setValue(long value) {
+ this.value = value;
+ }
+
+ @Override
+ public void update() {
+ setStatistics(this);
+ }
+ }
+}
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStNativeMetricOptions.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStNativeMetricOptions.java
new file mode 100644
index 00000000000..40a973b61ef
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStNativeMetricOptions.java
@@ -0,0 +1,700 @@
+/*
+ * 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.flink.state.forst;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.ReadableConfig;
+
+import org.rocksdb.TickerType;
+
+import java.io.Serializable;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Enable which ForSt metrics to forward to Flink's metrics reporter.
+ *
+ * <p>Property based metrics would report at the column family level and return unsigned long
+ * values.
+ *
+ * <p>Statistics based metrics would report at the database level, it can return ticker or histogram
+ * kind results.
+ *
+ * <p>Properties and doc comments are taken from RocksDB documentation. See <a
+ * href="https://github.com/facebook/rocksdb/blob/64324e329eb0a9b4e77241a425a1615ff524c7f1/include/rocksdb/db.h#L429">
+ * db.h</a> for more information.
+ */
+@Experimental
+public class ForStNativeMetricOptions implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ // --------------------------------------------------------------------------------------------
+ // RocksDB property based metrics, report at column family level
+ // --------------------------------------------------------------------------------------------
+
+ public static final String METRICS_COLUMN_FAMILY_AS_VARIABLE_KEY =
+ "state.backend.rocksdb.metrics" + ".column-family-as-variable";
+
+ public static final ConfigOption<Boolean> MONITOR_NUM_IMMUTABLE_MEM_TABLES =
+ ConfigOptions.key(ForStProperty.NumImmutableMemTable.getConfigKey())
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Monitor the number of immutable memtables in RocksDB.");
+
+ public static final ConfigOption<Boolean> MONITOR_MEM_TABLE_FLUSH_PENDING =
+ ConfigOptions.key(ForStProperty.MemTableFlushPending.getConfigKey())
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Monitor the number of pending memtable flushes in RocksDB.");
+
+ public static final ConfigOption<Boolean> TRACK_COMPACTION_PENDING =
+ ConfigOptions.key(ForStProperty.CompactionPending.getConfigKey())
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Track pending compactions in RocksDB. Returns 1 if a compaction is pending, 0 otherwise.");
+
+ public static final ConfigOption<Boolean> MONITOR_BACKGROUND_ERRORS =
+ ConfigOptions.key(ForStProperty.BackgroundErrors.getConfigKey())
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Monitor the number of background errors in RocksDB.");
+
+ public static final ConfigOption<Boolean> MONITOR_CUR_SIZE_ACTIVE_MEM_TABLE =
+ ConfigOptions.key(ForStProperty.CurSizeActiveMemTable.getConfigKey())
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Monitor the approximate size of the active memtable in bytes.");
+
+ public static final ConfigOption<Boolean> MONITOR_CUR_SIZE_ALL_MEM_TABLE =
+ ConfigOptions.key(ForStProperty.CurSizeAllMemTables.getConfigKey())
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Monitor the approximate size of the active and unflushed immutable memtables"
+ + " in bytes.");
+
+ public static final ConfigOption<Boolean> MONITOR_SIZE_ALL_MEM_TABLES =
+ ConfigOptions.key(ForStProperty.SizeAllMemTables.getConfigKey())
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Monitor the approximate size of the active, unflushed immutable, "
+ + "and pinned immutable memtables in bytes.");
+
+ public static final ConfigOption<Boolean> MONITOR_NUM_ENTRIES_ACTIVE_MEM_TABLE =
+ ConfigOptions.key(ForStProperty.NumEntriesActiveMemTable.getConfigKey())
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Monitor the total number of entries in the active memtable.");
+
+ public static final ConfigOption<Boolean> MONITOR_NUM_ENTRIES_IMM_MEM_TABLES =
+ ConfigOptions.key(ForStProperty.NumEntriesImmMemTables.getConfigKey())
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Monitor the total number of entries in the unflushed immutable memtables.");
+
+ public static final ConfigOption<Boolean> MONITOR_NUM_DELETES_ACTIVE_MEM_TABLE =
+ ConfigOptions.key(ForStProperty.NumDeletesActiveMemTable.getConfigKey())
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Monitor the total number of delete entries in the active memtable.");
+
+ public static final ConfigOption<Boolean> MONITOR_NUM_DELETES_IMM_MEM_TABLE =
+ ConfigOptions.key(ForStProperty.NumDeletesImmMemTables.getConfigKey())
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Monitor the total number of delete entries in the unflushed immutable memtables.");
+
+ public static final ConfigOption<Boolean> ESTIMATE_NUM_KEYS =
+ ConfigOptions.key(ForStProperty.EstimateNumKeys.getConfigKey())
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Estimate the number of keys in RocksDB.");
+
+ public static final ConfigOption<Boolean> ESTIMATE_TABLE_READERS_MEM =
+ ConfigOptions.key(ForStProperty.EstimateTableReadersMem.getConfigKey())
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Estimate the memory used for reading SST tables, excluding memory"
+ + " used in block cache (e.g.,filter and index blocks) in bytes.");
+
+ public static final ConfigOption<Boolean> MONITOR_NUM_SNAPSHOTS =
+ ConfigOptions.key(ForStProperty.NumSnapshots.getConfigKey())
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Monitor the number of unreleased snapshots of the database.");
+
+ public static final ConfigOption<Boolean> MONITOR_NUM_LIVE_VERSIONS =
+ ConfigOptions.key(ForStProperty.NumLiveVersions.getConfigKey())
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Monitor number of live versions. Version is an internal data structure. "
+ + "See RocksDB file version_set.h for details. More live versions often mean more SST files are held "
+ + "from being deleted, by iterators or unfinished compactions.");
+
+ public static final ConfigOption<Boolean> ESTIMATE_LIVE_DATA_SIZE =
+ ConfigOptions.key(ForStProperty.EstimateLiveDataSize.getConfigKey())
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Estimate of the amount of live data in bytes (usually smaller than sst files size due to space amplification).");
+
+ public static final ConfigOption<Boolean> MONITOR_TOTAL_SST_FILES_SIZE =
+ ConfigOptions.key(ForStProperty.TotalSstFilesSize.getConfigKey())
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Monitor the total size (bytes) of all SST files of all versions."
+ + "WARNING: may slow down online queries if there are too many files.");
+
+ public static final ConfigOption<Boolean> MONITOR_LIVE_SST_FILES_SIZE =
+ ConfigOptions.key(ForStProperty.LiveSstFilesSize.getConfigKey())
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Monitor the total size (bytes) of all SST files belonging to the latest version."
+ + "WARNING: may slow down online queries if there are too many files.");
+
+ public static final ConfigOption<Boolean> ESTIMATE_PENDING_COMPACTION_BYTES =
+ ConfigOptions.key(ForStProperty.EstimatePendingCompactionBytes.getConfigKey())
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Estimated total number of bytes compaction needs to rewrite to get all levels "
+ + "down to under target size. Not valid for other compactions than level-based.");
+
+ public static final ConfigOption<Boolean> MONITOR_NUM_RUNNING_COMPACTIONS =
+ ConfigOptions.key(ForStProperty.NumRunningCompactions.getConfigKey())
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Monitor the number of currently running compactions.");
+
+ public static final ConfigOption<Boolean> MONITOR_NUM_RUNNING_FLUSHES =
+ ConfigOptions.key(ForStProperty.NumRunningFlushes.getConfigKey())
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Monitor the number of currently running flushes.");
+
+ public static final ConfigOption<Boolean> MONITOR_ACTUAL_DELAYED_WRITE_RATE =
+ ConfigOptions.key(ForStProperty.ActualDelayedWriteRate.getConfigKey())
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Monitor the current actual delayed write rate. 0 means no delay.");
+
+ public static final ConfigOption<Boolean> IS_WRITE_STOPPED =
+ ConfigOptions.key(ForStProperty.IsWriteStopped.getConfigKey())
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Track whether write has been stopped in RocksDB. Returns 1 if write has been stopped, 0 otherwise.");
+
+ public static final ConfigOption<Boolean> BLOCK_CACHE_CAPACITY =
+ ConfigOptions.key(ForStProperty.BlockCacheCapacity.getConfigKey())
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Monitor block cache capacity.");
+
+ public static final ConfigOption<Boolean> BLOCK_CACHE_USAGE =
+ ConfigOptions.key(ForStProperty.BlockCacheUsage.getConfigKey())
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Monitor the memory size for the entries residing in block cache.");
+
+ public static final ConfigOption<Boolean> BLOCK_CACHE_PINNED_USAGE =
+ ConfigOptions.key(ForStProperty.BlockCachePinnedUsage.getConfigKey())
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Monitor the memory size for the entries being pinned in block cache.");
+
+ public static final ConfigOption<Boolean> COLUMN_FAMILY_AS_VARIABLE =
+ ConfigOptions.key(METRICS_COLUMN_FAMILY_AS_VARIABLE_KEY)
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Whether to expose the column family as a variable for RocksDB property based metrics.");
+
+ public static final ConfigOption<Boolean> MONITOR_NUM_FILES_AT_LEVEL =
+ ConfigOptions.key("state.backend.rocksdb.metrics.num-files-at-level")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Monitor the number of files at each level.");
+
+ // --------------------------------------------------------------------------------------------
+ // RocksDB statistics based metrics, report at database level
+ // --------------------------------------------------------------------------------------------
+
+ public static final ConfigOption<Boolean> MONITOR_BLOCK_CACHE_HIT =
+ ConfigOptions.key("state.backend.rocksdb.metrics.block-cache-hit")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Monitor the total count of block cache hit in RocksDB (BLOCK_CACHE_HIT == BLOCK_CACHE_INDEX_HIT + BLOCK_CACHE_FILTER_HIT + BLOCK_CACHE_DATA_HIT).");
+
+ public static final ConfigOption<Boolean> MONITOR_BLOCK_CACHE_MISS =
+ ConfigOptions.key("state.backend.rocksdb.metrics.block-cache-miss")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Monitor the total count of block cache misses in RocksDB (BLOCK_CACHE_MISS == BLOCK_CACHE_INDEX_MISS + BLOCK_CACHE_FILTER_MISS + BLOCK_CACHE_DATA_MISS).");
+
+ public static final ConfigOption<Boolean> MONITOR_BLOOM_FILTER_USEFUL =
+ ConfigOptions.key("state.backend.rocksdb.metrics.bloom-filter-useful")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Monitor the total count of reads avoided by bloom filter.");
+
+ public static final ConfigOption<Boolean> MONITOR_BLOOM_FILTER_FULL_POSITIVE =
+ ConfigOptions.key("state.backend.rocksdb.metrics.bloom-filter-full-positive")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Monitor the total count of reads not avoided by bloom full filter.");
+
+ public static final ConfigOption<Boolean> MONITOR_BLOOM_FILTER_FULL_TRUE_POSITIVE =
+ ConfigOptions.key("state.backend.rocksdb.metrics.bloom-filter-full-true-positive")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Monitor the total count of reads not avoided by bloom full filter and the data actually exists in RocksDB.");
+
+ public static final ConfigOption<Boolean> MONITOR_BYTES_READ =
+ ConfigOptions.key("state.backend.rocksdb.metrics.bytes-read")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Monitor the number of uncompressed bytes read (from memtables/cache/sst) from Get() operation in RocksDB.");
+
+ public static final ConfigOption<Boolean> MONITOR_ITER_BYTES_READ =
+ ConfigOptions.key("state.backend.rocksdb.metrics.iter-bytes-read")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Monitor the number of uncompressed bytes read (from memtables/cache/sst) from an iterator operation in RocksDB.");
+
+ public static final ConfigOption<Boolean> MONITOR_BYTES_WRITTEN =
+ ConfigOptions.key("state.backend.rocksdb.metrics.bytes-written")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Monitor the number of uncompressed bytes written by DB::{Put(), Delete(), Merge(), Write()} operations, which does not include the compaction written bytes, in RocksDB.");
+
+ public static final ConfigOption<Boolean> MONITOR_COMPACTION_READ_BYTES =
+ ConfigOptions.key("state.backend.rocksdb.metrics.compaction-read-bytes")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Monitor the bytes read during compaction in RocksDB.");
+
+ public static final ConfigOption<Boolean> MONITOR_COMPACTION_WRITE_BYTES =
+ ConfigOptions.key("state.backend.rocksdb.metrics.compaction-write-bytes")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Monitor the bytes written during compaction in RocksDB.");
+
+ public static final ConfigOption<Boolean> MONITOR_STALL_MICROS =
+ ConfigOptions.key("state.backend.rocksdb.metrics.stall-micros")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Monitor the duration of writer requiring to wait for compaction or flush to finish in RocksDB.");
+
+ /** Creates a {@link ForStNativeMetricOptions} based on an external configuration. */
+ public static ForStNativeMetricOptions fromConfig(ReadableConfig config) {
+ ForStNativeMetricOptions options = new ForStNativeMetricOptions();
+ configurePropertyMetrics(options, config);
+ configureStatisticsMetrics(options, config);
+ return options;
+ }
+
+ private static void configurePropertyMetrics(
+ ForStNativeMetricOptions options, ReadableConfig config) {
+ if (config.get(MONITOR_NUM_IMMUTABLE_MEM_TABLES)) {
+ options.enableNumImmutableMemTable();
+ }
+
+ if (config.get(MONITOR_MEM_TABLE_FLUSH_PENDING)) {
+ options.enableMemTableFlushPending();
+ }
+
+ if (config.get(TRACK_COMPACTION_PENDING)) {
+ options.enableCompactionPending();
+ }
+
+ if (config.get(MONITOR_BACKGROUND_ERRORS)) {
+ options.enableBackgroundErrors();
+ }
+
+ if (config.get(MONITOR_CUR_SIZE_ACTIVE_MEM_TABLE)) {
+ options.enableCurSizeActiveMemTable();
+ }
+
+ if (config.get(MONITOR_CUR_SIZE_ALL_MEM_TABLE)) {
+ options.enableCurSizeAllMemTables();
+ }
+
+ if (config.get(MONITOR_SIZE_ALL_MEM_TABLES)) {
+ options.enableSizeAllMemTables();
+ }
+
+ if (config.get(MONITOR_NUM_ENTRIES_ACTIVE_MEM_TABLE)) {
+ options.enableNumEntriesActiveMemTable();
+ }
+
+ if (config.get(MONITOR_NUM_ENTRIES_IMM_MEM_TABLES)) {
+ options.enableNumEntriesImmMemTables();
+ }
+
+ if (config.get(MONITOR_NUM_DELETES_ACTIVE_MEM_TABLE)) {
+ options.enableNumDeletesActiveMemTable();
+ }
+
+ if (config.get(MONITOR_NUM_DELETES_IMM_MEM_TABLE)) {
+ options.enableNumDeletesImmMemTables();
+ }
+
+ if (config.get(ESTIMATE_NUM_KEYS)) {
+ options.enableEstimateNumKeys();
+ }
+
+ if (config.get(ESTIMATE_TABLE_READERS_MEM)) {
+ options.enableEstimateTableReadersMem();
+ }
+
+ if (config.get(MONITOR_NUM_SNAPSHOTS)) {
+ options.enableNumSnapshots();
+ }
+
+ if (config.get(MONITOR_NUM_LIVE_VERSIONS)) {
+ options.enableNumLiveVersions();
+ }
+
+ if (config.get(ESTIMATE_LIVE_DATA_SIZE)) {
+ options.enableEstimateLiveDataSize();
+ }
+
+ if (config.get(MONITOR_TOTAL_SST_FILES_SIZE)) {
+ options.enableTotalSstFilesSize();
+ }
+
+ if (config.get(MONITOR_LIVE_SST_FILES_SIZE)) {
+ options.enableLiveSstFilesSize();
+ }
+
+ if (config.get(ESTIMATE_PENDING_COMPACTION_BYTES)) {
+ options.enableEstimatePendingCompactionBytes();
+ }
+
+ if (config.get(MONITOR_NUM_RUNNING_COMPACTIONS)) {
+ options.enableNumRunningCompactions();
+ }
+
+ if (config.get(MONITOR_NUM_RUNNING_FLUSHES)) {
+ options.enableNumRunningFlushes();
+ }
+
+ if (config.get(MONITOR_ACTUAL_DELAYED_WRITE_RATE)) {
+ options.enableActualDelayedWriteRate();
+ }
+
+ if (config.get(IS_WRITE_STOPPED)) {
+ options.enableIsWriteStopped();
+ }
+
+ if (config.get(BLOCK_CACHE_CAPACITY)) {
+ options.enableBlockCacheCapacity();
+ }
+
+ if (config.get(BLOCK_CACHE_USAGE)) {
+ options.enableBlockCacheUsage();
+ }
+
+ if (config.get(BLOCK_CACHE_PINNED_USAGE)) {
+ options.enableBlockCachePinnedUsage();
+ }
+
+ if (config.get(MONITOR_NUM_FILES_AT_LEVEL)) {
+ options.enableNumFilesAtLevel();
+ }
+
+ options.setColumnFamilyAsVariable(config.get(COLUMN_FAMILY_AS_VARIABLE));
+ }
+
+ private static void configureStatisticsMetrics(
+ ForStNativeMetricOptions options, ReadableConfig config) {
+ for (Map.Entry<ConfigOption<Boolean>, TickerType> entry : tickerTypeMapping.entrySet()) {
+ if (config.get(entry.getKey())) {
+ options.monitorTickerTypes.add(entry.getValue());
+ }
+ }
+ }
+
+ private static final Map<ConfigOption<Boolean>, TickerType> tickerTypeMapping =
+ new HashMap<ConfigOption<Boolean>, TickerType>() {
+ private static final long serialVersionUID = 1L;
+
+ {
+ put(MONITOR_BLOCK_CACHE_HIT, TickerType.BLOCK_CACHE_HIT);
+ put(MONITOR_BLOCK_CACHE_MISS, TickerType.BLOCK_CACHE_MISS);
+ put(MONITOR_BLOOM_FILTER_USEFUL, TickerType.BLOOM_FILTER_USEFUL);
+ put(MONITOR_BLOOM_FILTER_FULL_POSITIVE, TickerType.BLOOM_FILTER_FULL_POSITIVE);
+ put(
+ MONITOR_BLOOM_FILTER_FULL_TRUE_POSITIVE,
+ TickerType.BLOOM_FILTER_FULL_TRUE_POSITIVE);
+ put(MONITOR_BYTES_READ, TickerType.BYTES_READ);
+ put(MONITOR_ITER_BYTES_READ, TickerType.ITER_BYTES_READ);
+ put(MONITOR_BYTES_WRITTEN, TickerType.BYTES_WRITTEN);
+ put(MONITOR_COMPACTION_READ_BYTES, TickerType.COMPACT_READ_BYTES);
+ put(MONITOR_COMPACTION_WRITE_BYTES, TickerType.COMPACT_WRITE_BYTES);
+ put(MONITOR_STALL_MICROS, TickerType.STALL_MICROS);
+ }
+ };
+
+ private final Set<ForStProperty> properties;
+ private final Set<TickerType> monitorTickerTypes;
+ private boolean columnFamilyAsVariable = COLUMN_FAMILY_AS_VARIABLE.defaultValue();
+
+ public ForStNativeMetricOptions() {
+ this.properties = new HashSet<>();
+ this.monitorTickerTypes = new HashSet<>();
+ }
+
+ @VisibleForTesting
+ public void enableNativeStatistics(ConfigOption<Boolean> nativeStatisticsOption) {
+ TickerType tickerType = tickerTypeMapping.get(nativeStatisticsOption);
+ if (tickerType != null) {
+ monitorTickerTypes.add(tickerType);
+ } else {
+ throw new IllegalArgumentException(
+ "Unknown configurable native statistics option " + nativeStatisticsOption);
+ }
+ }
+
+ /** Returns number of immutable memtables that have not yet been flushed. */
+ public void enableNumImmutableMemTable() {
+ this.properties.add(ForStProperty.NumImmutableMemTable);
+ }
+
+ /** Returns 1 if a memtable flush is pending; otherwise, returns 0. */
+ public void enableMemTableFlushPending() {
+ this.properties.add(ForStProperty.MemTableFlushPending);
+ }
+
+ /** Returns 1 if at least one compaction is pending; otherwise, returns 0. */
+ public void enableCompactionPending() {
+ this.properties.add(ForStProperty.CompactionPending);
+ }
+
+ /** Returns accumulated number of background errors. */
+ public void enableBackgroundErrors() {
+ this.properties.add(ForStProperty.BackgroundErrors);
+ }
+
+ /** Returns approximate size of active memtable (bytes). */
+ public void enableCurSizeActiveMemTable() {
+ this.properties.add(ForStProperty.CurSizeActiveMemTable);
+ }
+
+ /** Returns approximate size of active and unflushed immutable memtables (bytes). */
+ public void enableCurSizeAllMemTables() {
+ this.properties.add(ForStProperty.CurSizeAllMemTables);
+ }
+
+ /**
+ * Returns approximate size of active, unflushed immutable, and pinned immutable memtables
+ * (bytes).
+ */
+ public void enableSizeAllMemTables() {
+ this.properties.add(ForStProperty.SizeAllMemTables);
+ }
+
+ /** Returns total number of entries in the active memtable. */
+ public void enableNumEntriesActiveMemTable() {
+ this.properties.add(ForStProperty.NumEntriesActiveMemTable);
+ }
+
+ /** Returns total number of entries in the unflushed immutable memtables. */
+ public void enableNumEntriesImmMemTables() {
+ this.properties.add(ForStProperty.NumEntriesImmMemTables);
+ }
+
+ /** Returns total number of delete entries in the active memtable. */
+ public void enableNumDeletesActiveMemTable() {
+ this.properties.add(ForStProperty.NumDeletesActiveMemTable);
+ }
+
+ /** Returns total number of delete entries in the unflushed immutable memtables. */
+ public void enableNumDeletesImmMemTables() {
+ this.properties.add(ForStProperty.NumDeletesImmMemTables);
+ }
+
+ /**
+ * Returns estimated number of total keys in the active and unflushed immutable memtables and
+ * storage.
+ */
+ public void enableEstimateNumKeys() {
+ this.properties.add(ForStProperty.EstimateNumKeys);
+ }
+
+ /**
+ * Returns estimated memory used for reading SST tables, excluding memory used in block cache
+ * (e.g.,filter and index blocks).
+ */
+ public void enableEstimateTableReadersMem() {
+ this.properties.add(ForStProperty.EstimateTableReadersMem);
+ }
+
+ /** Returns number of unreleased snapshots of the database. */
+ public void enableNumSnapshots() {
+ this.properties.add(ForStProperty.NumSnapshots);
+ }
+
+ /**
+ * Returns number of live versions. `Version` is an internal data structure. See version_set.h
+ * for details. More live versions often mean more SST files are held from being deleted, by
+ * iterators or unfinished compactions.
+ */
+ public void enableNumLiveVersions() {
+ this.properties.add(ForStProperty.NumLiveVersions);
+ }
+
+ /** Returns an estimate of the amount of live data in bytes. */
+ public void enableEstimateLiveDataSize() {
+ this.properties.add(ForStProperty.EstimateLiveDataSize);
+ }
+
+ /**
+ * Returns total size (bytes) of all SST files. <strong>WARNING</strong>: may slow down online
+ * queries if there are too many files.
+ */
+ public void enableTotalSstFilesSize() {
+ this.properties.add(ForStProperty.TotalSstFilesSize);
+ }
+
+ public void enableLiveSstFilesSize() {
+ this.properties.add(ForStProperty.LiveSstFilesSize);
+ }
+
+ /**
+ * Returns estimated total number of bytes compaction needs to rewrite to get all levels down to
+ * under target size. Not valid for other compactions than level-based.
+ */
+ public void enableEstimatePendingCompactionBytes() {
+ this.properties.add(ForStProperty.EstimatePendingCompactionBytes);
+ }
+
+ /** Returns the number of currently running compactions. */
+ public void enableNumRunningCompactions() {
+ this.properties.add(ForStProperty.NumRunningCompactions);
+ }
+
+ /** Returns the number of currently running flushes. */
+ public void enableNumRunningFlushes() {
+ this.properties.add(ForStProperty.NumRunningFlushes);
+ }
+
+ /** Returns the current actual delayed write rate. 0 means no delay. */
+ public void enableActualDelayedWriteRate() {
+ this.properties.add(ForStProperty.ActualDelayedWriteRate);
+ }
+
+ /** Returns 1 if write has been stopped. */
+ public void enableIsWriteStopped() {
+ this.properties.add(ForStProperty.IsWriteStopped);
+ }
+
+ /** Returns block cache capacity. */
+ public void enableBlockCacheCapacity() {
+ this.properties.add(ForStProperty.BlockCacheCapacity);
+ }
+
+ /** Returns the memory size for the entries residing in block cache. */
+ public void enableBlockCacheUsage() {
+ this.properties.add(ForStProperty.BlockCacheUsage);
+ }
+
+ /** Returns the memory size for the entries being pinned in block cache. */
+ public void enableBlockCachePinnedUsage() {
+ this.properties.add(ForStProperty.BlockCachePinnedUsage);
+ }
+
+ /** Returns the number of files per level. */
+ public void enableNumFilesAtLevel() {
+ this.properties.add(ForStProperty.NumFilesAtLevel0);
+ this.properties.add(ForStProperty.NumFilesAtLevel1);
+ this.properties.add(ForStProperty.NumFilesAtLevel2);
+ this.properties.add(ForStProperty.NumFilesAtLevel3);
+ this.properties.add(ForStProperty.NumFilesAtLevel4);
+ this.properties.add(ForStProperty.NumFilesAtLevel5);
+ this.properties.add(ForStProperty.NumFilesAtLevel6);
+ }
+
+ /** Returns the column family as variable. */
+ public void setColumnFamilyAsVariable(boolean columnFamilyAsVariable) {
+ this.columnFamilyAsVariable = columnFamilyAsVariable;
+ }
+
+ /** @return the enabled RocksDB property-based metrics */
+ public Collection<ForStProperty> getProperties() {
+ return Collections.unmodifiableCollection(properties);
+ }
+
+ /** @return the enabled RocksDB statistics metrics. */
+ public Collection<TickerType> getMonitorTickerTypes() {
+ return Collections.unmodifiableCollection(monitorTickerTypes);
+ }
+
+ /**
+ * {{@link ForStNativeMetricMonitor}} is enabled if any property or ticker type is set.
+ *
+ * @return true if {{RocksDBNativeMetricMonitor}} should be enabled, false otherwise.
+ */
+ public boolean isEnabled() {
+ return !properties.isEmpty() || isStatisticsEnabled();
+ }
+
+ /** @return true if RocksDB statistics metrics are enabled, false otherwise. */
+ public boolean isStatisticsEnabled() {
+ return !monitorTickerTypes.isEmpty();
+ }
+
+ /**
+ * {{@link ForStNativeMetricMonitor}} Whether to expose the column family as a variable..
+ *
+ * @return true is column family to expose variable, false otherwise.
+ */
+ public boolean isColumnFamilyAsVariable() {
+ return this.columnFamilyAsVariable;
+ }
+}
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStOptionsFactory.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStOptionsFactory.java
index 7fbc432e659..80ac8511fa6 100644
--- a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStOptionsFactory.java
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStOptionsFactory.java
@@ -66,7 +66,19 @@ public interface ForStOptionsFactory extends java.io.Serializable {
ColumnFamilyOptions createColumnOptions(
ColumnFamilyOptions currentOptions, Collection<AutoCloseable> handlesToClose);
- // TODO: Support Metrics Options
+ /**
+ * This method should enable certain ForSt metrics to be forwarded to Flink's metrics reporter.
+ *
+ * <p>Enabling these monitoring options may degrade ForSt performance and should be set with
+ * care.
+ *
+ * @param nativeMetricOptions The options object with the pre-defined options.
+ * @return The options object on which the additional options are set.
+ */
+ default ForStNativeMetricOptions createNativeMetricsOptions(
+ ForStNativeMetricOptions nativeMetricOptions) {
+ return nativeMetricOptions;
+ }
/**
* This method should set the additional options on top of the current options object. The
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStProperty.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStProperty.java
new file mode 100644
index 00000000000..042fb5916b8
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStProperty.java
@@ -0,0 +1,115 @@
+/*
+ * 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.flink.state.forst;
+
+import org.apache.flink.annotation.Internal;
+
+import org.rocksdb.ColumnFamilyHandle;
+import org.rocksdb.RocksDB;
+
+/**
+ * {@link RocksDB} properties that can be queried by Flink's metrics reporter.
+ *
+ * <p>Note: Metrics properties are added in each new version of {@link RocksDB}, when upgrading to a
+ * latter version consider updating this class with newly added properties.
+ */
+@Internal
+public enum ForStProperty {
+ NumImmutableMemTable("num-immutable-mem-table", PropertyType.NUMBER),
+ MemTableFlushPending("mem-table-flush-pending", PropertyType.NUMBER),
+ CompactionPending("compaction-pending", PropertyType.NUMBER),
+ BackgroundErrors("background-errors", PropertyType.NUMBER),
+ CurSizeActiveMemTable("cur-size-active-mem-table", PropertyType.NUMBER),
+ CurSizeAllMemTables("cur-size-all-mem-tables", PropertyType.NUMBER),
+ SizeAllMemTables("size-all-mem-tables", PropertyType.NUMBER),
+ NumEntriesActiveMemTable("num-entries-active-mem-table", PropertyType.NUMBER),
+ NumEntriesImmMemTables("num-entries-imm-mem-tables", PropertyType.NUMBER),
+ NumDeletesActiveMemTable("num-deletes-active-mem-table", PropertyType.NUMBER),
+ NumDeletesImmMemTables("num-deletes-imm-mem-tables", PropertyType.NUMBER),
+ EstimateNumKeys("estimate-num-keys", PropertyType.NUMBER),
+ EstimateTableReadersMem("estimate-table-readers-mem", PropertyType.NUMBER),
+ NumSnapshots("num-snapshots", PropertyType.NUMBER),
+ NumLiveVersions("num-live-versions", PropertyType.NUMBER),
+ EstimateLiveDataSize("estimate-live-data-size", PropertyType.NUMBER),
+ TotalSstFilesSize("total-sst-files-size", PropertyType.NUMBER),
+ LiveSstFilesSize("live-sst-files-size", PropertyType.NUMBER),
+ EstimatePendingCompactionBytes("estimate-pending-compaction-bytes", PropertyType.NUMBER),
+ NumRunningCompactions("num-running-compactions", PropertyType.NUMBER),
+ NumRunningFlushes("num-running-flushes", PropertyType.NUMBER),
+ ActualDelayedWriteRate("actual-delayed-write-rate", PropertyType.NUMBER),
+ IsWriteStopped("is-write-stopped", PropertyType.NUMBER),
+ BlockCacheCapacity("block-cache-capacity", PropertyType.NUMBER),
+ BlockCacheUsage("block-cache-usage", PropertyType.NUMBER),
+ BlockCachePinnedUsage("block-cache-pinned-usage", PropertyType.NUMBER),
+ NumFilesAtLevel0("num-files-at-level0", PropertyType.STRING),
+ NumFilesAtLevel1("num-files-at-level1", PropertyType.STRING),
+ NumFilesAtLevel2("num-files-at-level2", PropertyType.STRING),
+ NumFilesAtLevel3("num-files-at-level3", PropertyType.STRING),
+ NumFilesAtLevel4("num-files-at-level4", PropertyType.STRING),
+ NumFilesAtLevel5("num-files-at-level5", PropertyType.STRING),
+ NumFilesAtLevel6("num-files-at-level6", PropertyType.STRING);
+
+ private static final String ROCKS_DB_PROPERTY_FORMAT = "rocksdb.%s";
+
+ private static final String CONFIG_KEY_FORMAT = "state.backend.forst.metrics.%s";
+
+ private final String property;
+
+ private final PropertyType type;
+
+ /** Property type. */
+ private enum PropertyType {
+ NUMBER,
+ STRING
+ }
+
+ ForStProperty(String property, PropertyType type) {
+ this.property = property;
+ this.type = type;
+ }
+
+ /**
+ * @return property string that can be used to query {@link
+ * RocksDB#getLongProperty(ColumnFamilyHandle, String)}.
+ */
+ public String getForStProperty() {
+ return String.format(ROCKS_DB_PROPERTY_FORMAT, property);
+ }
+
+ public long getNumericalPropertyValue(RocksDB rocksDB, ColumnFamilyHandle handle)
+ throws Exception {
+ String forStProperty = getForStProperty();
+ switch (type) {
+ case NUMBER:
+ return rocksDB.getLongProperty(handle, forStProperty);
+ case STRING:
+ return Long.parseLong(rocksDB.getProperty(handle, forStProperty));
+ default:
+ throw new RuntimeException(
+ String.format("ForSt property type: %s not supported", type));
+ }
+ }
+
+ /**
+ * @return key for enabling metric using {@link org.apache.flink.configuration.Configuration}.
+ */
+ public String getConfigKey() {
+ return String.format(CONFIG_KEY_FORMAT, property);
+ }
+}
diff --git a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStExtension.java b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStExtension.java
new file mode 100644
index 00000000000..a6a2bfa9a22
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStExtension.java
@@ -0,0 +1,212 @@
+/*
+ * 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.flink.state.forst;
+
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.IOUtils;
+
+import org.junit.jupiter.api.extension.AfterEachCallback;
+import org.junit.jupiter.api.extension.BeforeEachCallback;
+import org.junit.jupiter.api.extension.ExtensionContext;
+import org.junit.rules.TemporaryFolder;
+import org.rocksdb.ColumnFamilyDescriptor;
+import org.rocksdb.ColumnFamilyHandle;
+import org.rocksdb.ColumnFamilyOptions;
+import org.rocksdb.DBOptions;
+import org.rocksdb.InfoLogLevel;
+import org.rocksdb.ReadOptions;
+import org.rocksdb.RocksDB;
+import org.rocksdb.Statistics;
+import org.rocksdb.WriteOptions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nonnull;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+
+/** External extension for tests that require an instance of ForSt. */
+public class ForStExtension implements BeforeEachCallback, AfterEachCallback {
+ private static final Logger LOG = LoggerFactory.getLogger(ForStExtension.class);
+
+ /** Factory for {@link DBOptions} and {@link ColumnFamilyOptions}. */
+ private final ForStOptionsFactory optionsFactory;
+
+ private final boolean enableStatistics;
+
+ /** Temporary folder that provides the working directory for the ForSt instance. */
+ private TemporaryFolder temporaryFolder;
+
+ /** The options for the ForSt instance. */
+ private DBOptions dbOptions;
+
+ /** The options for column families created with the ForSt instance. */
+ private ColumnFamilyOptions columnFamilyOptions;
+
+ /** The options for writes. */
+ private WriteOptions writeOptions;
+
+ /** The options for reads. */
+ private ReadOptions readOptions;
+
+ /** The ForSt instance object. */
+ private RocksDB db;
+
+ /** List of all column families that have been created with the ForSt instance. */
+ private List<ColumnFamilyHandle> columnFamilyHandles;
+
+ /** Resources to close. */
+ private final ArrayList<AutoCloseable> handlesToClose = new ArrayList<>();
+
+ public ForStExtension() {
+ this(false);
+ }
+
+ public ForStExtension(boolean enableStatistics) {
+ this(
+ new ForStOptionsFactory() {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public DBOptions createDBOptions(
+ DBOptions currentOptions, Collection<AutoCloseable> handlesToClose) {
+ // close it before reuse the reference.
+ try {
+ currentOptions.close();
+ } catch (Exception e) {
+ LOG.error("Close previous DBOptions's instance failed.", e);
+ }
+
+ return new DBOptions()
+ .setMaxBackgroundJobs(4)
+ .setUseFsync(false)
+ .setMaxOpenFiles(-1)
+ .setInfoLogLevel(InfoLogLevel.HEADER_LEVEL)
+ .setStatsDumpPeriodSec(0);
+ }
+
+ @Override
+ public ColumnFamilyOptions createColumnOptions(
+ ColumnFamilyOptions currentOptions,
+ Collection<AutoCloseable> handlesToClose) {
+ // close it before reuse the reference.
+ try {
+ currentOptions.close();
+ } catch (Exception e) {
+ LOG.error("Close previous ColumnOptions's instance failed.", e);
+ }
+
+ return new ColumnFamilyOptions().optimizeForPointLookup(40960);
+ }
+ },
+ enableStatistics);
+ }
+
+ public ForStExtension(@Nonnull ForStOptionsFactory optionsFactory, boolean enableStatistics) {
+ this.optionsFactory = optionsFactory;
+ this.enableStatistics = enableStatistics;
+ }
+
+ public ColumnFamilyHandle getDefaultColumnFamily() {
+ return columnFamilyHandles.get(0);
+ }
+
+ public RocksDB getDB() {
+ return db;
+ }
+
+ public DBOptions getDbOptions() {
+ return dbOptions;
+ }
+
+ /** Creates and returns a new column family with the given name. */
+ public ColumnFamilyHandle createNewColumnFamily(String name) {
+ try {
+ final ColumnFamilyHandle columnFamily =
+ db.createColumnFamily(
+ new ColumnFamilyDescriptor(name.getBytes(), columnFamilyOptions));
+ columnFamilyHandles.add(columnFamily);
+ return columnFamily;
+ } catch (Exception ex) {
+ throw new FlinkRuntimeException("Could not create column family.", ex);
+ }
+ }
+
+ public void before() throws Exception {
+ this.temporaryFolder = new TemporaryFolder();
+ this.temporaryFolder.create();
+ final File rocksFolder = temporaryFolder.newFolder();
+ this.dbOptions =
+ optionsFactory
+ .createDBOptions(
+ new DBOptions()
+ .setUseFsync(false)
+ .setInfoLogLevel(InfoLogLevel.HEADER_LEVEL)
+ .setStatsDumpPeriodSec(0),
+ handlesToClose)
+ .setCreateIfMissing(true);
+ if (enableStatistics) {
+ Statistics statistics = new Statistics();
+ dbOptions.setStatistics(statistics);
+ handlesToClose.add(statistics);
+ }
+ this.columnFamilyOptions =
+ optionsFactory.createColumnOptions(new ColumnFamilyOptions(), handlesToClose);
+ this.writeOptions = new WriteOptions();
+ this.writeOptions.disableWAL();
+ this.readOptions = new ReadOptions();
+ this.columnFamilyHandles = new ArrayList<>(1);
+ this.db =
+ RocksDB.open(
+ dbOptions,
+ rocksFolder.getAbsolutePath(),
+ Collections.singletonList(
+ new ColumnFamilyDescriptor(
+ "default".getBytes(), columnFamilyOptions)),
+ columnFamilyHandles);
+ }
+
+ public void after() {
+ // destruct in reversed order of creation.
+ for (ColumnFamilyHandle columnFamilyHandle : columnFamilyHandles) {
+ IOUtils.closeQuietly(columnFamilyHandle);
+ }
+ IOUtils.closeQuietly(this.db);
+ IOUtils.closeQuietly(this.readOptions);
+ IOUtils.closeQuietly(this.writeOptions);
+ IOUtils.closeQuietly(this.columnFamilyOptions);
+ IOUtils.closeQuietly(this.dbOptions);
+ handlesToClose.forEach(IOUtils::closeQuietly);
+ temporaryFolder.delete();
+ }
+
+ @Override
+ public void beforeEach(ExtensionContext context) throws Exception {
+ before();
+ }
+
+ @Override
+ public void afterEach(ExtensionContext context) {
+ after();
+ }
+}
diff --git a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStNativeMetricMonitorTest.java b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStNativeMetricMonitorTest.java
new file mode 100644
index 00000000000..b1a2716e8a3
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStNativeMetricMonitorTest.java
@@ -0,0 +1,254 @@
+/*
+ * 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.flink.state.forst;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.MetricOptions;
+import org.apache.flink.metrics.Metric;
+import org.apache.flink.runtime.metrics.MetricRegistry;
+import org.apache.flink.runtime.metrics.groups.AbstractMetricGroup;
+import org.apache.flink.runtime.metrics.groups.GenericMetricGroup;
+import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
+import org.apache.flink.runtime.metrics.scope.ScopeFormats;
+import org.apache.flink.traces.SpanBuilder;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.rocksdb.ColumnFamilyHandle;
+import org.rocksdb.RocksDBException;
+
+import java.math.BigInteger;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** validate native metric monitor. */
+class ForStNativeMetricMonitorTest {
+
+ private static final String OPERATOR_NAME = "dummy";
+
+ private static final String COLUMN_FAMILY_NAME = "column-family";
+
+ @RegisterExtension public ForStExtension forStExtension = new ForStExtension(true);
+
+ @Test
+ void testMetricMonitorLifecycle() throws Throwable {
+ // We use a local variable here to manually control the life-cycle.
+ // This allows us to verify that metrics do not try to access
+ // RocksDB after the monitor was closed.
+ ForStExtension localForStExtension = new ForStExtension(true);
+ localForStExtension.before();
+
+ SimpleMetricRegistry registry = new SimpleMetricRegistry();
+ GenericMetricGroup group =
+ new GenericMetricGroup(
+ registry,
+ UnregisteredMetricGroups.createUnregisteredTaskMetricGroup(),
+ OPERATOR_NAME);
+
+ ForStNativeMetricOptions options = new ForStNativeMetricOptions();
+ // always returns a non-zero
+ // value since empty memtables
+ // have overhead.
+ options.enableSizeAllMemTables();
+ options.enableNativeStatistics(ForStNativeMetricOptions.MONITOR_BYTES_WRITTEN);
+
+ ForStNativeMetricMonitor monitor =
+ new ForStNativeMetricMonitor(
+ options,
+ group,
+ localForStExtension.getDB(),
+ localForStExtension.getDbOptions().statistics());
+
+ ColumnFamilyHandle handle = localForStExtension.createNewColumnFamily(COLUMN_FAMILY_NAME);
+ monitor.registerColumnFamily(COLUMN_FAMILY_NAME, handle);
+
+ assertThat(registry.propertyMetrics)
+ .withFailMessage("Failed to register metrics for column family")
+ .hasSize(1);
+
+ // write something to ensure the bytes-written is not zero.
+ localForStExtension.getDB().put(new byte[4], new byte[10]);
+
+ for (ForStNativeMetricMonitor.ForStNativePropertyMetricView view :
+ registry.propertyMetrics) {
+ view.update();
+ assertThat(view.getValue())
+ .withFailMessage("Failed to pull metric from ForSt")
+ .isNotEqualTo(BigInteger.ZERO);
+ view.setValue(0L);
+ }
+
+ for (ForStNativeMetricMonitor.ForStNativeStatisticsMetricView view :
+ registry.statisticsMetrics) {
+ view.update();
+ assertThat(view.getValue()).isNotZero();
+ view.setValue(0L);
+ }
+
+ // After the monitor is closed no metric should be accessing RocksDB anymore.
+ // If they do, then this test will likely fail with a segmentation fault.
+ monitor.close();
+
+ localForStExtension.after();
+
+ for (ForStNativeMetricMonitor.ForStNativePropertyMetricView view :
+ registry.propertyMetrics) {
+ view.update();
+ assertThat(view.getValue())
+ .withFailMessage("Failed to release ForSt reference")
+ .isEqualTo(BigInteger.ZERO);
+ }
+
+ for (ForStNativeMetricMonitor.ForStNativeStatisticsMetricView view :
+ registry.statisticsMetrics) {
+ view.update();
+ assertThat(view.getValue()).isZero();
+ }
+ }
+
+ @Test
+ void testReturnsUnsigned() throws Throwable {
+ ForStExtension localForStExtension = new ForStExtension();
+ localForStExtension.before();
+
+ SimpleMetricRegistry registry = new SimpleMetricRegistry();
+ GenericMetricGroup group =
+ new GenericMetricGroup(
+ registry,
+ UnregisteredMetricGroups.createUnregisteredTaskMetricGroup(),
+ OPERATOR_NAME);
+
+ ForStNativeMetricOptions options = new ForStNativeMetricOptions();
+ options.enableSizeAllMemTables();
+
+ ForStNativeMetricMonitor monitor =
+ new ForStNativeMetricMonitor(
+ options,
+ group,
+ localForStExtension.getDB(),
+ localForStExtension.getDbOptions().statistics());
+
+ ColumnFamilyHandle handle = forStExtension.createNewColumnFamily(COLUMN_FAMILY_NAME);
+ monitor.registerColumnFamily(COLUMN_FAMILY_NAME, handle);
+ ForStNativeMetricMonitor.ForStNativePropertyMetricView view =
+ registry.propertyMetrics.get(0);
+
+ view.setValue(-1);
+ BigInteger result = view.getValue();
+
+ localForStExtension.after();
+
+ assertThat(result.signum())
+ .withFailMessage("Failed to interpret ForSt result as an unsigned long")
+ .isOne();
+ }
+
+ @Test
+ void testClosedGaugesDontRead() throws RocksDBException {
+ SimpleMetricRegistry registry = new SimpleMetricRegistry();
+ GenericMetricGroup group =
+ new GenericMetricGroup(
+ registry,
+ UnregisteredMetricGroups.createUnregisteredTaskMetricGroup(),
+ OPERATOR_NAME);
+
+ ForStNativeMetricOptions options = new ForStNativeMetricOptions();
+ options.enableSizeAllMemTables();
+ options.enableNativeStatistics(ForStNativeMetricOptions.MONITOR_BLOCK_CACHE_HIT);
+
+ ForStNativeMetricMonitor monitor =
+ new ForStNativeMetricMonitor(
+ options,
+ group,
+ forStExtension.getDB(),
+ forStExtension.getDbOptions().statistics());
+
+ ColumnFamilyHandle handle = forStExtension.createNewColumnFamily(COLUMN_FAMILY_NAME);
+ monitor.registerColumnFamily(COLUMN_FAMILY_NAME, handle);
+
+ forStExtension.getDB().put(new byte[4], new byte[10]);
+
+ for (ForStNativeMetricMonitor.ForStNativePropertyMetricView view :
+ registry.propertyMetrics) {
+ view.close();
+ view.update();
+ assertThat(view.getValue())
+ .withFailMessage("Closed gauge still queried ForSt")
+ .isEqualTo(BigInteger.ZERO);
+ }
+
+ for (ForStNativeMetricMonitor.ForStNativeStatisticsMetricView view :
+ registry.statisticsMetrics) {
+ view.close();
+ view.update();
+ assertThat(view.getValue())
+ .withFailMessage("Closed gauge still queried ForSt")
+ .isZero();
+ }
+ }
+
+ static class SimpleMetricRegistry implements MetricRegistry {
+ List<ForStNativeMetricMonitor.ForStNativePropertyMetricView> propertyMetrics =
+ new ArrayList<>();
+
+ List<ForStNativeMetricMonitor.ForStNativeStatisticsMetricView> statisticsMetrics =
+ new ArrayList<>();
+
+ @Override
+ public char getDelimiter() {
+ return 0;
+ }
+
+ @Override
+ public int getNumberReporters() {
+ return 0;
+ }
+
+ @Override
+ public void addSpan(SpanBuilder spanBuilder) {}
+
+ @Override
+ public void register(Metric metric, String metricName, AbstractMetricGroup group) {
+ if (metric instanceof ForStNativeMetricMonitor.ForStNativePropertyMetricView) {
+ propertyMetrics.add(
+ (ForStNativeMetricMonitor.ForStNativePropertyMetricView) metric);
+ } else if (metric instanceof ForStNativeMetricMonitor.ForStNativeStatisticsMetricView) {
+ statisticsMetrics.add(
+ (ForStNativeMetricMonitor.ForStNativeStatisticsMetricView) metric);
+ }
+ }
+
+ @Override
+ public void unregister(Metric metric, String metricName, AbstractMetricGroup group) {}
+
+ @Override
+ public ScopeFormats getScopeFormats() {
+ Configuration config = new Configuration();
+
+ config.set(MetricOptions.SCOPE_NAMING_TM, "A");
+ config.set(MetricOptions.SCOPE_NAMING_TM_JOB, "B");
+ config.set(MetricOptions.SCOPE_NAMING_TASK, "C");
+ config.set(MetricOptions.SCOPE_NAMING_OPERATOR, "D");
+
+ return ScopeFormats.fromConfig(config);
+ }
+ }
+}
diff --git a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStNativeMetricOptionsTest.java b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStNativeMetricOptionsTest.java
new file mode 100644
index 00000000000..1c8a3bddb61
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStNativeMetricOptionsTest.java
@@ -0,0 +1,53 @@
+/*
+ * 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.flink.state.forst;
+
+import org.apache.flink.configuration.Configuration;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+/** Test all native metrics can be set using configuration. */
+public class ForStNativeMetricOptionsTest {
+ @Test
+ public void testNativeMetricsConfigurable() {
+ for (ForStProperty property : ForStProperty.values()) {
+ Configuration config = new Configuration();
+ if (property.getConfigKey().contains("num-files-at-level")) {
+ config.set(ForStNativeMetricOptions.MONITOR_NUM_FILES_AT_LEVEL, true);
+ } else {
+ config.setBoolean(property.getConfigKey(), true);
+ }
+
+ ForStNativeMetricOptions options = ForStNativeMetricOptions.fromConfig(config);
+
+ Assert.assertTrue(
+ String.format(
+ "Failed to enable native metrics with property %s",
+ property.getConfigKey()),
+ options.isEnabled());
+
+ Assert.assertTrue(
+ String.format(
+ "Failed to enable native metric %s using config",
+ property.getConfigKey()),
+ options.getProperties().contains(property));
+ }
+ }
+}
diff --git a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStPropertyTest.java b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStPropertyTest.java
new file mode 100644
index 00000000000..56e577b803f
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStPropertyTest.java
@@ -0,0 +1,47 @@
+/*
+ * 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.flink.state.forst;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.rocksdb.ColumnFamilyHandle;
+import org.rocksdb.RocksDB;
+
+/** Validate ForSt properties. */
+class ForStPropertyTest {
+
+ @RegisterExtension public ForStExtension forStExtension = new ForStExtension();
+
+ @Test
+ void testForStPropertiesValid() {
+ RocksDB db = forStExtension.getDB();
+ ColumnFamilyHandle handle = forStExtension.getDefaultColumnFamily();
+
+ for (ForStProperty forStProperty : ForStProperty.values()) {
+ try {
+ forStProperty.getNumericalPropertyValue(db, handle);
+ } catch (Exception e) {
+ throw new AssertionError(
+ String.format(
+ "Invalid ForSt property %s", forStProperty.getForStProperty()),
+ e);
+ }
+ }
+ }
+}
(flink) 01/07: [FLINK-35047][state] Introduce forstjni dependency
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hangxiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 629c35712dc84eb833e9f960d16640b1e2f16feb
Author: Hangxiang Yu <ma...@gmail.com>
AuthorDate: Mon Apr 15 17:24:25 2024 +0800
[FLINK-35047][state] Introduce forstjni dependency
---
flink-state-backends/flink-statebackend-forst/pom.xml | 11 +++++------
1 file changed, 5 insertions(+), 6 deletions(-)
diff --git a/flink-state-backends/flink-statebackend-forst/pom.xml b/flink-state-backends/flink-statebackend-forst/pom.xml
index c4fb6252c8b..abf750379a7 100644
--- a/flink-state-backends/flink-statebackend-forst/pom.xml
+++ b/flink-state-backends/flink-statebackend-forst/pom.xml
@@ -60,12 +60,11 @@ under the License.
<scope>provided</scope>
</dependency>
-<!-- TODO: Enable forst dependency when it's available in the maven central repository -->
-<!-- <dependency>-->
-<!-- <groupId>com.ververica</groupId>-->
-<!-- <artifactId>forstjni</artifactId>-->
-<!-- <version>0.1.0-beta</version>-->
-<!-- </dependency>-->
+ <dependency>
+ <groupId>com.ververica</groupId>
+ <artifactId>forstjni</artifactId>
+ <version>0.1.0-beta</version>
+ </dependency>
<!-- test dependencies -->