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/05/29 12:52:13 UTC
(flink) branch master updated (2c35e48addf -> 30d75f1511b)
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 2c35e48addf [FLINK-35348][table] Introduce refresh materialized table rest api
new 9a7b976364d [hotfix][statebackend] Normalize ForSt option names
new 30d75f1511b [hotfix][statebackend] Normalize ForSt working dir
The 2 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/state/forst/ForStKeyedStateBackend.java | 4 +-
.../state/forst/ForStKeyedStateBackendBuilder.java | 18 +++--
.../org/apache/flink/state/forst/ForStOptions.java | 4 +-
.../flink/state/forst/ForStResourceContainer.java | 78 ++++++++++++++--------
.../flink/state/forst/ForStStateBackend.java | 30 ++++-----
.../state/forst/ForStResourceContainerTest.java | 8 +--
6 files changed, 85 insertions(+), 57 deletions(-)
(flink) 01/02: [hotfix][statebackend] Normalize ForSt option names
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 9a7b976364d2c93e064fb985d51e9bdee672cac6
Author: wangfeifan <zo...@163.com>
AuthorDate: Mon May 27 19:00:40 2024 +0800
[hotfix][statebackend] Normalize ForSt option names
---
.../src/main/java/org/apache/flink/state/forst/ForStOptions.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
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 c0bd1e123a6..2ff6d2e1534 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
@@ -32,7 +32,7 @@ public class ForStOptions {
/** 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")
+ ConfigOptions.key("state.backend.forst.local-dir")
.stringType()
.noDefaultValue()
.withDescription(
@@ -47,7 +47,7 @@ public class ForStOptions {
/** The remote directory where ForSt puts its SST files. */
public static final ConfigOption<String> REMOTE_DIRECTORY =
- ConfigOptions.key("state.backend.forst.remotedir")
+ ConfigOptions.key("state.backend.forst.remote-dir")
.stringType()
.noDefaultValue()
.withDescription(
(flink) 02/02: [hotfix][statebackend] Normalize ForSt working dir
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 30d75f1511b2b2beac4d7f18ee20941b3bd4cbe8
Author: wangfeifan <zo...@163.com>
AuthorDate: Mon May 27 20:46:03 2024 +0800
[hotfix][statebackend] Normalize ForSt working dir
Co-authored-by: yhx <ma...@gmail.com>
---
.../flink/state/forst/ForStKeyedStateBackend.java | 4 +-
.../state/forst/ForStKeyedStateBackendBuilder.java | 18 +++--
.../flink/state/forst/ForStResourceContainer.java | 78 ++++++++++++++--------
.../flink/state/forst/ForStStateBackend.java | 30 ++++-----
.../state/forst/ForStResourceContainerTest.java | 8 +--
5 files changed, 83 insertions(+), 55 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 5192536a8dc..c5ec58bccc0 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
@@ -20,6 +20,7 @@ package org.apache.flink.state.forst;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.state.v2.State;
import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.fs.Path;
import org.apache.flink.core.memory.DataInputDeserializer;
import org.apache.flink.core.memory.DataOutputSerializer;
import org.apache.flink.runtime.asyncprocessing.StateExecutor;
@@ -43,7 +44,6 @@ import javax.annotation.concurrent.GuardedBy;
import java.io.File;
import java.io.IOException;
-import java.net.URI;
import java.util.HashSet;
import java.util.Set;
import java.util.function.Function;
@@ -231,7 +231,7 @@ public class ForStKeyedStateBackend<K> implements AsyncKeyedStateBackend {
}
@VisibleForTesting
- URI getRemoteBasePath() {
+ Path 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 d3fad83c13f..5a52d9dad6a 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
@@ -37,13 +37,13 @@ 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.util.Collection;
import java.util.function.Function;
import java.util.function.Supplier;
@@ -180,11 +180,21 @@ public class ForStKeyedStateBackendBuilder<K>
}
private ForStRestoreOperation getForStRestoreOperation() {
- DBOptions dbOptions = optionsContainer.getDbOptions();
+ // Currently, ForStDB does not support mixing local-dir and remote-dir, and ForStDB will
+ // concatenates the dfs directory with the local directory as working dir when using flink
+ // env. We expect to directly use the dfs directory in flink env or local directory as
+ // working dir. We will implement this in ForStDB later, but before that, we achieved this
+ // by setting the dbPath to "/" when the dfs directory existed.
+ // TODO: use localForStPath as dbPath after ForSt Support mixing local-dir and remote-dir
+ File instanceForStPath =
+ optionsContainer.getRemoteForStPath() == null
+ ? optionsContainer.getLocalForStPath()
+ : new File("/");
+
if (CollectionUtil.isEmptyOrAllElementsNull(restoreStateHandles)) {
return new ForStNoneRestoreOperation(
- optionsContainer.getLocalForStPath(),
- dbOptions,
+ instanceForStPath,
+ optionsContainer.getDbOptions(),
columnFamilyOptionsFactory,
nativeMetricOptions,
metricGroup);
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 2b912ca9732..a5644c3d3b7 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
@@ -23,6 +23,7 @@ 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.core.fs.Path;
import org.apache.flink.runtime.memory.OpaqueMemoryResource;
import org.apache.flink.util.FileUtils;
import org.apache.flink.util.IOUtils;
@@ -48,9 +49,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;
import java.util.Arrays;
import java.util.Collection;
@@ -71,9 +70,9 @@ public final class ForStResourceContainer implements AutoCloseable {
private static final String DB_DIR_STRING = "db";
- @Nullable private final URI remoteBasePath;
+ @Nullable private final Path remoteBasePath;
- @Nullable private final URI remoteForStPath;
+ @Nullable private final Path remoteForStPath;
@Nullable private final File localBasePath;
@@ -97,7 +96,7 @@ public final class ForStResourceContainer implements AutoCloseable {
/** The handles to be closed when the container is closed. */
private final ArrayList<AutoCloseable> handlesToClose;
- @Nullable private Path relocatedDbLogBaseDir;
+ @Nullable private java.nio.file.Path relocatedDbLogBaseDir;
@VisibleForTesting
public ForStResourceContainer() {
@@ -121,7 +120,7 @@ public final class ForStResourceContainer implements AutoCloseable {
@Nullable ForStOptionsFactory optionsFactory,
@Nullable OpaqueMemoryResource<ForStSharedResources> sharedResources,
@Nullable File localBasePath,
- @Nullable URI remoteBasePath,
+ @Nullable Path remoteBasePath,
boolean enableStatistics) {
this.configuration = configuration;
@@ -132,7 +131,7 @@ public final class ForStResourceContainer implements AutoCloseable {
this.localForStPath = localBasePath != null ? new File(localBasePath, DB_DIR_STRING) : null;
this.remoteBasePath = remoteBasePath;
this.remoteForStPath =
- remoteBasePath != null ? remoteBasePath.resolve(DB_DIR_STRING) : null;
+ remoteBasePath != null ? new Path(remoteBasePath, DB_DIR_STRING) : null;
this.enableStatistics = enableStatistics;
this.handlesToClose = new ArrayList<>();
@@ -259,10 +258,15 @@ public final class ForStResourceContainer implements AutoCloseable {
}
@Nullable
- public URI getRemoteBasePath() {
+ public Path getRemoteBasePath() {
return remoteBasePath;
}
+ @Nullable
+ public Path getRemoteForStPath() {
+ return remoteForStPath;
+ }
+
/**
* Prepare local and remote directories.
*
@@ -273,28 +277,27 @@ public final class ForStResourceContainer implements AutoCloseable {
prepareDirectories(remoteBasePath, remoteForStPath);
}
if (localBasePath != null && localForStPath != null) {
- prepareDirectories(new URI(localBasePath.getPath()), new URI(localForStPath.getPath()));
+ prepareDirectories(
+ new Path(localBasePath.getPath()), new Path(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);
+ private static void prepareDirectories(Path basePath, Path dbPath) throws IOException {
+ FileSystem fileSystem = basePath.getFileSystem();
+ if (fileSystem.exists(basePath)) {
+ if (!fileSystem.getFileStatus(basePath).isDir()) {
+ throw new IOException("Not a directory: " + basePath);
}
- } else if (!fileSystem.mkdirs(tempBasePath)) {
+ } else if (!fileSystem.mkdirs(basePath)) {
throw new IOException(
- String.format("Could not create ForSt directory at %s.", tempBasePath));
+ String.format("Could not create ForSt directory at %s.", basePath));
}
- if (fileSystem.exists(tempDBPath)) {
- fileSystem.delete(tempDBPath, true);
+ if (fileSystem.exists(dbPath)) {
+ fileSystem.delete(dbPath, true);
}
- if (!fileSystem.mkdirs(tempDBPath)) {
+ if (!fileSystem.mkdirs(dbPath)) {
throw new IOException(
- String.format("Could not create ForSt db directory at %s.", tempDBPath));
+ String.format("Could not create ForSt db directory at %s.", dbPath));
}
}
@@ -308,15 +311,14 @@ public final class ForStResourceContainer implements AutoCloseable {
clearDirectories(remoteBasePath);
}
if (localBasePath != null) {
- clearDirectories(new URI(localBasePath.getPath()));
+ clearDirectories(new Path(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);
+ private static void clearDirectories(Path basePath) throws IOException {
+ FileSystem fileSystem = basePath.getFileSystem();
+ if (fileSystem.exists(basePath)) {
+ fileSystem.delete(basePath, true);
}
}
@@ -493,7 +495,27 @@ public final class ForStResourceContainer implements AutoCloseable {
String relocatedDbLogDir = logFile.getParent();
this.relocatedDbLogBaseDir = new File(relocatedDbLogDir).toPath();
dbOptions.setDbLogDir(relocatedDbLogDir);
+ } else {
+ setLocalForStPathAsLogDir(dbOptions);
}
+ } else {
+ setLocalForStPathAsLogDir(dbOptions);
+ }
+ }
+
+ private void setLocalForStPathAsLogDir(DBOptions dbOptions) {
+ // Currently, ForStDB does not support mixing local-dir and remote-dir, and ForStDB will
+ // concatenates the dfs directory with the local directory as working dir when using flink
+ // env. We expect to directly use the dfs directory in flink env or local directory as
+ // working dir. We will implement this in ForStDB later, but before that, we achieved this
+ // by setting the dbPath to "/" when the dfs directory existed. Another problem is that when
+ // the system property "log.file" is not set, ForSt directly uses the instance path as the
+ // log dir, which results in "/" being used as the log directory. This often has permission
+ // issues, so the db log dir is temporarily set explicitly here.
+ // TODO: remove this method after ForSt deal log dir well
+ if (localForStPath != null) {
+ this.relocatedDbLogBaseDir = localForStPath.toPath();
+ dbOptions.setDbLogDir(localForStPath.getPath());
}
}
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 c2521772b44..7528b4d5265 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,7 +50,6 @@ 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;
@@ -90,7 +89,7 @@ public class ForStStateBackend extends AbstractManagedMemoryStateBackend
* 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;
+ @Nullable private Path remoteForStDirectory;
/**
* Base paths for ForSt directory, as configured. Null if not yet set, in which case the
@@ -154,15 +153,7 @@ public class ForStStateBackend extends AbstractManagedMemoryStateBackend
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);
- }
+ this.remoteForStDirectory = remoteDirStr == null ? null : new Path(remoteDirStr);
}
// configure local directories
@@ -308,12 +299,17 @@ public class ForStStateBackend extends AbstractManagedMemoryStateBackend
lazyInitializeForJob(env, fileCompatibleIdentifier);
- String childPath =
- "job_" + jobId + "_op_" + fileCompatibleIdentifier + "_uuid_" + UUID.randomUUID();
+ String opChildPath =
+ String.format(
+ "op_%s_attempt_%s",
+ fileCompatibleIdentifier, env.getTaskInfo().getAttemptNumber());
- File localBasePath = new File(getNextStoragePath(), childPath);
- URI remoteBasePath =
- remoteForStDirectory != null ? remoteForStDirectory.resolve(childPath) : null;
+ File localBasePath =
+ new File(new File(getNextStoragePath(), jobId.toHexString()), opChildPath);
+ Path remoteBasePath =
+ remoteForStDirectory != null
+ ? new Path(new Path(remoteForStDirectory, jobId.toHexString()), opChildPath)
+ : null;
final OpaqueMemoryResource<ForStSharedResources> sharedResources =
ForStOperationUtils.allocateSharedCachesIfConfigured(
@@ -570,7 +566,7 @@ public class ForStStateBackend extends AbstractManagedMemoryStateBackend
private ForStResourceContainer createOptionsAndResourceContainer(
@Nullable OpaqueMemoryResource<ForStSharedResources> sharedResources,
@Nullable File localBasePath,
- @Nullable URI remoteBasePath,
+ @Nullable Path remoteBasePath,
boolean enableStatistics) {
return new ForStResourceContainer(
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 bec699931c6..a3d4e7fa979 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
@@ -19,6 +19,7 @@
package org.apache.flink.state.forst;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.Path;
import org.apache.flink.runtime.memory.OpaqueMemoryResource;
import org.apache.flink.util.function.ThrowingRunnable;
@@ -43,7 +44,6 @@ 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;
@@ -303,18 +303,18 @@ public class ForStResourceContainerTest {
@Test
public void testDirectoryResources() throws Exception {
File localBasePath = TMP_FOLDER.newFolder();
- URI remoteBasePath = TMP_FOLDER.newFolder().toURI();
+ Path remoteBasePath = new Path(TMP_FOLDER.newFolder().getPath());
try (final ForStResourceContainer optionsContainer =
new ForStResourceContainer(
new Configuration(), null, null, localBasePath, remoteBasePath, false)) {
optionsContainer.prepareDirectories();
assertTrue(localBasePath.exists());
- assertTrue(new File(remoteBasePath).exists());
+ assertTrue(new File(remoteBasePath.getPath()).exists());
assertTrue(optionsContainer.getDbOptions().getEnv() instanceof FlinkEnv);
optionsContainer.clearDirectories();
assertFalse(localBasePath.exists());
- assertFalse(new File(remoteBasePath).exists());
+ assertFalse(new File(remoteBasePath.getPath()).exists());
}
}
}