This is an automated email from the ASF dual-hosted git repository. zakelly pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit ae52bf2d6cc9723a1af3b1119b3fcee35603d56b Author: Zakelly <[email protected]> AuthorDate: Tue Dec 3 22:02:30 2024 +0800 [FLINK-36838][state] Join background threads when ForSt state backend quit --- .../apache/flink/state/forst/ForStResourceContainer.java | 15 ++++++++++++++- 1 file changed, 14 insertions(+), 1 deletion(-) 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 b807e4d6c4b..377496d40c8 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 @@ -39,6 +39,7 @@ import org.forstdb.Filter; import org.forstdb.FlinkEnv; import org.forstdb.IndexType; import org.forstdb.PlainTableConfig; +import org.forstdb.Priority; import org.forstdb.ReadOptions; import org.forstdb.Statistics; import org.forstdb.TableFormatConfig; @@ -76,6 +77,8 @@ 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(); + @Nullable private FlinkEnv flinkEnv = null; + @Nullable private final Path remoteBasePath; @Nullable private final Path remoteForStPath; @@ -187,7 +190,8 @@ public final class ForStResourceContainer implements AutoCloseable { // configured, // fallback to local directory currently temporarily. if (remoteForStPath != null) { - opt.setEnv(new FlinkEnv(remoteForStPath.toString(), forstFileSystem)); + flinkEnv = new FlinkEnv(remoteForStPath.toString(), forstFileSystem); + opt.setEnv(flinkEnv); } return opt; @@ -408,6 +412,15 @@ public final class ForStResourceContainer implements AutoCloseable { sharedResources.close(); } cleanRelocatedDbLogs(); + if (flinkEnv != null) { + // There is something wrong with the FlinkEnv, the background threads won't quit during + // the disposal of DB. We explicit shrink the thread pool here until the ForSt repo + // fixes that. + flinkEnv.setBackgroundThreads(0, Priority.LOW); + flinkEnv.setBackgroundThreads(0, Priority.HIGH); + flinkEnv.close(); + flinkEnv = null; + } } /**
