You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by sr...@apache.org on 2024/02/28 08:33:58 UTC
(flink) branch master updated (b9069227098 -> e2b6588ef39)
This is an automated email from the ASF dual-hosted git repository.
srichter pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from b9069227098 [FLINK-34454][doc/annotation] Use claim mode instead of restore mode everywhere
new 5be8b3e1dfd [hotfix] Fix configuration through TernaryBoolean in EmbeddedRocksDBStateBackend.
new e2b6588ef39 [hotfix] Reduce code duplication and add logging in RocksDBIncrementalCheckpointUtils#deleteRangeData.
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:
.../java/org/apache/flink/util/TernaryBoolean.java | 18 ++++++
.../state/EmbeddedRocksDBStateBackend.java | 21 ++++---
.../state/RocksDBIncrementalCheckpointUtils.java | 66 +++++++++++++++-------
.../state/EmbeddedRocksDBStateBackendTest.java | 57 ++++++++++++++++++-
.../state/RocksDBStateBackendConfigTest.java | 25 ++++++++
5 files changed, 153 insertions(+), 34 deletions(-)
(flink) 02/02: [hotfix] Reduce code duplication and add logging in RocksDBIncrementalCheckpointUtils#deleteRangeData.
Posted by sr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
srichter pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit e2b6588ef39cbeb1825cacf553a80272ae7b9418
Author: Stefan Richter <sr...@confluent.io>
AuthorDate: Tue Feb 27 11:54:28 2024 +0100
[hotfix] Reduce code duplication and add logging in RocksDBIncrementalCheckpointUtils#deleteRangeData.
---
.../state/RocksDBIncrementalCheckpointUtils.java | 66 +++++++++++++++-------
1 file changed, 46 insertions(+), 20 deletions(-)
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBIncrementalCheckpointUtils.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBIncrementalCheckpointUtils.java
index bdb94466de9..67bf02cd31a 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBIncrementalCheckpointUtils.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBIncrementalCheckpointUtils.java
@@ -29,21 +29,28 @@ import org.rocksdb.ColumnFamilyHandle;
import org.rocksdb.LiveFileMetaData;
import org.rocksdb.RocksDB;
import org.rocksdb.RocksDBException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import javax.annotation.Nonnegative;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collection;
import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
import java.util.Optional;
+import java.util.stream.Collectors;
/** Utils for RocksDB Incremental Checkpoint. */
public class RocksDBIncrementalCheckpointUtils {
+ private static final Logger logger =
+ LoggerFactory.getLogger(RocksDBIncrementalCheckpointUtils.class);
+
/**
* Evaluates state handle's "score" regarding the target range when choosing the best state
* handle to init the initial db for recovery, if the overlap fraction is less than
@@ -112,7 +119,7 @@ public class RocksDBIncrementalCheckpointUtils {
* @param targetKeyGroupRange the target key group range.
* @param currentKeyGroupRange the key group range of the db instance.
* @param keyGroupPrefixBytes Number of bytes required to prefix the key groups.
- * @param useDeleteFilesInRange Whether to use deleteFilesInRange to clean up redundant files.
+ * @param useDeleteFilesInRange whether to call db.deleteFilesInRanges for the deleted ranges.
*/
public static void clipDBWithKeyGroupRange(
@Nonnull RocksDB db,
@@ -122,31 +129,45 @@ public class RocksDBIncrementalCheckpointUtils {
@Nonnegative int keyGroupPrefixBytes,
boolean useDeleteFilesInRange)
throws RocksDBException {
- List<byte[]> deletedRanges = new ArrayList<>(4);
+
+ List<byte[]> deleteFilesRanges = new ArrayList<>(4);
if (currentKeyGroupRange.getStartKeyGroup() < targetKeyGroupRange.getStartKeyGroup()) {
- final byte[] beginKeyGroupBytes = new byte[keyGroupPrefixBytes];
- final byte[] endKeyGroupBytes = new byte[keyGroupPrefixBytes];
- CompositeKeySerializationUtils.serializeKeyGroup(
- currentKeyGroupRange.getStartKeyGroup(), beginKeyGroupBytes);
- CompositeKeySerializationUtils.serializeKeyGroup(
- targetKeyGroupRange.getStartKeyGroup(), endKeyGroupBytes);
- deletedRanges.add(beginKeyGroupBytes);
- deletedRanges.add(endKeyGroupBytes);
+ prepareRangeDeletes(
+ keyGroupPrefixBytes,
+ currentKeyGroupRange.getStartKeyGroup(),
+ targetKeyGroupRange.getStartKeyGroup(),
+ deleteFilesRanges);
}
if (currentKeyGroupRange.getEndKeyGroup() > targetKeyGroupRange.getEndKeyGroup()) {
- final byte[] beginKeyGroupBytes = new byte[keyGroupPrefixBytes];
- final byte[] endKeyGroupBytes = new byte[keyGroupPrefixBytes];
- CompositeKeySerializationUtils.serializeKeyGroup(
- targetKeyGroupRange.getEndKeyGroup() + 1, beginKeyGroupBytes);
- CompositeKeySerializationUtils.serializeKeyGroup(
- currentKeyGroupRange.getEndKeyGroup() + 1, endKeyGroupBytes);
- deletedRanges.add(beginKeyGroupBytes);
- deletedRanges.add(endKeyGroupBytes);
+ prepareRangeDeletes(
+ keyGroupPrefixBytes,
+ targetKeyGroupRange.getEndKeyGroup() + 1,
+ currentKeyGroupRange.getEndKeyGroup() + 1,
+ deleteFilesRanges);
}
- deleteRangeData(db, columnFamilyHandles, deletedRanges, useDeleteFilesInRange);
+ logger.info(
+ "Performing range delete for backend with target key-groups range {} with boundaries set {} - deleteFilesInRanges = {}.",
+ targetKeyGroupRange.prettyPrintInterval(),
+ deleteFilesRanges.stream().map(Arrays::toString).collect(Collectors.toList()),
+ useDeleteFilesInRange);
+
+ deleteRangeData(db, columnFamilyHandles, deleteFilesRanges, useDeleteFilesInRange);
+ }
+
+ private static void prepareRangeDeletes(
+ int keyGroupPrefixBytes,
+ int beginKeyGroup,
+ int endKeyGroup,
+ List<byte[]> deleteFilesRangesOut) {
+ byte[] beginKeyGroupBytes = new byte[keyGroupPrefixBytes];
+ byte[] endKeyGroupBytes = new byte[keyGroupPrefixBytes];
+ CompositeKeySerializationUtils.serializeKeyGroup(beginKeyGroup, beginKeyGroupBytes);
+ CompositeKeySerializationUtils.serializeKeyGroup(endKeyGroup, endKeyGroupBytes);
+ deleteFilesRangesOut.add(beginKeyGroupBytes);
+ deleteFilesRangesOut.add(endKeyGroupBytes);
}
/**
@@ -156,7 +177,7 @@ public class RocksDBIncrementalCheckpointUtils {
* @param columnFamilyHandles the column family need to be clipped.
* @param deleteRanges - pairs of deleted ranges (from1, to1, from2, to2, ...). For each pair
* [from, to), the startKey ('from') is inclusive, the endKey ('to') is exclusive.
- * @param useDeleteFilesInRange Whether to use deleteFilesInRange to clean up redundant files.
+ * @param useDeleteFilesInRange whether to use deleteFilesInRange to clean up redundant files.
*/
private static void deleteRangeData(
RocksDB db,
@@ -165,6 +186,11 @@ public class RocksDBIncrementalCheckpointUtils {
boolean useDeleteFilesInRange)
throws RocksDBException {
+ if (deleteRanges.isEmpty()) {
+ // nothing to do.
+ return;
+ }
+
Preconditions.checkArgument(deleteRanges.size() % 2 == 0);
for (ColumnFamilyHandle columnFamilyHandle : columnFamilyHandles) {
// First delete the files in ranges
(flink) 01/02: [hotfix] Fix configuration through TernaryBoolean in EmbeddedRocksDBStateBackend.
Posted by sr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
srichter pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 5be8b3e1dfdb5a009677b2e194def2ad3b1122ec
Author: Stefan Richter <sr...@confluent.io>
AuthorDate: Tue Feb 27 10:48:40 2024 +0100
[hotfix] Fix configuration through TernaryBoolean in EmbeddedRocksDBStateBackend.
---
.../java/org/apache/flink/util/TernaryBoolean.java | 18 +++++++
.../state/EmbeddedRocksDBStateBackend.java | 21 ++++----
.../state/EmbeddedRocksDBStateBackendTest.java | 57 ++++++++++++++++++++--
.../state/RocksDBStateBackendConfigTest.java | 25 ++++++++++
4 files changed, 107 insertions(+), 14 deletions(-)
diff --git a/flink-core/src/main/java/org/apache/flink/util/TernaryBoolean.java b/flink-core/src/main/java/org/apache/flink/util/TernaryBoolean.java
index 4cfd71ac238..4f06c15e9f8 100644
--- a/flink-core/src/main/java/org/apache/flink/util/TernaryBoolean.java
+++ b/flink-core/src/main/java/org/apache/flink/util/TernaryBoolean.java
@@ -19,6 +19,8 @@
package org.apache.flink.util;
import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ReadableConfig;
import javax.annotation.Nullable;
@@ -71,6 +73,22 @@ public enum TernaryBoolean {
return this == UNDEFINED ? null : (this == TRUE ? Boolean.TRUE : Boolean.FALSE);
}
+ /**
+ * Merges an existing value with a config, accepting the config's value only if the existing
+ * value is undefined.
+ *
+ * @param original the value to merge with the config.
+ * @param configOption the config option to merge with from the config.
+ * @param config the config to merge with.
+ */
+ public static TernaryBoolean mergeTernaryBooleanWithConfig(
+ TernaryBoolean original, ConfigOption<Boolean> configOption, ReadableConfig config) {
+ if (original != TernaryBoolean.UNDEFINED) {
+ return original;
+ }
+ return TernaryBoolean.fromBoxedBoolean(config.getOptional(configOption).orElse(null));
+ }
+
// ------------------------------------------------------------------------
/**
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/EmbeddedRocksDBStateBackend.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/EmbeddedRocksDBStateBackend.java
index 1a489fad8c8..baca5f4a013 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/EmbeddedRocksDBStateBackend.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/EmbeddedRocksDBStateBackend.java
@@ -321,21 +321,20 @@ public class EmbeddedRocksDBStateBackend extends AbstractManagedMemoryStateBacke
"Overlap fraction threshold of restoring should be between 0 and 1");
incrementalRestoreAsyncCompactAfterRescale =
- original.incrementalRestoreAsyncCompactAfterRescale == TernaryBoolean.UNDEFINED
- ? TernaryBoolean.fromBoxedBoolean(
- config.get(INCREMENTAL_RESTORE_ASYNC_COMPACT_AFTER_RESCALE))
- : original.incrementalRestoreAsyncCompactAfterRescale;
+ TernaryBoolean.mergeTernaryBooleanWithConfig(
+ original.incrementalRestoreAsyncCompactAfterRescale,
+ INCREMENTAL_RESTORE_ASYNC_COMPACT_AFTER_RESCALE,
+ config);
useIngestDbRestoreMode =
- original.useIngestDbRestoreMode == TernaryBoolean.UNDEFINED
- ? TernaryBoolean.fromBoxedBoolean(config.get(USE_INGEST_DB_RESTORE_MODE))
- : TernaryBoolean.fromBoolean(original.getUseIngestDbRestoreMode());
+ TernaryBoolean.mergeTernaryBooleanWithConfig(
+ original.useIngestDbRestoreMode, USE_INGEST_DB_RESTORE_MODE, config);
rescalingUseDeleteFilesInRange =
- original.rescalingUseDeleteFilesInRange == TernaryBoolean.UNDEFINED
- ? TernaryBoolean.fromBoxedBoolean(
- config.get(USE_DELETE_FILES_IN_RANGE_DURING_RESCALING))
- : original.rescalingUseDeleteFilesInRange;
+ TernaryBoolean.mergeTernaryBooleanWithConfig(
+ original.rescalingUseDeleteFilesInRange,
+ USE_DELETE_FILES_IN_RANGE_DURING_RESCALING,
+ config);
this.rocksDBMemoryFactory = original.rocksDBMemoryFactory;
}
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/EmbeddedRocksDBStateBackendTest.java b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/EmbeddedRocksDBStateBackendTest.java
index 5feeeaf5608..41773e4be28 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/EmbeddedRocksDBStateBackendTest.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/EmbeddedRocksDBStateBackendTest.java
@@ -27,6 +27,7 @@ import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.common.typeutils.base.IntSerializer;
import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.Path;
import org.apache.flink.core.testutils.OneShotLatch;
@@ -92,9 +93,12 @@ import java.util.Queue;
import java.util.concurrent.RunnableFuture;
import java.util.stream.Collectors;
+import static org.apache.flink.contrib.streaming.state.RocksDBConfigurableOptions.INCREMENTAL_RESTORE_ASYNC_COMPACT_AFTER_RESCALE;
+import static org.apache.flink.contrib.streaming.state.RocksDBConfigurableOptions.USE_DELETE_FILES_IN_RANGE_DURING_RESCALING;
import static org.apache.flink.contrib.streaming.state.RocksDBConfigurableOptions.USE_INGEST_DB_RESTORE_MODE;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.reset;
@@ -186,14 +190,19 @@ public class EmbeddedRocksDBStateBackendTest
dbPath = TempDirUtils.newFolder(tempFolder).getAbsolutePath();
EmbeddedRocksDBStateBackend backend =
new EmbeddedRocksDBStateBackend(enableIncrementalCheckpointing);
+ Configuration configuration = createBackendConfig();
+ backend = backend.configure(configuration, Thread.currentThread().getContextClassLoader());
+ backend.setDbStoragePath(dbPath);
+ return backend;
+ }
+
+ private Configuration createBackendConfig() {
Configuration configuration = new Configuration();
configuration.set(USE_INGEST_DB_RESTORE_MODE, useIngestDB);
configuration.set(
RocksDBOptions.TIMER_SERVICE_FACTORY,
EmbeddedRocksDBStateBackend.PriorityQueueStateType.ROCKSDB);
- backend = backend.configure(configuration, Thread.currentThread().getContextClassLoader());
- backend.setDbStoragePath(dbPath);
- return backend;
+ return configuration;
}
@Override
@@ -656,6 +665,48 @@ public class EmbeddedRocksDBStateBackendTest
assertThatThrownBy(state::clear).isInstanceOf(FlinkRuntimeException.class);
}
+ /** Test for all configs that use {@link org.apache.flink.util.TernaryBoolean}. */
+ @TestTemplate
+ public void testConfigureTernaryBooleanConfigs() throws Exception {
+ ConfigurableStateBackend stateBackend = getStateBackend();
+ if (!(stateBackend instanceof EmbeddedRocksDBStateBackend)) {
+ return;
+ }
+ EmbeddedRocksDBStateBackend rocksDBStateBackend =
+ (EmbeddedRocksDBStateBackend) stateBackend;
+ Configuration baseConfig = createBackendConfig();
+ Configuration testConfig = new Configuration();
+ testConfig.setBoolean(
+ USE_INGEST_DB_RESTORE_MODE, !USE_INGEST_DB_RESTORE_MODE.defaultValue());
+ testConfig.setBoolean(
+ INCREMENTAL_RESTORE_ASYNC_COMPACT_AFTER_RESCALE,
+ !INCREMENTAL_RESTORE_ASYNC_COMPACT_AFTER_RESCALE.defaultValue());
+ testConfig.setBoolean(
+ USE_DELETE_FILES_IN_RANGE_DURING_RESCALING,
+ !USE_DELETE_FILES_IN_RANGE_DURING_RESCALING.defaultValue());
+ EmbeddedRocksDBStateBackend configuredBackend =
+ rocksDBStateBackend.configure(
+ testConfig, Thread.currentThread().getContextClassLoader());
+
+ checkBooleanWithBaseConf(
+ baseConfig,
+ USE_INGEST_DB_RESTORE_MODE,
+ configuredBackend.getUseIngestDbRestoreMode());
+ checkBooleanWithBaseConf(
+ baseConfig,
+ USE_DELETE_FILES_IN_RANGE_DURING_RESCALING,
+ configuredBackend.isRescalingUseDeleteFilesInRange());
+ checkBooleanWithBaseConf(
+ baseConfig,
+ INCREMENTAL_RESTORE_ASYNC_COMPACT_AFTER_RESCALE,
+ configuredBackend.getIncrementalRestoreAsyncCompactAfterRescale());
+ }
+
+ private void checkBooleanWithBaseConf(
+ Configuration testConfig, ConfigOption<Boolean> option, boolean value) {
+ assertEquals(testConfig.getOptional(option).orElse(!option.defaultValue()), value);
+ }
+
private void runStateUpdates() throws Exception {
for (int i = 50; i < 150; ++i) {
if (i % 10 == 0) {
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackendConfigTest.java b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackendConfigTest.java
index d54210e727e..79869038bd7 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackendConfigTest.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackendConfigTest.java
@@ -890,6 +890,31 @@ public class RocksDBStateBackendConfigTest {
assertTrue(rocksDBStateBackend.getUseIngestDbRestoreMode());
}
+ @Test
+ public void testDefaultUseDeleteFilesInRange() {
+ EmbeddedRocksDBStateBackend rocksDBStateBackend = new EmbeddedRocksDBStateBackend(true);
+ assertEquals(
+ RocksDBConfigurableOptions.USE_DELETE_FILES_IN_RANGE_DURING_RESCALING
+ .defaultValue(),
+ rocksDBStateBackend.isRescalingUseDeleteFilesInRange());
+ }
+
+ @Test
+ public void testConfigureUseFilesInRange() {
+ EmbeddedRocksDBStateBackend rocksDBStateBackend = new EmbeddedRocksDBStateBackend(true);
+ Configuration configuration = new Configuration();
+ configuration.set(
+ RocksDBConfigurableOptions.USE_DELETE_FILES_IN_RANGE_DURING_RESCALING,
+ !RocksDBConfigurableOptions.USE_DELETE_FILES_IN_RANGE_DURING_RESCALING
+ .defaultValue());
+ rocksDBStateBackend =
+ rocksDBStateBackend.configure(configuration, getClass().getClassLoader());
+ assertEquals(
+ !RocksDBConfigurableOptions.USE_DELETE_FILES_IN_RANGE_DURING_RESCALING
+ .defaultValue(),
+ rocksDBStateBackend.isRescalingUseDeleteFilesInRange());
+ }
+
@Test
public void testDefaultIncrementalRestoreInstanceBufferSize() {
EmbeddedRocksDBStateBackend rocksDBStateBackend = new EmbeddedRocksDBStateBackend(true);