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/09/10 07:09:21 UTC
(flink) branch master updated (7adeecd3445 -> 95ef07b789b)
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 7adeecd3445 [FLINK-34510][Runtime/State]Rename RestoreMode to RecoveryClaimMode (#25192)
new c11632674a8 [FLINK-35580] Prevent potential JVM crashes from async compaction when RocksDB is already closed.
new a0ace618ad1 [FLINK-35580] Register async compact task with the backend's lifecycle instead of the restore lifecycle.
new 9b2b922a159 [FLINK-35580] Introduce artificial map key for RegisteredStateMetaInfoBase for safe deduplication in RocksDBIncrementalRestoreOperation.
new 9ba7e670512 [FLINK-35580] Handle async compaction exceptions in RocksDB restore.
new deeb42ec897 [FLINK-35580] Improve logging and problem analysis in RockDBHandle.
new 42757607f23 [FLINK-35580] Fix synchronization issue when closing RocksDBWriteBatchWrapper
new 830a17155cf [FLINK-35580] Ensure RocksDB working directory is created before opening DB.
new 95ef07b789b [FLINK-35580] Improve logging for IngestDB to make key range checks more observable.
The 8 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:
.../apache/flink/core/fs/ICloseableRegistry.java | 15 --
.../state/CompositeKeySerializationUtils.java | 12 ++
.../runtime/state/RegisteredStateMetaInfoBase.java | 46 +++++
.../state/EmbeddedRocksDBStateBackend.java | 4 +-
.../state/RocksDBIncrementalCheckpointUtils.java | 222 ++++++++++++++------
.../streaming/state/RocksDBKeyedStateBackend.java | 8 +-
.../state/RocksDBKeyedStateBackendBuilder.java | 37 +++-
.../streaming/state/RocksDBOperationUtils.java | 26 ++-
.../streaming/state/RocksDBWriteBatchWrapper.java | 58 ++++++
.../state/restore/RocksDBFullRestoreOperation.java | 4 +-
.../streaming/state/restore/RocksDBHandle.java | 11 +-
.../RocksDBHeapTimersFullRestoreOperation.java | 4 +-
.../RocksDBIncrementalRestoreOperation.java | 229 ++++++++++++---------
.../state/restore/RocksDBRestoreResult.java | 11 +-
.../streaming/state/RocksDBRecoveryTest.java | 47 ++++-
.../contrib/streaming/state/RocksDBTestUtils.java | 2 +-
.../state/RocksDBWriteBatchWrapperTest.java | 57 +++++
17 files changed, 576 insertions(+), 217 deletions(-)
(flink) 06/08: [FLINK-35580] Fix synchronization issue when closing RocksDBWriteBatchWrapper
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 42757607f2332c297f1a3e0586942ee0c876f2da
Author: Roman Khachatryan <kh...@gmail.com>
AuthorDate: Tue May 7 17:42:33 2024 +0200
[FLINK-35580] Fix synchronization issue when closing RocksDBWriteBatchWrapper
---
.../apache/flink/core/fs/ICloseableRegistry.java | 15 ------
.../streaming/state/RocksDBKeyedStateBackend.java | 6 ++-
.../streaming/state/RocksDBWriteBatchWrapper.java | 58 ++++++++++++++++++++++
.../state/restore/RocksDBFullRestoreOperation.java | 4 +-
.../RocksDBHeapTimersFullRestoreOperation.java | 4 +-
.../RocksDBIncrementalRestoreOperation.java | 3 +-
.../state/RocksDBWriteBatchWrapperTest.java | 57 +++++++++++++++++++++
7 files changed, 123 insertions(+), 24 deletions(-)
diff --git a/flink-core/src/main/java/org/apache/flink/core/fs/ICloseableRegistry.java b/flink-core/src/main/java/org/apache/flink/core/fs/ICloseableRegistry.java
index ebdb0f13692..90fd201f341 100644
--- a/flink-core/src/main/java/org/apache/flink/core/fs/ICloseableRegistry.java
+++ b/flink-core/src/main/java/org/apache/flink/core/fs/ICloseableRegistry.java
@@ -19,11 +19,9 @@
package org.apache.flink.core.fs;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.util.ExceptionUtils;
import java.io.Closeable;
import java.io.IOException;
-import java.util.concurrent.atomic.AtomicBoolean;
/**
* This class allows to register instances of {@link Closeable}, which are all closed if this
@@ -39,19 +37,6 @@ import java.util.concurrent.atomic.AtomicBoolean;
@Internal
public interface ICloseableRegistry extends Closeable {
- static Closeable asCloseable(AutoCloseable autoCloseable) {
- AtomicBoolean closed = new AtomicBoolean(false);
- return () -> {
- if (closed.compareAndSet(false, true)) {
- try {
- autoCloseable.close();
- } catch (Exception e) {
- ExceptionUtils.rethrowIOException(e);
- }
- }
- };
- }
-
/**
* Registers a {@link Closeable} with the registry. In case the registry is already closed, this
* method throws an {@link IllegalStateException} and closes the passed {@link Closeable}.
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
index ba4c6570acd..9f1857a6ccf 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
@@ -84,6 +84,7 @@ import javax.annotation.Nonnegative;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
+import java.io.Closeable;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
@@ -840,7 +841,10 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
try (RocksIteratorWrapper iterator =
RocksDBOperationUtils.getRocksIterator(db, stateMetaInfo.f0, readOptions);
RocksDBWriteBatchWrapper batchWriter =
- new RocksDBWriteBatchWrapper(db, getWriteOptions(), getWriteBatchSize())) {
+ new RocksDBWriteBatchWrapper(db, getWriteOptions(), getWriteBatchSize());
+ Closeable ignored =
+ cancelStreamRegistry.registerCloseableTemporarily(
+ writeBatchWrapper.getCancelCloseable())) {
iterator.seekToFirst();
DataInputDeserializer serializedValueInput = new DataInputDeserializer();
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBWriteBatchWrapper.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBWriteBatchWrapper.java
index 354009e335c..c2da5517566 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBWriteBatchWrapper.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBWriteBatchWrapper.java
@@ -19,6 +19,7 @@
package org.apache.flink.contrib.streaming.state;
import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.runtime.execution.CancelTaskException;
import org.apache.flink.util.IOUtils;
import org.apache.flink.util.Preconditions;
@@ -32,6 +33,7 @@ import javax.annotation.Nonnegative;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
+import java.io.Closeable;
import java.util.ArrayList;
import java.util.List;
@@ -47,6 +49,9 @@ public class RocksDBWriteBatchWrapper implements AutoCloseable {
private static final int PER_RECORD_BYTES = 100;
// default 0 for disable memory size based flush
private static final long DEFAULT_BATCH_SIZE = 0;
+ private static final int DEFAULT_CANCELLATION_CHECK_INTERVAL = MIN_CAPACITY;
+ private static final int DEFAULT_CANCELLATION_CHECK_INTERVAL_BYTES =
+ DEFAULT_CANCELLATION_CHECK_INTERVAL * PER_RECORD_BYTES;
private final RocksDB db;
@@ -61,6 +66,14 @@ public class RocksDBWriteBatchWrapper implements AutoCloseable {
/** List of all objects that we need to close in close(). */
private final List<AutoCloseable> toClose;
+ private volatile boolean cancelled;
+
+ private final int cancellationCheckInterval;
+
+ private final long cancellationCheckIntervalBytes;
+
+ private long lastCancellationCheckBatchSize = 0L;
+
public RocksDBWriteBatchWrapper(@Nonnull RocksDB rocksDB, long writeBatchSize) {
this(rocksDB, null, 500, writeBatchSize);
}
@@ -79,6 +92,22 @@ public class RocksDBWriteBatchWrapper implements AutoCloseable {
@Nullable WriteOptions options,
int capacity,
long batchSize) {
+ this(
+ rocksDB,
+ options,
+ capacity,
+ batchSize,
+ DEFAULT_CANCELLATION_CHECK_INTERVAL,
+ DEFAULT_CANCELLATION_CHECK_INTERVAL_BYTES);
+ }
+
+ public RocksDBWriteBatchWrapper(
+ @Nonnull RocksDB rocksDB,
+ @Nullable WriteOptions options,
+ int capacity,
+ long batchSize,
+ int cancellationCheckInterval,
+ long cancellationCheckIntervalBytes) {
Preconditions.checkArgument(
capacity >= MIN_CAPACITY && capacity <= MAX_CAPACITY,
"capacity should be between " + MIN_CAPACITY + " and " + MAX_CAPACITY);
@@ -104,16 +133,27 @@ public class RocksDBWriteBatchWrapper implements AutoCloseable {
// We own this object, so we must ensure that we close it.
this.toClose.add(this.options);
}
+ this.cancellationCheckInterval = cancellationCheckInterval;
+ this.cancellationCheckIntervalBytes = cancellationCheckIntervalBytes;
}
public void put(@Nonnull ColumnFamilyHandle handle, @Nonnull byte[] key, @Nonnull byte[] value)
throws RocksDBException {
+ maybeEnsureNotCancelled();
batch.put(handle, key, value);
flushIfNeeded();
}
+ private void maybeEnsureNotCancelled() {
+ if (batch.count() % cancellationCheckInterval == 0
+ || batch.getDataSize() - lastCancellationCheckBatchSize
+ >= cancellationCheckIntervalBytes) {
+ ensureNotCancelled();
+ }
+ }
+
public void remove(@Nonnull ColumnFamilyHandle handle, @Nonnull byte[] key)
throws RocksDBException {
@@ -123,8 +163,10 @@ public class RocksDBWriteBatchWrapper implements AutoCloseable {
}
public void flush() throws RocksDBException {
+ ensureNotCancelled();
db.write(options, batch);
batch.clear();
+ lastCancellationCheckBatchSize = 0;
}
@VisibleForTesting
@@ -132,9 +174,18 @@ public class RocksDBWriteBatchWrapper implements AutoCloseable {
return options;
}
+ public void markCancelled() {
+ this.cancelled = true;
+ }
+
+ public Closeable getCancelCloseable() {
+ return this::markCancelled;
+ }
+
@Override
public void close() throws RocksDBException {
try {
+ ensureNotCancelled();
if (batch.count() != 0) {
flush();
}
@@ -143,6 +194,13 @@ public class RocksDBWriteBatchWrapper implements AutoCloseable {
}
}
+ private void ensureNotCancelled() {
+ if (cancelled) {
+ throw new CancelTaskException();
+ }
+ lastCancellationCheckBatchSize = batch.getDataSize();
+ }
+
private void flushIfNeeded() throws RocksDBException {
boolean needFlush =
batch.count() == capacity || (batchSize > 0 && getDataSize() >= batchSize);
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBFullRestoreOperation.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBFullRestoreOperation.java
index fdaf21fb180..052404b4957 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBFullRestoreOperation.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBFullRestoreOperation.java
@@ -52,8 +52,6 @@ import java.util.List;
import java.util.Map;
import java.util.function.Function;
-import static org.apache.flink.core.fs.ICloseableRegistry.asCloseable;
-
/** Encapsulates the process of restoring a RocksDB instance from a full snapshot. */
public class RocksDBFullRestoreOperation<K> implements RocksDBRestoreOperation {
private final FullSnapshotRestoreOperation<K> savepointRestoreOperation;
@@ -149,7 +147,7 @@ public class RocksDBFullRestoreOperation<K> implements RocksDBRestoreOperation {
new RocksDBWriteBatchWrapper(this.rocksHandle.getDb(), writeBatchSize);
Closeable ignored =
cancelStreamRegistryForRestore.registerCloseableTemporarily(
- asCloseable(writeBatchWrapper))) {
+ writeBatchWrapper.getCancelCloseable())) {
ColumnFamilyHandle handle = null;
while (keyGroups.hasNext()) {
KeyGroup keyGroup = keyGroups.next();
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBHeapTimersFullRestoreOperation.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBHeapTimersFullRestoreOperation.java
index c4eed5ff3d4..da2facae299 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBHeapTimersFullRestoreOperation.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBHeapTimersFullRestoreOperation.java
@@ -64,8 +64,6 @@ import java.util.List;
import java.util.Map;
import java.util.function.Function;
-import static org.apache.flink.core.fs.ICloseableRegistry.asCloseable;
-
/** Encapsulates the process of restoring a RocksDB instance from a full snapshot. */
public class RocksDBHeapTimersFullRestoreOperation<K> implements RocksDBRestoreOperation {
private final FullSnapshotRestoreOperation<K> savepointRestoreOperation;
@@ -194,7 +192,7 @@ public class RocksDBHeapTimersFullRestoreOperation<K> implements RocksDBRestoreO
new RocksDBWriteBatchWrapper(this.rocksHandle.getDb(), writeBatchSize);
Closeable ignored =
cancelStreamRegistryForRestore.registerCloseableTemporarily(
- asCloseable(writeBatchWrapper))) {
+ writeBatchWrapper.getCancelCloseable())) {
HeapPriorityQueueSnapshotRestoreWrapper<HeapPriorityQueueElement> restoredPQ = null;
ColumnFamilyHandle handle = null;
while (keyGroups.hasNext()) {
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java
index 6d7ed6e374d..89ec9d1bb35 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java
@@ -90,7 +90,6 @@ import java.util.concurrent.ExecutorService;
import java.util.function.Function;
import java.util.stream.Collectors;
-import static org.apache.flink.core.fs.ICloseableRegistry.asCloseable;
import static org.apache.flink.runtime.metrics.MetricNames.DOWNLOAD_STATE_DURATION;
import static org.apache.flink.runtime.metrics.MetricNames.RESTORE_ASYNC_COMPACTION_DURATION;
import static org.apache.flink.runtime.metrics.MetricNames.RESTORE_STATE_DURATION;
@@ -798,7 +797,7 @@ public class RocksDBIncrementalRestoreOperation<K> implements RocksDBRestoreOper
new RocksDBWriteBatchWrapper(this.rocksHandle.getDb(), writeBatchSize);
Closeable ignored =
cancelStreamRegistryForRestore.registerCloseableTemporarily(
- asCloseable(writeBatchWrapper))) {
+ writeBatchWrapper.getCancelCloseable())) {
for (IncrementalLocalKeyedStateHandle handleToCopy : toImport) {
try (RestoredDBInstance restoredDBInstance =
restoreTempDBInstanceFromLocalState(handleToCopy)) {
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBWriteBatchWrapperTest.java b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBWriteBatchWrapperTest.java
index 6a7c95af067..05465ca5a3b 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBWriteBatchWrapperTest.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBWriteBatchWrapperTest.java
@@ -19,6 +19,8 @@
package org.apache.flink.contrib.streaming.state;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.core.fs.CloseableRegistry;
+import org.apache.flink.runtime.execution.CancelTaskException;
import org.junit.Assert;
import org.junit.Rule;
@@ -29,11 +31,14 @@ import org.rocksdb.ColumnFamilyHandle;
import org.rocksdb.RocksDB;
import org.rocksdb.WriteOptions;
+import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadLocalRandom;
import static org.apache.flink.contrib.streaming.state.RocksDBConfigurableOptions.WRITE_BATCH_SIZE;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
@@ -43,6 +48,58 @@ public class RocksDBWriteBatchWrapperTest {
@Rule public TemporaryFolder folder = new TemporaryFolder();
+ @Test(expected = CancelTaskException.class)
+ public void testAsyncCancellation() throws Exception {
+ final CompletableFuture<Void> writeStartedFuture = new CompletableFuture<>();
+ final CompletableFuture<Void> cancellationRequestedFuture = new CompletableFuture<>();
+ final CloseableRegistry registry = new CloseableRegistry();
+ new Thread(
+ () -> {
+ writeStartedFuture.join();
+ try {
+ registry.close();
+ cancellationRequestedFuture.complete(null);
+ } catch (IOException e) {
+ cancellationRequestedFuture.completeExceptionally(e);
+ }
+ })
+ .start();
+
+ final int capacity = 1000; // max
+ final int cancellationCheckInterval = 1;
+ long batchSizeBytes = WRITE_BATCH_SIZE.defaultValue().getBytes();
+
+ try (RocksDB db = RocksDB.open(folder.newFolder().getAbsolutePath());
+ WriteOptions options = new WriteOptions().setDisableWAL(true);
+ ColumnFamilyHandle handle =
+ db.createColumnFamily(new ColumnFamilyDescriptor("test".getBytes()));
+ RocksDBWriteBatchWrapper writeBatchWrapper =
+ new RocksDBWriteBatchWrapper(
+ db,
+ options,
+ capacity,
+ batchSizeBytes,
+ cancellationCheckInterval,
+ batchSizeBytes)) {
+ registry.registerCloseable(writeBatchWrapper.getCancelCloseable());
+ writeStartedFuture.complete(null);
+
+ //noinspection InfiniteLoopStatement
+ for (int i = 0; ; i++) {
+ try {
+ writeBatchWrapper.put(
+ handle, ("key:" + i).getBytes(), ("value:" + i).getBytes());
+ } catch (Exception e) {
+ cancellationRequestedFuture.join(); // shouldn't have any errors
+ throw e;
+ }
+ // make sure that cancellation is triggered earlier than periodic flush
+ // but allow some delay of cancellation propagation
+ assertThat(i).isLessThan(cancellationCheckInterval * 2);
+ }
+ }
+ }
+
@Test
public void basicTest() throws Exception {
(flink) 02/08: [FLINK-35580] Register async compact task with the backend's lifecycle instead of the restore lifecycle.
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 a0ace618ad1df15c5d906d7d0ecc0f9d7f3ea685
Author: Stefan Richter <sr...@confluent.io>
AuthorDate: Thu Mar 7 17:02:13 2024 +0100
[FLINK-35580] Register async compact task with the backend's lifecycle instead of the restore lifecycle.
---
.../state/RocksDBKeyedStateBackendBuilder.java | 13 ++++++----
.../RocksDBIncrementalRestoreOperation.java | 30 +++++++++++++++++-----
2 files changed, 31 insertions(+), 12 deletions(-)
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackendBuilder.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackendBuilder.java
index 88b44462146..7b312d759ad 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackendBuilder.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackendBuilder.java
@@ -330,7 +330,7 @@ public class RocksDBKeyedStateBackendBuilder<K> extends AbstractKeyedStateBacken
RocksDBWriteBatchWrapper writeBatchWrapper = null;
ColumnFamilyHandle defaultColumnFamilyHandle = null;
RocksDBNativeMetricMonitor nativeMetricMonitor = null;
- CloseableRegistry cancelStreamRegistryForBackend = new CloseableRegistry();
+ CloseableRegistry cancelRegistryForBackend = new CloseableRegistry();
LinkedHashMap<String, RocksDBKeyedStateBackend.RocksDbKvStateInfo> kvStateInformation =
new LinkedHashMap<>();
LinkedHashMap<String, HeapPriorityQueueSnapshotRestoreWrapper<?>> registeredPQStates =
@@ -375,6 +375,7 @@ public class RocksDBKeyedStateBackendBuilder<K> extends AbstractKeyedStateBacken
keyGroupPrefixBytes,
rocksDBResourceGuard,
cancelStreamRegistry,
+ cancelRegistryForBackend,
kvStateInformation,
registeredPQStates,
ttlCompactFiltersManager);
@@ -435,7 +436,7 @@ public class RocksDBKeyedStateBackendBuilder<K> extends AbstractKeyedStateBacken
// Do clean up
List<ColumnFamilyOptions> columnFamilyOptions =
new ArrayList<>(kvStateInformation.values().size());
- IOUtils.closeQuietly(cancelStreamRegistryForBackend);
+ IOUtils.closeQuietly(cancelRegistryForBackend);
IOUtils.closeQuietly(writeBatchWrapper);
IOUtils.closeQuietly(rocksDBResourceGuard);
RocksDBOperationUtils.addColumnFamilyOptionsToCloseLater(
@@ -487,7 +488,7 @@ public class RocksDBKeyedStateBackendBuilder<K> extends AbstractKeyedStateBacken
kvStateInformation,
registeredPQStates,
keyGroupPrefixBytes,
- cancelStreamRegistryForBackend,
+ cancelRegistryForBackend,
this.keyGroupCompressionDecorator,
rocksDBResourceGuard,
checkpointStrategy,
@@ -506,7 +507,8 @@ public class RocksDBKeyedStateBackendBuilder<K> extends AbstractKeyedStateBacken
private RocksDBRestoreOperation getRocksDBRestoreOperation(
int keyGroupPrefixBytes,
ResourceGuard rocksDBResourceGuard,
- CloseableRegistry cancelStreamRegistry,
+ CloseableRegistry cancelStreamRegistryForRestore,
+ CloseableRegistry cancelRegistryForBackend,
LinkedHashMap<String, RocksDBKeyedStateBackend.RocksDbKvStateInfo> kvStateInformation,
LinkedHashMap<String, HeapPriorityQueueSnapshotRestoreWrapper<?>> registeredPQStates,
RocksDbTtlCompactFiltersManager ttlCompactFiltersManager) {
@@ -530,7 +532,8 @@ public class RocksDBKeyedStateBackendBuilder<K> extends AbstractKeyedStateBacken
keyGroupPrefixBytes,
numberOfTransferingThreads,
rocksDBResourceGuard,
- cancelStreamRegistry,
+ cancelStreamRegistryForRestore,
+ cancelRegistryForBackend,
userCodeClassLoader,
kvStateInformation,
keySerializerProvider,
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java
index 9aae10f2b5e..5c72e78766b 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java
@@ -113,7 +113,19 @@ public class RocksDBIncrementalRestoreOperation<K> implements RocksDBRestoreOper
private final SortedMap<Long, Collection<HandleAndLocalPath>> restoredSstFiles;
private final RocksDBHandle rocksHandle;
private final Collection<IncrementalKeyedStateHandle> restoreStateHandles;
- private final CloseableRegistry cancelStreamRegistry;
+
+ /**
+ * This registry will be closed after restore and should only contain Closeables that are closed
+ * by the end of the restore operation.
+ */
+ private final CloseableRegistry cancelStreamRegistryForRestore;
+
+ /**
+ * This registry will only be closed when the created backend is closed and should be used for
+ * all Closeables that are closed at some later point after restore.
+ */
+ private final CloseableRegistry cancelRegistryForBackend;
+
private final KeyGroupRange keyGroupRange;
private final File instanceBasePath;
private final int numberOfTransferringThreads;
@@ -144,7 +156,8 @@ public class RocksDBIncrementalRestoreOperation<K> implements RocksDBRestoreOper
int keyGroupPrefixBytes,
int numberOfTransferringThreads,
ResourceGuard dbResourceGuard,
- CloseableRegistry cancelStreamRegistry,
+ CloseableRegistry cancelStreamRegistryForRestore,
+ CloseableRegistry cancelRegistryForBackend,
ClassLoader userCodeClassLoader,
Map<String, RocksDbKvStateInfo> kvStateInformation,
StateSerializerProvider<K> keySerializerProvider,
@@ -183,7 +196,8 @@ public class RocksDBIncrementalRestoreOperation<K> implements RocksDBRestoreOper
this.customInitializationMetrics = customInitializationMetrics;
this.restoreStateHandles = restoreStateHandles;
this.dbResourceGuard = dbResourceGuard;
- this.cancelStreamRegistry = cancelStreamRegistry;
+ this.cancelStreamRegistryForRestore = cancelStreamRegistryForRestore;
+ this.cancelRegistryForBackend = cancelRegistryForBackend;
this.keyGroupRange = keyGroupRange;
this.instanceBasePath = instanceBasePath;
this.numberOfTransferringThreads = numberOfTransferringThreads;
@@ -279,7 +293,9 @@ public class RocksDBIncrementalRestoreOperation<K> implements RocksDBRestoreOper
keyGroupPrefixBytes,
keyGroupRange,
dbResourceGuard,
- cancelStreamRegistry);
+ // This task will be owned by the backend's lifecycle because it
+ // continues to exist after restore is completed.
+ cancelRegistryForBackend);
runAndReportDuration(asyncRangeCompactionTask, RESTORE_ASYNC_COMPACTION_DURATION);
logger.info(
"Completed async compaction after restore for backend {} in operator {} after {} ms.",
@@ -742,7 +758,7 @@ public class RocksDBIncrementalRestoreOperation<K> implements RocksDBRestoreOper
RocksDBStateDataTransferHelper.forThreadNumIfSpecified(
numberOfTransferringThreads, ioExecutor))) {
rocksDBStateDownloader.transferAllStateDataToDirectory(
- downloadSpecs, cancelStreamRegistry);
+ downloadSpecs, cancelStreamRegistryForRestore);
logger.info(
"Finished downloading remote state to local directory in operator {} for target key-group range {}.",
operatorIdentifier,
@@ -999,11 +1015,11 @@ public class RocksDBIncrementalRestoreOperation<K> implements RocksDBRestoreOper
try {
inputStream = metaStateHandle.openInputStream();
- cancelStreamRegistry.registerCloseable(inputStream);
+ cancelStreamRegistryForRestore.registerCloseable(inputStream);
DataInputView in = new DataInputViewStreamWrapper(inputStream);
return readMetaData(in);
} finally {
- if (cancelStreamRegistry.unregisterCloseable(inputStream)) {
+ if (cancelStreamRegistryForRestore.unregisterCloseable(inputStream)) {
inputStream.close();
}
}
(flink) 08/08: [FLINK-35580] Improve logging for IngestDB to make key range checks more observable.
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 95ef07b789bf9a57c85bd0a2647da8d1de34f462
Author: Stefan Richter <sr...@confluent.io>
AuthorDate: Wed Apr 17 16:54:41 2024 +0200
[FLINK-35580] Improve logging for IngestDB to make key range checks more observable.
---
.../state/CompositeKeySerializationUtils.java | 12 +++
.../state/RocksDBIncrementalCheckpointUtils.java | 119 +++++++++++++++------
.../RocksDBIncrementalRestoreOperation.java | 29 +++--
3 files changed, 118 insertions(+), 42 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/CompositeKeySerializationUtils.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/CompositeKeySerializationUtils.java
index 8088c19ec68..eddb5345df5 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/CompositeKeySerializationUtils.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/CompositeKeySerializationUtils.java
@@ -40,6 +40,18 @@ public class CompositeKeySerializationUtils {
return keyGroup;
}
+ public static int extractKeyGroup(int keyGroupPrefixBytes, byte[] rocksDBKey) {
+ if (rocksDBKey.length < keyGroupPrefixBytes) {
+ return -1;
+ }
+ int keyGroup = 0;
+ for (int i = 0; i < keyGroupPrefixBytes; ++i) {
+ keyGroup <<= 8;
+ keyGroup |= (rocksDBKey[i] & 0xFF);
+ }
+ return keyGroup;
+ }
+
public static <K> K readKey(
TypeSerializer<K> keySerializer,
DataInputDeserializer inputView,
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 e9ab287cf74..0d952eede34 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
@@ -221,21 +221,6 @@ public class RocksDBIncrementalCheckpointUtils {
}
}
- /**
- * Returns true, if all entries in the sst files of the given DB is strictly within the expected
- * key-group range for the DB.
- *
- * @param db the DB to check.
- * @param dbExpectedKeyGroupRange the expected key-groups range of the DB.
- * @param keyGroupPrefixBytes the number of bytes used to serialize the key-group prefix of keys
- * in the DB.
- */
- public static boolean isSstDataInKeyGroupRange(
- RocksDB db, int keyGroupPrefixBytes, KeyGroupRange dbExpectedKeyGroupRange) {
- return checkSstDataAgainstKeyGroupRange(db, keyGroupPrefixBytes, dbExpectedKeyGroupRange)
- .allInRange();
- }
-
/**
* Returns a range compaction task as runnable if any data in the SST files of the given DB
* exceeds the proclaimed key-group range.
@@ -309,7 +294,7 @@ public class RocksDBIncrementalCheckpointUtils {
columnFamilyHandle,
// TODO: change to null once this API is fixed
new byte[] {},
- rangeCheckResult.minKey,
+ rangeCheckResult.getProclaimedMinKey(),
compactionOptions);
}
}
@@ -325,7 +310,7 @@ public class RocksDBIncrementalCheckpointUtils {
rocksDBResourceGuard.acquireResource()) {
db.compactRange(
columnFamilyHandle,
- rangeCheckResult.maxKey,
+ rangeCheckResult.getProclaimedMaxKey(),
// TODO: change to null once this API is fixed
new byte[] {
(byte) 0xff, (byte) 0xff, (byte) 0xff, (byte) 0xff
@@ -351,7 +336,7 @@ public class RocksDBIncrementalCheckpointUtils {
* @param dbExpectedKeyGroupRange the expected key-groups range of the DB.
* @return the check result with detailed info about lower and upper bound violations.
*/
- private static RangeCheckResult checkSstDataAgainstKeyGroupRange(
+ public static RangeCheckResult checkSstDataAgainstKeyGroupRange(
RocksDB db, int keyGroupPrefixBytes, KeyGroupRange dbExpectedKeyGroupRange) {
final byte[] beginKeyGroupBytes = new byte[keyGroupPrefixBytes];
final byte[] endKeyGroupBytes = new byte[keyGroupPrefixBytes];
@@ -363,12 +348,12 @@ public class RocksDBIncrementalCheckpointUtils {
dbExpectedKeyGroupRange.getEndKeyGroup() + 1, endKeyGroupBytes);
KeyRange dbKeyRange = getDBKeyRange(db);
- Comparator<byte[]> comparator = UnsignedBytes.lexicographicalComparator();
return RangeCheckResult.of(
- comparator.compare(dbKeyRange.minKey, beginKeyGroupBytes) >= 0,
- comparator.compare(dbKeyRange.maxKey, endKeyGroupBytes) < 0,
beginKeyGroupBytes,
- endKeyGroupBytes);
+ endKeyGroupBytes,
+ dbKeyRange.minKey,
+ dbKeyRange.maxKey,
+ keyGroupPrefixBytes);
}
/** Returns a pair of minimum and maximum key in the sst files of the given database. */
@@ -543,28 +528,94 @@ public class RocksDBIncrementalCheckpointUtils {
* instance contains any keys (or tombstones for keys) that don't belong in the instance's
* key-groups range.
*/
- private static final class RangeCheckResult {
- private final byte[] minKey;
-
- private final byte[] maxKey;
+ public static final class RangeCheckResult {
+ private final byte[] proclaimedMinKey;
+ private final byte[] proclaimedMaxKey;
+ private final byte[] actualMinKey;
+ private final byte[] actualMaxKey;
final boolean leftInRange;
final boolean rightInRange;
+ final int keyGroupPrefixBytes;
+
private RangeCheckResult(
- boolean leftInRange, boolean rightInRange, byte[] minKey, byte[] maxKey) {
- this.leftInRange = leftInRange;
- this.rightInRange = rightInRange;
- this.minKey = minKey;
- this.maxKey = maxKey;
+ byte[] proclaimedMinKey,
+ byte[] proclaimedMaxKey,
+ byte[] actualMinKey,
+ byte[] actualMaxKey,
+ int keyGroupPrefixBytes) {
+ Comparator<byte[]> comparator = UnsignedBytes.lexicographicalComparator();
+ this.proclaimedMinKey = proclaimedMinKey;
+ this.proclaimedMaxKey = proclaimedMaxKey;
+ this.actualMinKey = actualMinKey;
+ this.actualMaxKey = actualMaxKey;
+ this.leftInRange = comparator.compare(actualMinKey, proclaimedMinKey) >= 0;
+ // TODO: consider using <= here to avoid that range delete tombstones of
+ // (targetMaxKeyGroup + 1) prevent using ingest for no good reason.
+ this.rightInRange = comparator.compare(actualMaxKey, proclaimedMaxKey) < 0;
+ this.keyGroupPrefixBytes = keyGroupPrefixBytes;
}
- boolean allInRange() {
+ public boolean allInRange() {
return leftInRange && rightInRange;
}
+ public byte[] getProclaimedMinKey() {
+ return proclaimedMinKey;
+ }
+
+ public byte[] getProclaimedMaxKey() {
+ return proclaimedMaxKey;
+ }
+
+ public byte[] getActualMinKey() {
+ return actualMinKey;
+ }
+
+ public byte[] getActualMaxKey() {
+ return actualMaxKey;
+ }
+
+ public int getKeyGroupPrefixBytes() {
+ return keyGroupPrefixBytes;
+ }
+
+ public boolean isLeftInRange() {
+ return leftInRange;
+ }
+
+ public boolean isRightInRange() {
+ return rightInRange;
+ }
+
static RangeCheckResult of(
- boolean leftInRange, boolean rightInRange, byte[] minKey, byte[] maxKey) {
- return new RangeCheckResult(leftInRange, rightInRange, minKey, maxKey);
+ byte[] proclaimedMinKey,
+ byte[] proclaimedMaxKey,
+ byte[] actualMinKey,
+ byte[] actualMaxKey,
+ int keyGroupPrefixBytes) {
+ return new RangeCheckResult(
+ proclaimedMinKey,
+ proclaimedMaxKey,
+ actualMinKey,
+ actualMaxKey,
+ keyGroupPrefixBytes);
+ }
+
+ @Override
+ public String toString() {
+ return "RangeCheckResult{"
+ + "leftInRange="
+ + leftInRange
+ + ", rightInRange="
+ + rightInRange
+ + ", actualMinKeyGroup="
+ + CompositeKeySerializationUtils.extractKeyGroup(
+ keyGroupPrefixBytes, getActualMinKey())
+ + ", actualMaxKeyGroup="
+ + CompositeKeySerializationUtils.extractKeyGroup(
+ keyGroupPrefixBytes, getActualMaxKey())
+ + '}';
}
}
}
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java
index 89ec9d1bb35..22256b2aa72 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java
@@ -566,8 +566,15 @@ public class RocksDBIncrementalRestoreOperation<K> implements RocksDBRestoreOper
// Check if the data in all SST files referenced in the handle is within the
// proclaimed key-groups range of the handle.
- if (RocksDBIncrementalCheckpointUtils.isSstDataInKeyGroupRange(
- tmpRestoreDBInfo.db, keyGroupPrefixBytes, stateHandle.getKeyGroupRange())) {
+ RocksDBIncrementalCheckpointUtils.RangeCheckResult rangeCheckResult =
+ RocksDBIncrementalCheckpointUtils.checkSstDataAgainstKeyGroupRange(
+ tmpRestoreDBInfo.db,
+ keyGroupPrefixBytes,
+ stateHandle.getKeyGroupRange());
+
+ logger.info("{}" + logLineSuffix, rangeCheckResult);
+
+ if (rangeCheckResult.allInRange()) {
logger.debug("Start exporting" + logLineSuffix);
@@ -606,16 +613,21 @@ public class RocksDBIncrementalRestoreOperation<K> implements RocksDBRestoreOper
++index;
}
+ KeyGroupRange exportedKeyGroupsRange =
+ minExportKeyGroup <= maxExportKeyGroup
+ ? new KeyGroupRange(minExportKeyGroup, maxExportKeyGroup)
+ : KeyGroupRange.EMPTY_KEY_GROUP_RANGE;
+
logger.info(
- "Completed restore export for backend with range {} in operator {}. Number of Exported handles: {}. Skipped handles: {}.",
+ "Completed restore export for backend with range {} in operator {}. {} exported handles with overall exported range {}. {} Skipped handles: {}.",
keyGroupRange.prettyPrintInterval(),
operatorIdentifier,
localKeyedStateHandles.size() - skipped.size(),
+ exportedKeyGroupsRange.prettyPrintInterval(),
+ skipped.size(),
skipped);
- return minExportKeyGroup <= maxExportKeyGroup
- ? new KeyGroupRange(minExportKeyGroup, maxExportKeyGroup)
- : KeyGroupRange.EMPTY_KEY_GROUP_RANGE;
+ return exportedKeyGroupsRange;
}
/**
@@ -674,9 +686,10 @@ public class RocksDBIncrementalRestoreOperation<K> implements RocksDBRestoreOper
// We initialize the base DB by importing all the exported data.
logger.info(
- "Starting to import exported state handles for backend with range {} in operator {} using Clip/Ingest DB.",
+ "Starting to import exported state handles for backend with range {} in operator {} using Clip/Ingest DB with exported range {}.",
keyGroupRange.prettyPrintInterval(),
- operatorIdentifier);
+ operatorIdentifier,
+ exportKeyGroupRange.prettyPrintInterval());
rocksHandle.openDB();
for (Map.Entry<RegisteredStateMetaInfoBase.Key, List<ExportImportFilesMetaData>> entry :
exportedColumnFamilyMetaData.entrySet()) {
(flink) 01/08: [FLINK-35580] Prevent potential JVM crashes from async compaction when RocksDB is already closed.
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 c11632674a8a9636a5f73d906141af578e153d4f
Author: Stefan Richter <sr...@confluent.io>
AuthorDate: Thu Feb 22 12:18:02 2024 +0100
[FLINK-35580] Prevent potential JVM crashes from async compaction when RocksDB is already closed.
---
.../state/RocksDBIncrementalCheckpointUtils.java | 121 +++++++++++------
.../streaming/state/RocksDBKeyedStateBackend.java | 2 +-
.../state/RocksDBKeyedStateBackendBuilder.java | 13 +-
.../streaming/state/RocksDBOperationUtils.java | 13 +-
.../RocksDBIncrementalRestoreOperation.java | 147 +++++++++++----------
.../state/restore/RocksDBRestoreResult.java | 11 +-
.../streaming/state/RocksDBRecoveryTest.java | 15 ++-
7 files changed, 194 insertions(+), 128 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 3147f01747a..e3be413307d 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
@@ -17,12 +17,14 @@
package org.apache.flink.contrib.streaming.state;
+import org.apache.flink.core.fs.CloseableRegistry;
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.ResourceGuard;
import org.apache.flink.util.function.RunnableWithException;
import org.apache.flink.shaded.guava32.com.google.common.primitives.UnsignedBytes;
@@ -41,6 +43,7 @@ import javax.annotation.Nonnegative;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
+import java.io.Closeable;
import java.io.File;
import java.nio.file.Path;
import java.util.ArrayList;
@@ -50,7 +53,6 @@ import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
-import java.util.Optional;
import java.util.UUID;
import java.util.stream.Collectors;
@@ -243,60 +245,93 @@ public class RocksDBIncrementalCheckpointUtils {
* @param keyGroupPrefixBytes the number of bytes used to serialize the key-group prefix of keys
* in the DB.
* @param dbExpectedKeyGroupRange the expected key-groups range of the DB.
+ * @param rocksDBResourceGuard the resource guard for the given db instance.
* @return runnable that performs compaction upon execution if the key-groups range is exceeded.
* Otherwise, empty optional is returned.
*/
- public static Optional<RunnableWithException> createRangeCompactionTaskIfNeeded(
+ public static RunnableWithException createAsyncRangeCompactionTask(
RocksDB db,
Collection<ColumnFamilyHandle> columnFamilyHandles,
int keyGroupPrefixBytes,
- KeyGroupRange dbExpectedKeyGroupRange) {
-
- RangeCheckResult rangeCheckResult =
- checkSstDataAgainstKeyGroupRange(db, keyGroupPrefixBytes, dbExpectedKeyGroupRange);
+ KeyGroupRange dbExpectedKeyGroupRange,
+ ResourceGuard rocksDBResourceGuard,
+ CloseableRegistry closeableRegistry) {
+
+ return () -> {
+ logger.debug(
+ "Starting range check for async compaction targeting key-groups range {}.",
+ dbExpectedKeyGroupRange.prettyPrintInterval());
+ final RangeCheckResult rangeCheckResult;
+ try (ResourceGuard.Lease ignored = rocksDBResourceGuard.acquireResource()) {
+ rangeCheckResult =
+ checkSstDataAgainstKeyGroupRange(
+ db, keyGroupPrefixBytes, dbExpectedKeyGroupRange);
+ }
- if (rangeCheckResult.allInRange()) {
- // No keys exceed the proclaimed range of the backend, so we don't need a compaction
- // from this point of view.
- return Optional.empty();
- }
+ if (rangeCheckResult.allInRange()) {
+ logger.debug(
+ "Nothing to compact in async compaction targeting key-groups range {}.",
+ dbExpectedKeyGroupRange.prettyPrintInterval());
+ // No keys exceed the proclaimed range of the backend, so we don't need a compaction
+ // from this point of view.
+ return;
+ }
- return Optional.of(
- () -> {
- try (CompactRangeOptions compactionOptions =
- new CompactRangeOptions()
- .setExclusiveManualCompaction(true)
- .setBottommostLevelCompaction(
- CompactRangeOptions.BottommostLevelCompaction
- .kForceOptimized)) {
-
- if (!rangeCheckResult.leftInRange) {
- // Compact all keys before from the expected key-groups range
- for (ColumnFamilyHandle columnFamilyHandle : columnFamilyHandles) {
- db.compactRange(
- columnFamilyHandle,
- // TODO: change to null once this API is fixed
- new byte[] {},
- rangeCheckResult.minKey,
- compactionOptions);
- }
+ try (CompactRangeOptions compactionOptions =
+ new CompactRangeOptions()
+ .setBottommostLevelCompaction(
+ CompactRangeOptions.BottommostLevelCompaction
+ .kForceOptimized)) {
+
+ // To cancel an ongoing compaction asap, we register cancelling through the options
+ // with the registry
+ final Closeable cancelCompactionCloseable =
+ () -> {
+ logger.debug(
+ "Cancel request for async compaction targeting key-groups range {}.",
+ dbExpectedKeyGroupRange.prettyPrintInterval());
+ compactionOptions.setCanceled(true);
+ };
+
+ closeableRegistry.registerCloseable(cancelCompactionCloseable);
+
+ if (!rangeCheckResult.leftInRange) {
+ logger.debug(
+ "Compacting left interval in async compaction targeting key-groups range {}.",
+ dbExpectedKeyGroupRange.prettyPrintInterval());
+ // Compact all keys before from the expected key-groups range
+ for (ColumnFamilyHandle columnFamilyHandle : columnFamilyHandles) {
+ try (ResourceGuard.Lease ignored = rocksDBResourceGuard.acquireResource()) {
+ db.compactRange(
+ columnFamilyHandle,
+ // TODO: change to null once this API is fixed
+ new byte[] {},
+ rangeCheckResult.minKey,
+ compactionOptions);
}
+ }
+ }
- if (!rangeCheckResult.rightInRange) {
- // Compact all keys after the expected key-groups range
- for (ColumnFamilyHandle columnFamilyHandle : columnFamilyHandles) {
- db.compactRange(
- columnFamilyHandle,
- rangeCheckResult.maxKey,
- // TODO: change to null once this API is fixed
- new byte[] {
- (byte) 0xff, (byte) 0xff, (byte) 0xff, (byte) 0xff
- },
- compactionOptions);
- }
+ if (!rangeCheckResult.rightInRange) {
+ logger.debug(
+ "Compacting right interval in async compaction targeting key-groups range {}.",
+ dbExpectedKeyGroupRange.prettyPrintInterval());
+ // Compact all keys after the expected key-groups range
+ for (ColumnFamilyHandle columnFamilyHandle : columnFamilyHandles) {
+ try (ResourceGuard.Lease ignored = rocksDBResourceGuard.acquireResource()) {
+ db.compactRange(
+ columnFamilyHandle,
+ rangeCheckResult.maxKey,
+ // TODO: change to null once this API is fixed
+ new byte[] {(byte) 0xff, (byte) 0xff, (byte) 0xff, (byte) 0xff},
+ compactionOptions);
}
}
- });
+ }
+
+ closeableRegistry.unregisterCloseable(cancelCompactionCloseable);
+ }
+ };
}
/**
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
index cef331f2d67..ba4c6570acd 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
@@ -467,7 +467,6 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
// disposed, as
// working on the disposed object results in SEGFAULTS.
if (db != null) {
-
IOUtils.closeQuietly(writeBatchWrapper);
// Metric collection occurs on a background thread. When this method returns
@@ -1018,6 +1017,7 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
return writeBatchSize;
}
+ @VisibleForTesting
public Optional<CompletableFuture<Void>> getAsyncCompactAfterRestoreFuture() {
return Optional.ofNullable(asyncCompactAfterRestoreFuture);
}
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackendBuilder.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackendBuilder.java
index 948c1de2743..88b44462146 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackendBuilder.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackendBuilder.java
@@ -373,6 +373,7 @@ public class RocksDBKeyedStateBackendBuilder<K> extends AbstractKeyedStateBacken
restoreOperation =
getRocksDBRestoreOperation(
keyGroupPrefixBytes,
+ rocksDBResourceGuard,
cancelStreamRegistry,
kvStateInformation,
registeredPQStates,
@@ -381,8 +382,13 @@ public class RocksDBKeyedStateBackendBuilder<K> extends AbstractKeyedStateBacken
db = restoreResult.getDb();
defaultColumnFamilyHandle = restoreResult.getDefaultColumnFamilyHandle();
nativeMetricMonitor = restoreResult.getNativeMetricMonitor();
- asyncCompactAfterRestoreFuture =
- restoreResult.getAsyncCompactAfterRestoreFuture().orElse(null);
+ if (ioExecutor != null) {
+ asyncCompactAfterRestoreFuture =
+ restoreResult
+ .getAsyncCompactTaskAfterRestore()
+ .map((task) -> CompletableFuture.runAsync(task, ioExecutor))
+ .orElse(null);
+ }
if (restoreOperation instanceof RocksDBIncrementalRestoreOperation) {
backendUID = restoreResult.getBackendUID();
materializedSstFiles = restoreResult.getRestoredSstFiles();
@@ -431,6 +437,7 @@ public class RocksDBKeyedStateBackendBuilder<K> extends AbstractKeyedStateBacken
new ArrayList<>(kvStateInformation.values().size());
IOUtils.closeQuietly(cancelStreamRegistryForBackend);
IOUtils.closeQuietly(writeBatchWrapper);
+ IOUtils.closeQuietly(rocksDBResourceGuard);
RocksDBOperationUtils.addColumnFamilyOptionsToCloseLater(
columnFamilyOptions, defaultColumnFamilyHandle);
IOUtils.closeQuietly(defaultColumnFamilyHandle);
@@ -498,6 +505,7 @@ public class RocksDBKeyedStateBackendBuilder<K> extends AbstractKeyedStateBacken
private RocksDBRestoreOperation getRocksDBRestoreOperation(
int keyGroupPrefixBytes,
+ ResourceGuard rocksDBResourceGuard,
CloseableRegistry cancelStreamRegistry,
LinkedHashMap<String, RocksDBKeyedStateBackend.RocksDbKvStateInfo> kvStateInformation,
LinkedHashMap<String, HeapPriorityQueueSnapshotRestoreWrapper<?>> registeredPQStates,
@@ -521,6 +529,7 @@ public class RocksDBKeyedStateBackendBuilder<K> extends AbstractKeyedStateBacken
keyGroupRange,
keyGroupPrefixBytes,
numberOfTransferingThreads,
+ rocksDBResourceGuard,
cancelStreamRegistry,
userCodeClassLoader,
kvStateInformation,
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBOperationUtils.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBOperationUtils.java
index 4750969d050..b511bbfc620 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBOperationUtils.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBOperationUtils.java
@@ -142,22 +142,27 @@ public class RocksDBOperationUtils {
ttlCompactFiltersManager,
writeBufferManagerCapacity);
- final ColumnFamilyHandle columnFamilyHandle;
try {
- columnFamilyHandle =
+ ColumnFamilyHandle columnFamilyHandle =
createColumnFamily(
columnFamilyDescriptor,
db,
importFilesMetaData,
cancelStreamRegistryForRestore);
+ return new RocksDBKeyedStateBackend.RocksDbKvStateInfo(
+ columnFamilyHandle, metaInfoBase);
} catch (Exception ex) {
IOUtils.closeQuietly(columnFamilyDescriptor.getOptions());
throw new FlinkRuntimeException("Error creating ColumnFamilyHandle.", ex);
}
-
- return new RocksDBKeyedStateBackend.RocksDbKvStateInfo(columnFamilyHandle, metaInfoBase);
}
+ /**
+ * Create RocksDB-backed KV-state, including RocksDB ColumnFamily.
+ *
+ * @param cancelStreamRegistryForRestore {@link ICloseableRegistry#close closing} it interrupts
+ * KV state creation
+ */
public static RocksDBKeyedStateBackend.RocksDbKvStateInfo createStateInfo(
RegisteredStateMetaInfoBase metaInfoBase,
RocksDB db,
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java
index ae4d7b8ed0e..9aae10f2b5e 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java
@@ -50,9 +50,9 @@ 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.apache.flink.util.ResourceGuard;
import org.apache.flink.util.StateMigrationException;
import org.apache.flink.util.clock.SystemClock;
-import org.apache.flink.util.concurrent.FutureUtils;
import org.apache.flink.util.function.RunnableWithException;
import org.rocksdb.ColumnFamilyDescriptor;
@@ -68,6 +68,7 @@ import org.slf4j.LoggerFactory;
import javax.annotation.Nonnegative;
import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
import java.io.Closeable;
import java.io.File;
@@ -84,9 +85,7 @@ import java.util.Objects;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.UUID;
-import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
import java.util.function.Function;
import java.util.stream.Collectors;
@@ -122,6 +121,8 @@ public class RocksDBIncrementalRestoreOperation<K> implements RocksDBRestoreOper
private final StateSerializerProvider<K> keySerializerProvider;
private final ClassLoader userCodeClassLoader;
private final CustomInitializationMetrics customInitializationMetrics;
+ private final ResourceGuard dbResourceGuard;
+
private long lastCompletedCheckpointId;
private UUID backendUID;
private final long writeBatchSize;
@@ -142,6 +143,7 @@ public class RocksDBIncrementalRestoreOperation<K> implements RocksDBRestoreOper
KeyGroupRange keyGroupRange,
int keyGroupPrefixBytes,
int numberOfTransferringThreads,
+ ResourceGuard dbResourceGuard,
CloseableRegistry cancelStreamRegistry,
ClassLoader userCodeClassLoader,
Map<String, RocksDbKvStateInfo> kvStateInformation,
@@ -180,6 +182,7 @@ public class RocksDBIncrementalRestoreOperation<K> implements RocksDBRestoreOper
this.overlapFractionThreshold = overlapFractionThreshold;
this.customInitializationMetrics = customInitializationMetrics;
this.restoreStateHandles = restoreStateHandles;
+ this.dbResourceGuard = dbResourceGuard;
this.cancelStreamRegistry = cancelStreamRegistry;
this.keyGroupRange = keyGroupRange;
this.instanceBasePath = instanceBasePath;
@@ -233,57 +236,11 @@ public class RocksDBIncrementalRestoreOperation<K> implements RocksDBRestoreOper
runAndReportDuration(
() -> restoreFromLocalState(localKeyedStateHandles), RESTORE_STATE_DURATION);
- CompletableFuture<Void> asyncCompactFuture = null;
- if (asyncCompactAfterRescale) {
- asyncCompactFuture =
- RocksDBIncrementalCheckpointUtils.createRangeCompactionTaskIfNeeded(
- rocksHandle.getDb(),
- rocksHandle.getColumnFamilyHandles(),
- keyGroupPrefixBytes,
- keyGroupRange)
- .map(
- (run) -> {
- RunnableWithException runWithLogging =
- () -> {
- long t = System.currentTimeMillis();
- logger.info(
- "Starting async compaction after restore for backend {} in operator {}",
- backendUID,
- operatorIdentifier);
- try {
- runAndReportDuration(
- run,
- RESTORE_ASYNC_COMPACTION_DURATION);
- logger.info(
- "Completed async compaction after restore for backend {} in operator {} after {} ms.",
- backendUID,
- operatorIdentifier,
- System.currentTimeMillis() - t);
- } catch (Exception ex) {
- logger.info(
- "Failed async compaction after restore for backend {} in operator {} after {} ms.",
- backendUID,
- operatorIdentifier,
- System.currentTimeMillis() - t,
- ex);
- throw ex;
- }
- };
- ExecutorService executorService =
- Executors.newSingleThreadExecutor();
- CompletableFuture<Void> resultFuture =
- FutureUtils.runAsync(
- runWithLogging, executorService);
- executorService.shutdown();
- return resultFuture;
- })
- .orElse(null);
- logger.info(
- "Finished RocksDB incremental recovery in operator {} with "
- + "target key-group range range {}.",
- operatorIdentifier,
- keyGroupRange.prettyPrintInterval());
- }
+ logger.info(
+ "Finished RocksDB incremental recovery in operator {} with "
+ + "target key-group range range {}.",
+ operatorIdentifier,
+ keyGroupRange.prettyPrintInterval());
return new RocksDBRestoreResult(
this.rocksHandle.getDb(),
@@ -292,7 +249,7 @@ public class RocksDBIncrementalRestoreOperation<K> implements RocksDBRestoreOper
lastCompletedCheckpointId,
backendUID,
restoredSstFiles,
- asyncCompactFuture);
+ createAsyncCompactionTask());
} finally {
// Cleanup all download directories
allDownloadSpecs.stream()
@@ -301,6 +258,47 @@ public class RocksDBIncrementalRestoreOperation<K> implements RocksDBRestoreOper
}
}
+ @Nullable
+ private Runnable createAsyncCompactionTask() {
+
+ if (!asyncCompactAfterRescale) {
+ return null;
+ }
+
+ return () -> {
+ long t = System.currentTimeMillis();
+ logger.info(
+ "Starting async compaction after restore for backend {} in operator {}",
+ backendUID,
+ operatorIdentifier);
+ try {
+ RunnableWithException asyncRangeCompactionTask =
+ RocksDBIncrementalCheckpointUtils.createAsyncRangeCompactionTask(
+ rocksHandle.getDb(),
+ rocksHandle.getColumnFamilyHandles(),
+ keyGroupPrefixBytes,
+ keyGroupRange,
+ dbResourceGuard,
+ cancelStreamRegistry);
+ runAndReportDuration(asyncRangeCompactionTask, RESTORE_ASYNC_COMPACTION_DURATION);
+ logger.info(
+ "Completed async compaction after restore for backend {} in operator {} after {} ms.",
+ backendUID,
+ operatorIdentifier,
+ System.currentTimeMillis() - t);
+ } catch (Throwable throwable) {
+ // We don't rethrow because the executing thread might have a fatal exception
+ // handler.
+ logger.info(
+ "Failed async compaction after restore for backend {} in operator {} after {} ms.",
+ backendUID,
+ operatorIdentifier,
+ System.currentTimeMillis() - t,
+ throwable);
+ }
+ };
+ }
+
private void restoreFromLocalState(
List<IncrementalLocalKeyedStateHandle> localKeyedStateHandles) throws Exception {
if (localKeyedStateHandles.size() == 1) {
@@ -519,28 +517,40 @@ public class RocksDBIncrementalRestoreOperation<K> implements RocksDBRestoreOper
logger.info(
"Starting restore export for backend with range {} in operator {}.",
- keyGroupRange,
+ keyGroupRange.prettyPrintInterval(),
operatorIdentifier);
int minExportKeyGroup = Integer.MAX_VALUE;
int maxExportKeyGroup = Integer.MIN_VALUE;
+ int index = 0;
for (IncrementalLocalKeyedStateHandle stateHandle : localKeyedStateHandles) {
+
+ final String logLineSuffix =
+ " for state handle at index "
+ + index
+ + " with proclaimed key-group range "
+ + stateHandle.getKeyGroupRange().prettyPrintInterval()
+ + " for backend with range "
+ + keyGroupRange.prettyPrintInterval()
+ + " in operator "
+ + operatorIdentifier
+ + ".";
+
+ logger.debug("Opening temporary database" + logLineSuffix);
try (RestoredDBInstance tmpRestoreDBInfo =
restoreTempDBInstanceFromLocalState(stateHandle)) {
List<ColumnFamilyHandle> tmpColumnFamilyHandles =
tmpRestoreDBInfo.columnFamilyHandles;
+ logger.debug("Checking actual keys of sst files" + logLineSuffix);
+
// Check if the data in all SST files referenced in the handle is within the
// proclaimed key-groups range of the handle.
if (RocksDBIncrementalCheckpointUtils.isSstDataInKeyGroupRange(
tmpRestoreDBInfo.db, keyGroupPrefixBytes, stateHandle.getKeyGroupRange())) {
- logger.debug(
- "Exporting state handle {} for backend with range {} in operator {}.",
- stateHandle,
- keyGroupRange,
- operatorIdentifier);
+ logger.debug("Start exporting" + logLineSuffix);
List<RegisteredStateMetaInfoBase> registeredStateMetaInfoBases =
tmpRestoreDBInfo.stateMetaInfoSnapshots.stream()
@@ -565,22 +575,21 @@ public class RocksDBIncrementalRestoreOperation<K> implements RocksDBRestoreOper
maxExportKeyGroup,
stateHandle.getKeyGroupRange().getEndKeyGroup());
- logger.debug(
- "Done exporting state handle {} for backend with range {} in operator {}.",
- stateHandle,
- keyGroupRange,
- operatorIdentifier);
+ logger.debug("Done exporting" + logLineSuffix);
} else {
// Actual key range in files exceeds proclaimed range, cannot import. We
// will copy this handle using a temporary DB later.
skipped.add(stateHandle);
+ logger.debug("Skipped export" + logLineSuffix);
}
}
+
+ ++index;
}
logger.info(
"Completed restore export for backend with range {} in operator {}. Number of Exported handles: {}. Skipped handles: {}.",
- keyGroupRange,
+ keyGroupRange.prettyPrintInterval(),
operatorIdentifier,
localKeyedStateHandles.size() - skipped.size(),
skipped);
@@ -647,7 +656,7 @@ public class RocksDBIncrementalRestoreOperation<K> implements RocksDBRestoreOper
// We initialize the base DB by importing all the exported data.
logger.info(
"Starting to import exported state handles for backend with range {} in operator {} using Clip/Ingest DB.",
- keyGroupRange,
+ keyGroupRange.prettyPrintInterval(),
operatorIdentifier);
rocksHandle.openDB();
for (Map.Entry<RegisteredStateMetaInfoBase, List<ExportImportFilesMetaData>> entry :
@@ -667,7 +676,7 @@ public class RocksDBIncrementalRestoreOperation<K> implements RocksDBRestoreOper
logger.info(
"Completed importing exported state handles for backend with range {} in operator {} using Clip/Ingest DB.",
- keyGroupRange,
+ keyGroupRange.prettyPrintInterval(),
operatorIdentifier);
}
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBRestoreResult.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBRestoreResult.java
index 1e458eb45f4..5f565b28950 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBRestoreResult.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBRestoreResult.java
@@ -30,7 +30,6 @@ import java.util.Collection;
import java.util.Optional;
import java.util.SortedMap;
import java.util.UUID;
-import java.util.concurrent.CompletableFuture;
/** Entity holding result of RocksDB instance restore. */
public class RocksDBRestoreResult {
@@ -43,7 +42,7 @@ public class RocksDBRestoreResult {
private final UUID backendUID;
private final SortedMap<Long, Collection<HandleAndLocalPath>> restoredSstFiles;
- private final CompletableFuture<Void> asyncCompactAfterRestoreFuture;
+ private final Runnable asyncCompactTaskAfterRestore;
public RocksDBRestoreResult(
RocksDB db,
@@ -52,14 +51,14 @@ public class RocksDBRestoreResult {
long lastCompletedCheckpointId,
UUID backendUID,
SortedMap<Long, Collection<HandleAndLocalPath>> restoredSstFiles,
- @Nullable CompletableFuture<Void> asyncCompactAfterRestoreFuture) {
+ @Nullable Runnable asyncCompactTaskAfterRestore) {
this.db = db;
this.defaultColumnFamilyHandle = defaultColumnFamilyHandle;
this.nativeMetricMonitor = nativeMetricMonitor;
this.lastCompletedCheckpointId = lastCompletedCheckpointId;
this.backendUID = backendUID;
this.restoredSstFiles = restoredSstFiles;
- this.asyncCompactAfterRestoreFuture = asyncCompactAfterRestoreFuture;
+ this.asyncCompactTaskAfterRestore = asyncCompactTaskAfterRestore;
}
public RocksDB getDb() {
@@ -86,7 +85,7 @@ public class RocksDBRestoreResult {
return nativeMetricMonitor;
}
- public Optional<CompletableFuture<Void>> getAsyncCompactAfterRestoreFuture() {
- return Optional.ofNullable(asyncCompactAfterRestoreFuture);
+ public Optional<Runnable> getAsyncCompactTaskAfterRestore() {
+ return Optional.ofNullable(asyncCompactTaskAfterRestore);
}
}
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBRecoveryTest.java b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBRecoveryTest.java
index ce4d93ea0be..3e55dd553a7 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBRecoveryTest.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBRecoveryTest.java
@@ -45,6 +45,8 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.UUID;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
import java.util.concurrent.RunnableFuture;
import java.util.stream.Collectors;
@@ -108,6 +110,7 @@ public class RocksDBRecoveryTest {
int startParallelism, int targetParallelism, int numKeys, int updateDistance)
throws Exception {
+ ExecutorService ioExecutor = Executors.newSingleThreadExecutor();
OUTPUT.println("Rescaling from " + startParallelism + " to " + targetParallelism + "...");
final String stateName = "TestValueState";
final int maxParallelism = startParallelism * targetParallelism;
@@ -133,6 +136,7 @@ public class RocksDBRecoveryTest {
Collections.emptyList())
.setEnableIncrementalCheckpointing(true)
.setUseIngestDbRestoreMode(true)
+ .setIOExecutor(ioExecutor)
.build();
valueStates.add(
@@ -183,7 +187,8 @@ public class RocksDBRecoveryTest {
targetParallelism,
maxParallelism,
startSnapshotResult,
- backends);
+ backends,
+ ioExecutor);
backends.forEach(
backend ->
@@ -216,7 +221,8 @@ public class RocksDBRecoveryTest {
startParallelism,
maxParallelism,
rescaleSnapshotResult,
- backends);
+ backends,
+ ioExecutor);
count = 0;
for (RocksDBKeyedStateBackend<Integer> backend : backends) {
@@ -243,6 +249,7 @@ public class RocksDBRecoveryTest {
for (SnapshotResult<KeyedStateHandle> snapshotResult : cleanupSnapshotResult) {
snapshotResult.discardState();
}
+ ioExecutor.shutdown();
}
}
@@ -252,7 +259,8 @@ public class RocksDBRecoveryTest {
int targetParallelism,
int maxParallelism,
List<SnapshotResult<KeyedStateHandle>> snapshotResult,
- List<RocksDBKeyedStateBackend<Integer>> backendsOut)
+ List<RocksDBKeyedStateBackend<Integer>> backendsOut,
+ ExecutorService ioExecutor)
throws IOException {
List<KeyedStateHandle> stateHandles =
@@ -290,6 +298,7 @@ public class RocksDBRecoveryTest {
.setUseIngestDbRestoreMode(useIngest)
.setIncrementalRestoreAsyncCompactAfterRescale(asyncCompactAfterRescale)
.setRescalingUseDeleteFilesInRange(true)
+ .setIOExecutor(ioExecutor)
.build();
long instanceTime = System.currentTimeMillis() - tInstance;
(flink) 03/08: [FLINK-35580] Introduce artificial map key for RegisteredStateMetaInfoBase for safe deduplication in RocksDBIncrementalRestoreOperation.
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 9b2b922a159dec767b7a76cf0aab35c3e12ef5c2
Author: Stefan Richter <sr...@confluent.io>
AuthorDate: Thu Mar 7 18:27:00 2024 +0100
[FLINK-35580] Introduce artificial map key for RegisteredStateMetaInfoBase for safe deduplication in RocksDBIncrementalRestoreOperation.
---
.../runtime/state/RegisteredStateMetaInfoBase.java | 46 ++++++++++++++++++++++
.../state/RocksDBIncrementalCheckpointUtils.java | 7 ++--
.../streaming/state/restore/RocksDBHandle.java | 7 +++-
.../RocksDBIncrementalRestoreOperation.java | 16 ++++----
.../streaming/state/RocksDBRecoveryTest.java | 32 +++++++++++----
.../contrib/streaming/state/RocksDBTestUtils.java | 2 +-
6 files changed, 88 insertions(+), 22 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredStateMetaInfoBase.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredStateMetaInfoBase.java
index 47ef32ca678..f783f5a9575 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredStateMetaInfoBase.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredStateMetaInfoBase.java
@@ -22,6 +22,8 @@ import org.apache.flink.runtime.state.metainfo.StateMetaInfoSnapshot;
import javax.annotation.Nonnull;
+import java.util.Objects;
+
/** Base class for all registered state in state backends. */
public abstract class RegisteredStateMetaInfoBase {
@@ -68,4 +70,48 @@ public abstract class RegisteredStateMetaInfoBase {
"Unknown backend state type: " + backendStateType);
}
}
+
+ /** Returns a wrapper that can be used as a key in {@link java.util.Map}. */
+ public final Key asMapKey() {
+ return new Key(this);
+ }
+
+ /**
+ * Wrapper class that can be used to represent the wrapped {@link RegisteredStateMetaInfoBase}
+ * as key in a {@link java.util.Map}.
+ */
+ public static final class Key {
+ private final RegisteredStateMetaInfoBase registeredStateMetaInfoBase;
+
+ private Key(RegisteredStateMetaInfoBase metaInfoBase) {
+ this.registeredStateMetaInfoBase = metaInfoBase;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ Key that = (Key) o;
+ return Objects.equals(
+ registeredStateMetaInfoBase.getName(),
+ that.registeredStateMetaInfoBase.getName())
+ && Objects.equals(
+ registeredStateMetaInfoBase.getClass(),
+ that.registeredStateMetaInfoBase.getClass());
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(
+ registeredStateMetaInfoBase.getName(), registeredStateMetaInfoBase.getClass());
+ }
+
+ public RegisteredStateMetaInfoBase getRegisteredStateMetaInfoBase() {
+ return registeredStateMetaInfoBase;
+ }
+ }
}
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 e3be413307d..41c2f016d47 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
@@ -406,7 +406,7 @@ public class RocksDBIncrementalCheckpointUtils {
List<ColumnFamilyHandle> columnFamilyHandles,
List<RegisteredStateMetaInfoBase> registeredStateMetaInfoBases,
Path exportBasePath,
- Map<RegisteredStateMetaInfoBase, List<ExportImportFilesMetaData>> resultOutput)
+ Map<RegisteredStateMetaInfoBase.Key, List<ExportImportFilesMetaData>> resultOutput)
throws RocksDBException {
Preconditions.checkArgument(
@@ -415,7 +415,8 @@ public class RocksDBIncrementalCheckpointUtils {
try (final Checkpoint checkpoint = Checkpoint.create(db)) {
for (int i = 0; i < columnFamilyHandles.size(); i++) {
- RegisteredStateMetaInfoBase stateMetaInfo = registeredStateMetaInfoBases.get(i);
+ RegisteredStateMetaInfoBase.Key stateMetaInfoAsKey =
+ registeredStateMetaInfoBases.get(i).asMapKey();
Path subPath = exportBasePath.resolve(UUID.randomUUID().toString());
ExportImportFilesMetaData exportedColumnFamilyMetaData =
@@ -428,7 +429,7 @@ public class RocksDBIncrementalCheckpointUtils {
if (exportedSstFiles != null && exportedSstFiles.length > 0) {
resultOutput
- .computeIfAbsent(stateMetaInfo, (key) -> new ArrayList<>())
+ .computeIfAbsent(stateMetaInfoAsKey, (key) -> new ArrayList<>())
.add(exportedColumnFamilyMetaData);
} else {
// Close unused empty export result
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBHandle.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBHandle.java
index 0453c1eea49..16edc072102 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBHandle.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBHandle.java
@@ -198,14 +198,17 @@ class RocksDBHandle implements AutoCloseable {
/**
* Registers a new column family and imports data from the given export.
*
- * @param stateMetaInfo info about the state to create.
+ * @param stateMetaInfoKey info about the state to create.
* @param cfMetaDataList the data to import.
*/
void registerStateColumnFamilyHandleWithImport(
- RegisteredStateMetaInfoBase stateMetaInfo,
+ RegisteredStateMetaInfoBase.Key stateMetaInfoKey,
List<ExportImportFilesMetaData> cfMetaDataList,
ICloseableRegistry cancelStreamRegistryForRestore) {
+ RegisteredStateMetaInfoBase stateMetaInfo =
+ stateMetaInfoKey.getRegisteredStateMetaInfoBase();
+
Preconditions.checkState(!kvStateInformation.containsKey(stateMetaInfo.getName()));
RocksDbKvStateInfo stateInfo =
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java
index 5c72e78766b..c17c5f3d26e 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java
@@ -473,7 +473,7 @@ public class RocksDBIncrementalRestoreOperation<K> implements RocksDBRestoreOper
final Path exportCfBasePath = absolutInstanceBasePath.resolve("export-cfs");
Files.createDirectories(exportCfBasePath);
- final Map<RegisteredStateMetaInfoBase, List<ExportImportFilesMetaData>>
+ final Map<RegisteredStateMetaInfoBase.Key, List<ExportImportFilesMetaData>>
exportedColumnFamilyMetaData = new HashMap<>(localKeyedStateHandles.size());
final List<IncrementalLocalKeyedStateHandle> notImportableHandles =
@@ -526,7 +526,7 @@ public class RocksDBIncrementalRestoreOperation<K> implements RocksDBRestoreOper
private KeyGroupRange exportColumnFamiliesWithSstDataInKeyGroupsRange(
Path exportCfBasePath,
List<IncrementalLocalKeyedStateHandle> localKeyedStateHandles,
- Map<RegisteredStateMetaInfoBase, List<ExportImportFilesMetaData>>
+ Map<RegisteredStateMetaInfoBase.Key, List<ExportImportFilesMetaData>>
exportedColumnFamiliesOut,
List<IncrementalLocalKeyedStateHandle> skipped)
throws Exception {
@@ -664,7 +664,7 @@ public class RocksDBIncrementalRestoreOperation<K> implements RocksDBRestoreOper
* @throws Exception on import error.
*/
private void initBaseDBFromColumnFamilyImports(
- Map<RegisteredStateMetaInfoBase, List<ExportImportFilesMetaData>>
+ Map<RegisteredStateMetaInfoBase.Key, List<ExportImportFilesMetaData>>
exportedColumnFamilyMetaData,
KeyGroupRange exportKeyGroupRange)
throws Exception {
@@ -675,10 +675,10 @@ public class RocksDBIncrementalRestoreOperation<K> implements RocksDBRestoreOper
keyGroupRange.prettyPrintInterval(),
operatorIdentifier);
rocksHandle.openDB();
- for (Map.Entry<RegisteredStateMetaInfoBase, List<ExportImportFilesMetaData>> entry :
+ for (Map.Entry<RegisteredStateMetaInfoBase.Key, List<ExportImportFilesMetaData>> entry :
exportedColumnFamilyMetaData.entrySet()) {
rocksHandle.registerStateColumnFamilyHandleWithImport(
- entry.getKey(), entry.getValue(), cancelStreamRegistry);
+ entry.getKey(), entry.getValue(), cancelStreamRegistryForRestore);
}
// Use Range delete to clip the temp db to the target range of the backend
@@ -737,7 +737,7 @@ public class RocksDBIncrementalRestoreOperation<K> implements RocksDBRestoreOper
createColumnFamilyDescriptors(stateMetaInfoSnapshots, true),
stateMetaInfoSnapshots,
restoreSourcePath,
- cancelStreamRegistry);
+ cancelStreamRegistryForRestore);
}
/**
@@ -793,7 +793,7 @@ public class RocksDBIncrementalRestoreOperation<K> implements RocksDBRestoreOper
try (RocksDBWriteBatchWrapper writeBatchWrapper =
new RocksDBWriteBatchWrapper(this.rocksHandle.getDb(), writeBatchSize);
Closeable ignored =
- cancelStreamRegistry.registerCloseableTemporarily(
+ cancelStreamRegistryForRestore.registerCloseableTemporarily(
asCloseable(writeBatchWrapper))) {
for (IncrementalLocalKeyedStateHandle handleToCopy : toImport) {
try (RestoredDBInstance restoredDBInstance =
@@ -849,7 +849,7 @@ public class RocksDBIncrementalRestoreOperation<K> implements RocksDBRestoreOper
this.rocksHandle.getOrRegisterStateColumnFamilyHandle(
null,
tmpRestoreDBInfo.stateMetaInfoSnapshots.get(descIdx),
- cancelStreamRegistry)
+ cancelStreamRegistryForRestore)
.columnFamilyHandle;
try (RocksIteratorWrapper iterator =
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBRecoveryTest.java b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBRecoveryTest.java
index 3e55dd553a7..6f7de4967b9 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBRecoveryTest.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBRecoveryTest.java
@@ -24,12 +24,15 @@ import org.apache.flink.api.common.typeutils.base.IntSerializer;
import org.apache.flink.runtime.checkpoint.CheckpointOptions;
import org.apache.flink.runtime.state.KeyGroupRange;
import org.apache.flink.runtime.state.KeyGroupRangeAssignment;
+import org.apache.flink.runtime.state.KeyGroupedInternalPriorityQueue;
import org.apache.flink.runtime.state.KeyedStateHandle;
import org.apache.flink.runtime.state.SnapshotResult;
import org.apache.flink.runtime.state.StateObject;
import org.apache.flink.runtime.state.VoidNamespace;
import org.apache.flink.runtime.state.VoidNamespaceSerializer;
import org.apache.flink.runtime.state.filesystem.FsCheckpointStreamFactory;
+import org.apache.flink.streaming.api.operators.TimerHeapInternalTimer;
+import org.apache.flink.streaming.api.operators.TimerSerializer;
import org.apache.flink.testutils.junit.utils.TempDirUtils;
import org.apache.flink.util.IOUtils;
@@ -68,42 +71,42 @@ public class RocksDBRecoveryTest {
@Test
public void testScaleOut_1_2() throws Exception {
- testRescale(1, 2, 100_000, 10);
+ testRescale(1, 2, 50_000, 10);
}
@Test
public void testScaleOut_2_8() throws Exception {
- testRescale(2, 8, 100_000, 10);
+ testRescale(2, 8, 50_000, 10);
}
@Test
public void testScaleOut_2_7() throws Exception {
- testRescale(2, 7, 100_000, 10);
+ testRescale(2, 7, 50_000, 10);
}
@Test
public void testScaleIn_2_1() throws Exception {
- testRescale(2, 1, 100_000, 10);
+ testRescale(2, 1, 50_000, 10);
}
@Test
public void testScaleIn_8_2() throws Exception {
- testRescale(8, 2, 100_000, 10);
+ testRescale(8, 2, 50_000, 10);
}
@Test
public void testScaleIn_7_2() throws Exception {
- testRescale(7, 2, 100_000, 10);
+ testRescale(7, 2, 50_000, 10);
}
@Test
public void testScaleIn_2_3() throws Exception {
- testRescale(2, 3, 100_000, 10);
+ testRescale(2, 3, 50_000, 10);
}
@Test
public void testScaleIn_3_2() throws Exception {
- testRescale(3, 2, 100_000, 10);
+ testRescale(3, 2, 50_000, 10);
}
public void testRescale(
@@ -120,6 +123,8 @@ public class RocksDBRecoveryTest {
final List<SnapshotResult<KeyedStateHandle>> cleanupSnapshotResult = new ArrayList<>();
try {
final List<ValueState<Integer>> valueStates = new ArrayList<>(maxParallelism);
+ final List<KeyGroupedInternalPriorityQueue<TimerHeapInternalTimer<Integer, Integer>>>
+ timerStates = new ArrayList<>(maxParallelism);
try {
ValueStateDescriptor<Integer> stateDescriptor =
new ValueStateDescriptor<>(stateName, IntSerializer.INSTANCE);
@@ -143,6 +148,12 @@ public class RocksDBRecoveryTest {
backend.getOrCreateKeyedState(
VoidNamespaceSerializer.INSTANCE, stateDescriptor));
+ timerStates.add(
+ backend.create(
+ "timer-state",
+ new TimerSerializer<>(
+ IntSerializer.INSTANCE, IntSerializer.INSTANCE)));
+
backends.add(backend);
}
@@ -156,6 +167,11 @@ public class RocksDBRecoveryTest {
backends.get(index).setCurrentKey(key);
valueStates.get(index).update(i);
+ // Add at most one timer for each instance
+ KeyGroupedInternalPriorityQueue<TimerHeapInternalTimer<Integer, Integer>>
+ timerState = timerStates.get(index);
+ timerState.add(new TimerHeapInternalTimer<>(i, key, 0));
+
if (updateDistance > 0 && i % updateDistance == 0) {
key = i - updateDistance + 1;
index =
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBTestUtils.java b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBTestUtils.java
index 43111df7582..2dfe4256ff2 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBTestUtils.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBTestUtils.java
@@ -69,7 +69,7 @@ public final class RocksDBTestUtils {
return builderForTestDefaults(
instanceBasePath,
keySerializer,
- EmbeddedRocksDBStateBackend.PriorityQueueStateType.HEAP,
+ EmbeddedRocksDBStateBackend.PriorityQueueStateType.ROCKSDB,
numKeyGroups,
keyGroupRange,
stateHandles);
(flink) 07/08: [FLINK-35580] Ensure RocksDB working directory is created before opening DB.
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 830a17155cf04748ab471ddb6b4ad6a8ec1416e5
Author: Stefan Richter <sr...@confluent.io>
AuthorDate: Wed Jun 12 15:11:15 2024 +0200
[FLINK-35580] Ensure RocksDB working directory is created before opening DB.
---
.../contrib/streaming/state/RocksDBOperationUtils.java | 13 +++++++++++--
1 file changed, 11 insertions(+), 2 deletions(-)
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBOperationUtils.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBOperationUtils.java
index b511bbfc620..74800101cd0 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBOperationUtils.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBOperationUtils.java
@@ -43,6 +43,7 @@ import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
+import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
@@ -76,13 +77,21 @@ public class RocksDBOperationUtils {
RocksDB dbRef;
try {
+ // Ensure that the working directory exists and is a directory to make RocksDB happy
+ File pathFile = new File(Preconditions.checkNotNull(path));
+
+ if (!pathFile.exists() && !pathFile.mkdirs() && !pathFile.isDirectory()) {
+ throw new IOException(
+ "Could not create working directory for RocksDB instance: " + path);
+ }
+
dbRef =
RocksDB.open(
Preconditions.checkNotNull(dbOptions),
- Preconditions.checkNotNull(path),
+ path,
columnFamilyDescriptors,
stateColumnFamilyHandles);
- } catch (RocksDBException e) {
+ } catch (Exception e) {
IOUtils.closeQuietly(columnFamilyOptions);
columnFamilyDescriptors.forEach((cfd) -> IOUtils.closeQuietly(cfd.getOptions()));
(flink) 04/08: [FLINK-35580] Handle async compaction exceptions in RocksDB restore.
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 9ba7e67051251436c836d755c169321c16d556bc
Author: Roman Khachatryan <kh...@gmail.com>
AuthorDate: Fri Mar 22 21:08:36 2024 +0100
[FLINK-35580] Handle async compaction exceptions in RocksDB restore.
---
.../state/EmbeddedRocksDBStateBackend.java | 4 +-
.../state/RocksDBIncrementalCheckpointUtils.java | 73 ++++++++++++----------
.../state/RocksDBKeyedStateBackendBuilder.java | 11 +++-
.../RocksDBIncrementalRestoreOperation.java | 20 +++---
4 files changed, 65 insertions(+), 43 deletions(-)
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 8b9903620a8..dfc20ab4033 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
@@ -529,7 +529,9 @@ public class EmbeddedRocksDBStateBackend extends AbstractManagedMemoryStateBacke
.setManualCompactionConfig(
manualCompactionConfig == null
? RocksDBManualCompactionConfig.getDefault()
- : manualCompactionConfig);
+ : manualCompactionConfig)
+ .setAsyncExceptionHandler(
+ (ign, throwable) -> parameters.getEnv().failExternally(throwable));
return builder.build();
}
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 41c2f016d47..e9ab287cf74 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
@@ -287,49 +287,56 @@ public class RocksDBIncrementalCheckpointUtils {
// with the registry
final Closeable cancelCompactionCloseable =
() -> {
- logger.debug(
+ logger.info(
"Cancel request for async compaction targeting key-groups range {}.",
- dbExpectedKeyGroupRange.prettyPrintInterval());
+ dbExpectedKeyGroupRange.prettyPrintInterval(),
+ new Exception("StackTrace"));
compactionOptions.setCanceled(true);
};
- closeableRegistry.registerCloseable(cancelCompactionCloseable);
-
- if (!rangeCheckResult.leftInRange) {
- logger.debug(
- "Compacting left interval in async compaction targeting key-groups range {}.",
- dbExpectedKeyGroupRange.prettyPrintInterval());
- // Compact all keys before from the expected key-groups range
- for (ColumnFamilyHandle columnFamilyHandle : columnFamilyHandles) {
- try (ResourceGuard.Lease ignored = rocksDBResourceGuard.acquireResource()) {
- db.compactRange(
- columnFamilyHandle,
- // TODO: change to null once this API is fixed
- new byte[] {},
- rangeCheckResult.minKey,
- compactionOptions);
+ try {
+ closeableRegistry.registerCloseable(cancelCompactionCloseable);
+
+ if (!rangeCheckResult.leftInRange) {
+ logger.debug(
+ "Compacting left interval in async compaction targeting key-groups range {}.",
+ dbExpectedKeyGroupRange.prettyPrintInterval());
+ // Compact all keys before from the expected key-groups range
+ for (ColumnFamilyHandle columnFamilyHandle : columnFamilyHandles) {
+ try (ResourceGuard.Lease ignored =
+ rocksDBResourceGuard.acquireResource()) {
+ db.compactRange(
+ columnFamilyHandle,
+ // TODO: change to null once this API is fixed
+ new byte[] {},
+ rangeCheckResult.minKey,
+ compactionOptions);
+ }
}
}
- }
- if (!rangeCheckResult.rightInRange) {
- logger.debug(
- "Compacting right interval in async compaction targeting key-groups range {}.",
- dbExpectedKeyGroupRange.prettyPrintInterval());
- // Compact all keys after the expected key-groups range
- for (ColumnFamilyHandle columnFamilyHandle : columnFamilyHandles) {
- try (ResourceGuard.Lease ignored = rocksDBResourceGuard.acquireResource()) {
- db.compactRange(
- columnFamilyHandle,
- rangeCheckResult.maxKey,
- // TODO: change to null once this API is fixed
- new byte[] {(byte) 0xff, (byte) 0xff, (byte) 0xff, (byte) 0xff},
- compactionOptions);
+ if (!rangeCheckResult.rightInRange) {
+ logger.debug(
+ "Compacting right interval in async compaction targeting key-groups range {}.",
+ dbExpectedKeyGroupRange.prettyPrintInterval());
+ // Compact all keys after the expected key-groups range
+ for (ColumnFamilyHandle columnFamilyHandle : columnFamilyHandles) {
+ try (ResourceGuard.Lease ignored =
+ rocksDBResourceGuard.acquireResource()) {
+ db.compactRange(
+ columnFamilyHandle,
+ rangeCheckResult.maxKey,
+ // TODO: change to null once this API is fixed
+ new byte[] {
+ (byte) 0xff, (byte) 0xff, (byte) 0xff, (byte) 0xff
+ },
+ compactionOptions);
+ }
}
}
+ } finally {
+ closeableRegistry.unregisterCloseable(cancelCompactionCloseable);
}
-
- closeableRegistry.unregisterCloseable(cancelCompactionCloseable);
}
};
}
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackendBuilder.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackendBuilder.java
index 7b312d759ad..9eb1980ddf9 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackendBuilder.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackendBuilder.java
@@ -54,6 +54,7 @@ import org.apache.flink.runtime.state.heap.HeapPriorityQueueSetFactory;
import org.apache.flink.runtime.state.heap.HeapPriorityQueueSnapshotRestoreWrapper;
import org.apache.flink.runtime.state.metrics.LatencyTrackingStateConfig;
import org.apache.flink.runtime.state.ttl.TtlTimeProvider;
+import org.apache.flink.runtime.taskmanager.AsyncExceptionHandler;
import org.apache.flink.util.CollectionUtil;
import org.apache.flink.util.FileUtils;
import org.apache.flink.util.IOUtils;
@@ -145,6 +146,7 @@ public class RocksDBKeyedStateBackendBuilder<K> extends AbstractKeyedStateBacken
private RocksDBManualCompactionConfig manualCompactionConfig =
RocksDBManualCompactionConfig.getDefault();
private ExecutorService ioExecutor;
+ private AsyncExceptionHandler asyncExceptionHandler;
public RocksDBKeyedStateBackendBuilder(
String operatorIdentifier,
@@ -553,7 +555,8 @@ public class RocksDBKeyedStateBackendBuilder<K> extends AbstractKeyedStateBacken
useIngestDbRestoreMode,
incrementalRestoreAsyncCompactAfterRescale,
rescalingUseDeleteFilesInRange,
- ioExecutor);
+ ioExecutor,
+ asyncExceptionHandler);
} else if (priorityQueueConfig.getPriorityQueueStateType()
== EmbeddedRocksDBStateBackend.PriorityQueueStateType.HEAP) {
return new RocksDBHeapTimersFullRestoreOperation<>(
@@ -694,4 +697,10 @@ public class RocksDBKeyedStateBackendBuilder<K> extends AbstractKeyedStateBacken
this.manualCompactionConfig = checkNotNull(manualCompactionConfig);
return this;
}
+
+ public RocksDBKeyedStateBackendBuilder<K> setAsyncExceptionHandler(
+ AsyncExceptionHandler asyncExceptionHandler) {
+ this.asyncExceptionHandler = asyncExceptionHandler;
+ return this;
+ }
}
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java
index c17c5f3d26e..6d7ed6e374d 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java
@@ -47,6 +47,7 @@ import org.apache.flink.runtime.state.StateBackend.CustomInitializationMetrics;
import org.apache.flink.runtime.state.StateSerializerProvider;
import org.apache.flink.runtime.state.StreamStateHandle;
import org.apache.flink.runtime.state.metainfo.StateMetaInfoSnapshot;
+import org.apache.flink.runtime.taskmanager.AsyncExceptionHandler;
import org.apache.flink.util.FileUtils;
import org.apache.flink.util.IOUtils;
import org.apache.flink.util.Preconditions;
@@ -150,6 +151,8 @@ public class RocksDBIncrementalRestoreOperation<K> implements RocksDBRestoreOper
private final ExecutorService ioExecutor;
+ private final AsyncExceptionHandler asyncExceptionHandler;
+
public RocksDBIncrementalRestoreOperation(
String operatorIdentifier,
KeyGroupRange keyGroupRange,
@@ -176,7 +179,8 @@ public class RocksDBIncrementalRestoreOperation<K> implements RocksDBRestoreOper
boolean useIngestDbRestoreMode,
boolean asyncCompactAfterRescale,
boolean useDeleteFilesInRange,
- ExecutorService ioExecutor) {
+ ExecutorService ioExecutor,
+ AsyncExceptionHandler asyncExceptionHandler) {
this.rocksHandle =
new RocksDBHandle(
kvStateInformation,
@@ -208,6 +212,7 @@ public class RocksDBIncrementalRestoreOperation<K> implements RocksDBRestoreOper
this.asyncCompactAfterRescale = asyncCompactAfterRescale;
this.useDeleteFilesInRange = useDeleteFilesInRange;
this.ioExecutor = ioExecutor;
+ this.asyncExceptionHandler = asyncExceptionHandler;
}
/**
@@ -303,13 +308,12 @@ public class RocksDBIncrementalRestoreOperation<K> implements RocksDBRestoreOper
operatorIdentifier,
System.currentTimeMillis() - t);
} catch (Throwable throwable) {
- // We don't rethrow because the executing thread might have a fatal exception
- // handler.
- logger.info(
- "Failed async compaction after restore for backend {} in operator {} after {} ms.",
- backendUID,
- operatorIdentifier,
- System.currentTimeMillis() - t,
+ asyncExceptionHandler.handleAsyncException(
+ String.format(
+ "Failed async compaction after restore for backend {} in operator {} after {} ms.",
+ backendUID,
+ operatorIdentifier,
+ System.currentTimeMillis() - t),
throwable);
}
};
(flink) 05/08: [FLINK-35580] Improve logging and problem analysis in RockDBHandle.
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 deeb42ec897560413ca92ce8b4d09c57fe06882f
Author: Stefan Richter <sr...@confluent.io>
AuthorDate: Fri Mar 8 10:38:31 2024 +0100
[FLINK-35580] Improve logging and problem analysis in RockDBHandle.
---
.../apache/flink/contrib/streaming/state/restore/RocksDBHandle.java | 4 +++-
1 file changed, 3 insertions(+), 1 deletion(-)
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBHandle.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBHandle.java
index 16edc072102..acb34928bca 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBHandle.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBHandle.java
@@ -209,7 +209,9 @@ class RocksDBHandle implements AutoCloseable {
RegisteredStateMetaInfoBase stateMetaInfo =
stateMetaInfoKey.getRegisteredStateMetaInfoBase();
- Preconditions.checkState(!kvStateInformation.containsKey(stateMetaInfo.getName()));
+ Preconditions.checkState(
+ !kvStateInformation.containsKey(stateMetaInfo.getName()),
+ "Error: stateMetaInfo.name is not unique:" + stateMetaInfo.getName());
RocksDbKvStateInfo stateInfo =
RocksDBOperationUtils.createStateInfo(