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
The following commit(s) were added to refs/heads/master by this push:
new 8fc9538b0dd [FLINK-40952][state/forst] Close the DBOptions copy of
temporary restore DBs (#29430)
8fc9538b0dd is described below
commit 8fc9538b0dd67b76d08a369ba0a28cfed390f5b2
Author: YordanPavlov <[email protected]>
AuthorDate: Fri Oct 9 18:35:02 2026 +0300
[FLINK-40952][state/forst] Close the DBOptions copy of temporary restore
DBs (#29430)
Generated-by: Claude Code (Claude Opus 5.5)
---
.../restore/ForStIncrementalRestoreOperation.java | 30 ++++--
.../ForStIncrementalCheckpointRescalingTest.java | 105 +++++++++++++++++++++
2 files changed, 127 insertions(+), 8 deletions(-)
diff --git
a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/restore/ForStIncrementalRestoreOperation.java
b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/restore/ForStIncrementalRestoreOperation.java
index ca3d683a484..a634c36d5a4 100644
---
a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/restore/ForStIncrementalRestoreOperation.java
+++
b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/restore/ForStIncrementalRestoreOperation.java
@@ -964,17 +964,24 @@ public class ForStIncrementalRestoreOperation<K>
implements ForStRestoreOperatio
DBOptions dbOptions = new DBOptions(this.forstHandle.getDbOptions());
dbOptions.setDbLogDir("");
- RocksDB restoreDb =
- ForStOperationUtils.openDB(
- dbName,
- columnFamilyDescriptors,
- columnFamilyHandles,
- createColumnFamilyOptions(
-
this.forstHandle.getColumnFamilyOptionsFactory(), "default"),
- dbOptions);
+ RocksDB restoreDb;
+ try {
+ restoreDb =
+ ForStOperationUtils.openDB(
+ dbName,
+ columnFamilyDescriptors,
+ columnFamilyHandles,
+ createColumnFamilyOptions(
+
this.forstHandle.getColumnFamilyOptionsFactory(), "default"),
+ dbOptions);
+ } catch (Exception e) {
+ IOUtils.closeQuietly(dbOptions);
+ throw e;
+ }
return new RestoredDBInstance(
restoreDb,
+ dbOptions,
columnFamilyHandles,
columnFamilyDescriptors,
stateMetaInfoSnapshots,
@@ -995,6 +1002,10 @@ public class ForStIncrementalRestoreOperation<K>
implements ForStRestoreOperatio
@Nonnull final RocksDB db;
+ // A native copy that shares the backend's resources, e.g. the
slot-shared write buffer
+ // manager and block cache with managed memory, so it must be closed
with the DB.
+ @Nonnull final DBOptions dbOptions;
+
@Nonnull final ColumnFamilyHandle defaultColumnFamilyHandle;
@Nonnull final List<ColumnFamilyHandle> columnFamilyHandles;
@@ -1011,12 +1022,14 @@ public class ForStIncrementalRestoreOperation<K>
implements ForStRestoreOperatio
RestoredDBInstance(
@Nonnull RocksDB db,
+ @Nonnull DBOptions dbOptions,
@Nonnull List<ColumnFamilyHandle> columnFamilyHandles,
@Nonnull List<ColumnFamilyDescriptor> columnFamilyDescriptors,
@Nonnull List<StateMetaInfoSnapshot> stateMetaInfoSnapshots,
@Nonnull IncrementalRemoteKeyedStateHandle srcStateHandle,
@Nonnull String instancePath) {
this.db = db;
+ this.dbOptions = dbOptions;
this.defaultColumnFamilyHandle = columnFamilyHandles.remove(0);
this.columnFamilyHandles = columnFamilyHandles;
this.columnFamilyDescriptors = columnFamilyDescriptors;
@@ -1036,6 +1049,7 @@ public class ForStIncrementalRestoreOperation<K>
implements ForStRestoreOperatio
IOUtils.closeQuietly(defaultColumnFamilyHandle);
IOUtils.closeAllQuietly(columnFamilyHandles);
IOUtils.closeQuietly(db);
+ IOUtils.closeQuietly(dbOptions);
IOUtils.closeAllQuietly(columnFamilyOptions);
IOUtils.closeQuietly(readOptions);
}
diff --git
a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStIncrementalCheckpointRescalingTest.java
b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStIncrementalCheckpointRescalingTest.java
index 4fb2f3f7e34..a12fc7a1f0e 100644
---
a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStIncrementalCheckpointRescalingTest.java
+++
b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStIncrementalCheckpointRescalingTest.java
@@ -24,6 +24,7 @@ import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.testutils.CommonTestUtils;
import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
import org.apache.flink.runtime.checkpoint.StateAssignmentOperation;
import org.apache.flink.runtime.state.KeyGroupRange;
@@ -40,13 +41,23 @@ import
org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTe
import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
import org.apache.flink.util.Collector;
+import org.forstdb.ColumnFamilyOptions;
+import org.forstdb.DBOptions;
+import org.forstdb.InfoLogLevel;
+import org.forstdb.Logger;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.TestTemplate;
import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.jupiter.api.io.TempDir;
+import javax.annotation.Nullable;
+
+import java.lang.ref.WeakReference;
import java.nio.file.Path;
+import java.time.Duration;
+import java.util.Collection;
import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
import static
org.apache.flink.state.forst.ForStConfigurableOptions.USE_INGEST_DB_RESTORE_MODE;
import static org.assertj.core.api.Assertions.assertThat;
@@ -419,6 +430,70 @@ class ForStIncrementalCheckpointRescalingTest {
}
}
+ /**
+ * A scale-in restore opens temporary DBs with copies of the backend's DB
options. The native
+ * options keep everything set on them alive (here a logger; in production
the shared write
+ * buffer manager and block cache), so every copy must be closed with its
temporary DB.
+ */
+ @TestTemplate
+ void testScalingDownReleasesDBOptions() throws Exception {
+ TrackingLoggerOptionsFactory.LOGGERS.clear();
+
+ scaleDownFromTwoToOne();
+
+ assertThat(TrackingLoggerOptionsFactory.LOGGERS).hasSize(3);
+ CommonTestUtils.waitUtil(
+ () -> {
+ System.gc();
+ return TrackingLoggerOptionsFactory.LOGGERS.stream()
+ .allMatch(logger -> logger.get() == null);
+ },
+ Duration.ofSeconds(30),
+ Duration.ofMillis(100),
+ "The logger of a disposed backend is still referenced by
native DB options.");
+ }
+
+ /** Runs in its own frame so that no harness stays reachable from the
caller. */
+ @SuppressWarnings("unchecked")
+ private void scaleDownFromTwoToOne() throws Exception {
+ KeyedOneInputStreamOperatorTestHarness<String, String, Integer>[]
harness2 =
+ new KeyedOneInputStreamOperatorTestHarness[2];
+ OperatorSubtaskState snapshot2;
+ try {
+ for (int i = 0; i < 2; i++) {
+ harness2[i] = getHarnessTest(keySelector, maxParallelism, 2,
i);
+ harness2[i].setStateBackend(getStateBackend(new
TrackingLoggerOptionsFactory()));
+ harness2[i].setCheckpointStorage(
+ new FileSystemCheckpointStorage("file://" +
rootFolder.toAbsolutePath()));
+ harness2[i].open();
+ }
+ validHarnessResult(
+ harness2[0], 1, records[0], records[1], records[2],
records[3], records[4]);
+ validHarnessResult(
+ harness2[1], 1, records[5], records[6], records[7],
records[8], records[9]);
+ snapshot2 =
+ AbstractStreamOperatorTestHarness.repackageState(
+ harness2[0].snapshot(0, 0),
harness2[1].snapshot(0, 0));
+ } finally {
+ closeHarness(harness2);
+ }
+
+ OperatorSubtaskState initState =
+ AbstractStreamOperatorTestHarness.repartitionOperatorState(
+ snapshot2, maxParallelism, 2, 1, 0);
+ try (KeyedOneInputStreamOperatorTestHarness<String, String, Integer>
harness =
+ getHarnessTest(keySelector, maxParallelism, 1, 0)) {
+ harness.setStateBackend(getStateBackend(new
TrackingLoggerOptionsFactory()));
+ harness.setCheckpointStorage(
+ new FileSystemCheckpointStorage("file://" +
rootFolder.toAbsolutePath()));
+ harness.setup();
+ harness.initializeState(initState);
+ harness.open();
+
+ validHarnessResult(harness, 2, records);
+ }
+ }
+
private void closeHarness(KeyedOneInputStreamOperatorTestHarness<?, ?,
?>[] harnessArr)
throws Exception {
for (KeyedOneInputStreamOperatorTestHarness<?, ?, ?> harness :
harnessArr) {
@@ -458,7 +533,12 @@ class ForStIncrementalCheckpointRescalingTest {
}
private StateBackend getStateBackend() {
+ return getStateBackend(null);
+ }
+
+ private StateBackend getStateBackend(@Nullable ForStOptionsFactory
optionsFactory) {
ForStStateBackend forStStateBackend = new ForStStateBackend();
+ forStStateBackend.setForStOptions(optionsFactory);
Configuration configuration = new Configuration();
configuration.set(USE_INGEST_DB_RESTORE_MODE, useIngestDbRestoreMode);
return forStStateBackend.configure(configuration,
getClass().getClassLoader());
@@ -487,6 +567,31 @@ class ForStIncrementalCheckpointRescalingTest {
}
}
+ /** Sets a tracked logger on the DB options of every backend. */
+ private static class TrackingLoggerOptionsFactory implements
ForStOptionsFactory {
+
+ private static final List<WeakReference<Logger>> LOGGERS = new
CopyOnWriteArrayList<>();
+
+ @Override
+ public DBOptions createDBOptions(
+ DBOptions currentOptions, Collection<AutoCloseable>
handlesToClose) {
+ Logger logger =
+ new Logger(currentOptions) {
+ @Override
+ protected void log(InfoLogLevel infoLogLevel, String
logMsg) {}
+ };
+ handlesToClose.add(logger);
+ LOGGERS.add(new WeakReference<>(logger));
+ return currentOptions.setLogger(logger);
+ }
+
+ @Override
+ public ColumnFamilyOptions createColumnOptions(
+ ColumnFamilyOptions currentOptions, Collection<AutoCloseable>
handlesToClose) {
+ return currentOptions;
+ }
+ }
+
/** A simple key selector for tests. */
private class TestKeySelector implements KeySelector<String, String> {
@Override
