[FLINK-31238] Deactivate parts of the code until new FRocksDB release…

… is available.

Then this commit should be reverted.

flink-state-backends/flink-statebackend-rocksdb/pom.xml

@@ -61,9 +61,9 @@ under the License.
  </dependency>
 
  <dependency>
- <groupId>io.github.fredia</groupId>
+ <groupId>com.ververica</groupId>
  <artifactId>frocksdbjni</artifactId>
- <version>8.6.7-ververica-test-1.0</version>
+ <version>6.20.3-ververica-2.0</version>
  </dependency>
 
  <!-- test dependencies -->

flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBIncrementalCheckpointUtils.java

@@ -20,17 +20,11 @@
 import org.apache.flink.runtime.state.CompositeKeySerializationUtils;
 import org.apache.flink.runtime.state.KeyGroupRange;
 import org.apache.flink.runtime.state.KeyedStateHandle;
-import org.apache.flink.runtime.state.RegisteredStateMetaInfoBase;
-import org.apache.flink.util.IOUtils;
-import org.apache.flink.util.Preconditions;
 import org.apache.flink.util.function.RunnableWithException;
 
 import org.apache.flink.shaded.guava31.com.google.common.primitives.UnsignedBytes;
 
-import org.rocksdb.Checkpoint;
 import org.rocksdb.ColumnFamilyHandle;
-import org.rocksdb.CompactRangeOptions;
-import org.rocksdb.ExportImportFilesMetaData;
 import org.rocksdb.LiveFileMetaData;
 import org.rocksdb.RocksDB;
 import org.rocksdb.RocksDBException;
@@ -39,16 +33,11 @@
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 
-import java.io.File;
-import java.nio.file.Path;
-import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Comparator;
 import java.util.Iterator;
 import java.util.List;
-import java.util.Map;
 import java.util.Optional;
-import java.util.UUID;
 
 /** Utils for RocksDB Incremental Checkpoint. */
 public class RocksDBIncrementalCheckpointUtils {
@@ -218,6 +207,7 @@ public static Optional<RunnableWithException> createRangeCompactionTaskIfNeeded(
 
  return Optional.of(
  () -> {
+ /*
  try (CompactRangeOptions compactionOptions =
  new CompactRangeOptions()
  .setExclusiveManualCompaction(true)
@@ -251,6 +241,7 @@ public static Optional<RunnableWithException> createRangeCompactionTaskIfNeeded(
  }
  }
  }
+ */
  });
  }
 
@@ -321,6 +312,7 @@ private static KeyRange getDBKeyRange(RocksDB db) {
  * @param resultOutput output parameter for the metadata of the export.
  * @throws RocksDBException on problems inside RocksDB.
  */
+ /*
  public static void exportColumnFamilies(
  RocksDB db,
  List<ColumnFamilyHandle> columnFamilyHandles,
@@ -357,6 +349,7 @@ public static void exportColumnFamilies(
  }
  }
  }
+ */
 
  /** check whether the bytes is before prefixBytes in the character order. */
  public static boolean beforeThePrefixBytes(@Nonnull byte[] bytes, @Nonnull byte[] prefixBytes) {

flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBOperationUtils.java

@@ -31,8 +31,6 @@
 import org.rocksdb.ColumnFamilyHandle;
 import org.rocksdb.ColumnFamilyOptions;
 import org.rocksdb.DBOptions;
-import org.rocksdb.ExportImportFilesMetaData;
-import org.rocksdb.ImportColumnFamilyOptions;
 import org.rocksdb.ReadOptions;
 import org.rocksdb.RocksDB;
 import org.rocksdb.RocksDBException;
@@ -44,7 +42,6 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
-import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.function.Function;
@@ -129,8 +126,7 @@ public static RocksDBKeyedStateBackend.RocksDbKvStateInfo createStateInfo(
  RocksDB db,
  Function<String, ColumnFamilyOptions> columnFamilyOptionsFactory,
  @Nullable RocksDbTtlCompactFiltersManager ttlCompactFiltersManager,
- @Nullable Long writeBufferManagerCapacity,
- List<ExportImportFilesMetaData> importFilesMetaData) {
+ @Nullable Long writeBufferManagerCapacity) {
 
  ColumnFamilyDescriptor columnFamilyDescriptor =
  createColumnFamilyDescriptor(
@@ -141,8 +137,7 @@ public static RocksDBKeyedStateBackend.RocksDbKvStateInfo createStateInfo(
 
  final ColumnFamilyHandle columnFamilyHandle;
  try {
- columnFamilyHandle =
- createColumnFamily(columnFamilyDescriptor, db, importFilesMetaData);
+ columnFamilyHandle = createColumnFamily(columnFamilyDescriptor, db);
  } catch (Exception ex) {
  IOUtils.closeQuietly(columnFamilyDescriptor.getOptions());
  throw new FlinkRuntimeException("Error creating ColumnFamilyHandle.", ex);
@@ -151,21 +146,6 @@ public static RocksDBKeyedStateBackend.RocksDbKvStateInfo createStateInfo(
  return new RocksDBKeyedStateBackend.RocksDbKvStateInfo(columnFamilyHandle, metaInfoBase);
  }
 
- public static RocksDBKeyedStateBackend.RocksDbKvStateInfo createStateInfo(
- RegisteredStateMetaInfoBase metaInfoBase,
- RocksDB db,
- Function<String, ColumnFamilyOptions> columnFamilyOptionsFactory,
- @Nullable RocksDbTtlCompactFiltersManager ttlCompactFiltersManager,
- @Nullable Long writeBufferManagerCapacity) {
- return createStateInfo(
- metaInfoBase,
- db,
- columnFamilyOptionsFactory,
- ttlCompactFiltersManager,
- writeBufferManagerCapacity,
- Collections.emptyList());
- }
-
  /**
  * Creates a column descriptor for a state column family.
  *
@@ -253,20 +233,8 @@ public static ColumnFamilyOptions createColumnFamilyOptions(
  }
 
  private static ColumnFamilyHandle createColumnFamily(
- ColumnFamilyDescriptor columnDescriptor,
- RocksDB db,
- List<ExportImportFilesMetaData> importFilesMetaData)
- throws RocksDBException {
-
- if (importFilesMetaData.isEmpty()) {
- return db.createColumnFamily(columnDescriptor);
- } else {
- try (ImportColumnFamilyOptions importColumnFamilyOptions =
- new ImportColumnFamilyOptions().setMoveFiles(true)) {
- return db.createColumnFamilyWithImport(
- columnDescriptor, importColumnFamilyOptions, importFilesMetaData);
- }
- }
+ ColumnFamilyDescriptor columnDescriptor, RocksDB db) throws RocksDBException {
+ return db.createColumnFamily(columnDescriptor);
  }
 
  public static void addColumnFamilyOptionsToCloseLater(

flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBHandle.java

@@ -28,13 +28,11 @@
 import org.apache.flink.runtime.state.metainfo.StateMetaInfoSnapshot;
 import org.apache.flink.util.FileUtils;
 import org.apache.flink.util.IOUtils;
-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.ExportImportFilesMetaData;
 import org.rocksdb.RocksDB;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -193,6 +191,7 @@ RocksDbKvStateInfo getOrRegisterStateColumnFamilyHandle(
  * @param stateMetaInfo info about the state to create.
  * @param cfMetaDataList the data to import.
  */
+ /*
  void registerStateColumnFamilyHandleWithImport(
  RegisteredStateMetaInfoBase stateMetaInfo,
  List<ExportImportFilesMetaData> cfMetaDataList) {
@@ -213,6 +212,7 @@ void registerStateColumnFamilyHandleWithImport(
 
  columnFamilyHandles.add(stateInfo.columnFamilyHandle);
  }
+ */
 
  /**
  * This recreates the new working directory of the recovered RocksDB instance and links/copies

flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java

@@ -58,7 +58,6 @@
 import org.rocksdb.ColumnFamilyHandle;
 import org.rocksdb.ColumnFamilyOptions;
 import org.rocksdb.DBOptions;
-import org.rocksdb.ExportImportFilesMetaData;
 import org.rocksdb.ReadOptions;
 import org.rocksdb.RocksDB;
 import org.rocksdb.RocksDBException;
@@ -71,11 +70,9 @@
 import java.io.File;
 import java.io.IOException;
 import java.io.InputStream;
-import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.Collection;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -86,7 +83,6 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.function.Function;
-import java.util.stream.Collectors;
 
 import static org.apache.flink.runtime.state.StateUtil.unexpectedStateHandleException;
 
@@ -167,8 +163,10 @@ public RocksDBIncrementalRestoreOperation(
  this.keyGroupPrefixBytes = keyGroupPrefixBytes;
  this.keySerializerProvider = keySerializerProvider;
  this.userCodeClassLoader = userCodeClassLoader;
- this.useIngestDbRestoreMode = useIngestDbRestoreMode;
- this.asyncCompactAfterRescale = asyncCompactAfterRescale;
+ // this.useIngestDbRestoreMode = useIngestDbRestoreMode;
+ // this.asyncCompactAfterRescale = asyncCompactAfterRescale;
+ this.useIngestDbRestoreMode = false;
+ this.asyncCompactAfterRescale = false;
  }
 
  /** Root method that branches for different implementations of {@link KeyedStateHandle}. */
@@ -408,6 +406,7 @@ private void mergeStateHandlesWithClipAndIngest(
  byte[] stopKeyGroupPrefixBytes)
  throws Exception {
 
+ /*
  final Path absolutInstanceBasePath = instanceBasePath.getAbsoluteFile().toPath();
  final Path exportCfBasePath = absolutInstanceBasePath.resolve("export-cfs");
  Files.createDirectories(exportCfBasePath);
@@ -447,6 +446,7 @@ private void mergeStateHandlesWithClipAndIngest(
  // Cleanup export base directory
  cleanUpPathQuietly(exportCfBasePath);
  }
+ */
  }
 
  /**
@@ -462,6 +462,7 @@ private void mergeStateHandlesWithClipAndIngest(
  * @return the total key-groups range of the exported data.
  * @throws Exception on any export error.
  */
+ /*
  private KeyGroupRange exportColumnFamiliesWithSstDataInKeyGroupsRange(
  Path exportCfBasePath,
  List<IncrementalLocalKeyedStateHandle> localKeyedStateHandles,
@@ -542,6 +543,7 @@ private KeyGroupRange exportColumnFamiliesWithSstDataInKeyGroupsRange(
  ? new KeyGroupRange(minExportKeyGroup, maxExportKeyGroup)
  : KeyGroupRange.EMPTY_KEY_GROUP_RANGE;
  }
+ */
 
  /**
  * Helper method that merges the data from multiple state handles into the restoring base DB by
@@ -594,6 +596,7 @@ private void mergeStateHandlesWithCopyFromTemporaryInstance(
  * @param exportKeyGroupRange the total key-groups range of the exported data.
  * @throws Exception on import error.
  */
+ /*
  private void initBaseDBFromColumnFamilyImports(
  Map<RegisteredStateMetaInfoBase, List<ExportImportFilesMetaData>>
  exportedColumnFamilyMetaData,
@@ -622,6 +625,7 @@ private void initBaseDBFromColumnFamilyImports(
  keyGroupRange,
  operatorIdentifier);
  }
+ */
 
  /**
  * Restores the checkpointing status and state for this backend. This can only be done if the