You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by tk...@apache.org on 2023/03/14 17:44:17 UTC
[ignite-3] branch main updated: IGNITE-18763 Implement parallel Watch processing (#1748)
This is an automated email from the ASF dual-hosted git repository.
tkalkirill pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new 329283caf8 IGNITE-18763 Implement parallel Watch processing (#1748)
329283caf8 is described below
commit 329283caf8528d602d069fe7887ae5c639ce7ed8
Author: Alexander Polovtcev <al...@gmail.com>
AuthorDate: Tue Mar 14 19:44:10 2023 +0200
IGNITE-18763 Implement parallel Watch processing (#1748)
---
.../configuration/ConfigurationChanger.java | 105 +++--
.../configuration/ConfigurationRegistry.java | 95 ++--
.../storage/ConfigurationStorageListener.java | 9 +-
.../configuration/TestConfigurationChanger.java | 24 +-
.../ignite/internal/causality/VersionedValue.java | 107 ++---
.../internal/causality/VersionedValueTest.java | 39 +-
.../distributionzones/DistributionZoneManager.java | 17 +-
...ibutionZoneManagerConfigurationChangesTest.java | 10 +-
...butionZoneManagerLogicalTopologyEventsTest.java | 4 +-
.../DistributionZoneManagerScaleUpTest.java | 5 +-
.../DistributionZoneManagerWatchListenerTest.java | 10 +-
.../internal/metastorage/MetaStorageManager.java | 6 +-
.../ignite/internal/metastorage/WatchEvent.java | 28 +-
.../ignite/internal/metastorage/WatchListener.java | 21 +-
.../impl/ItMetaStorageManagerImplTest.java | 37 +-
.../impl/ItMetaStorageMultipleNodesTest.java | 10 +-
.../metastorage/impl/ItMetaStorageWatchTest.java | 46 +-
.../command/MetastorageCommandsMessageGroup.java | 3 +
.../metastorage/impl/MetaStorageManagerImpl.java | 86 ++--
.../metastorage/server/KeyValueStorage.java | 6 -
.../server/OnRevisionAppliedCallback.java | 13 +-
.../ignite/internal/metastorage/server/Watch.java | 46 +-
.../metastorage/server/WatchProcessor.java | 111 +++--
.../server/persistence/RocksDbKeyValueStorage.java | 103 ++---
.../server/AbstractKeyValueStorageTest.java | 52 ++-
.../server/RocksDbKeyValueStorageTest.java | 23 +-
.../metastorage/server/WatchProcessorTest.java | 225 ++++++++++
.../server/SimpleInMemoryKeyValueStorage.java | 17 +-
.../placementdriver/AssignmentsTracker.java | 11 +-
.../internal/placementdriver/LeaseTracker.java | 11 +-
.../ItDistributedConfigurationPropertiesTest.java | 23 +-
.../ItDistributedConfigurationStorageTest.java | 17 +-
.../storage/ItRebalanceDistributedTest.java | 29 +-
.../internal/test/WatchListenerInhibitor.java | 44 +-
.../storage/DistributedConfigurationStorage.java | 124 +++---
.../storage/ConfigurationStorageTest.java | 13 +-
.../DistributedConfigurationCatchUpTest.java | 31 +-
.../internal/table/distributed/TableManager.java | 495 ++++++++++++---------
38 files changed, 1319 insertions(+), 737 deletions(-)
diff --git a/modules/configuration/src/main/java/org/apache/ignite/internal/configuration/ConfigurationChanger.java b/modules/configuration/src/main/java/org/apache/ignite/internal/configuration/ConfigurationChanger.java
index 6a161ee804..93ce215b18 100644
--- a/modules/configuration/src/main/java/org/apache/ignite/internal/configuration/ConfigurationChanger.java
+++ b/modules/configuration/src/main/java/org/apache/ignite/internal/configuration/ConfigurationChanger.java
@@ -63,6 +63,7 @@ import org.apache.ignite.configuration.validation.ValidationIssue;
import org.apache.ignite.configuration.validation.Validator;
import org.apache.ignite.internal.configuration.direct.KeyPathNode;
import org.apache.ignite.internal.configuration.storage.ConfigurationStorage;
+import org.apache.ignite.internal.configuration.storage.ConfigurationStorageListener;
import org.apache.ignite.internal.configuration.storage.Data;
import org.apache.ignite.internal.configuration.tree.ConfigurationSource;
import org.apache.ignite.internal.configuration.tree.ConfigurationVisitor;
@@ -88,7 +89,7 @@ public abstract class ConfigurationChanger implements DynamicConfigurationChange
private final Map<MemberKey, Map<Annotation, Set<Validator<?, ?>>>> cachedAnnotations = new ConcurrentHashMap<>();
/** Closure to execute when an update from the storage is received. */
- private final Notificator notificator;
+ private final ConfigurationUpdateListener configurationUpdateListener;
/** Root keys. Mapping: {@link RootKey#key()} -> identity (itself). */
private final Map<String, RootKey<?, ?>> rootKeys;
@@ -102,7 +103,7 @@ public abstract class ConfigurationChanger implements DynamicConfigurationChange
/** Storage trees. */
private volatile StorageRoots storageRoots;
- /** Configuration listener notification counter, must be incremented before each use of {@link #notificator}. */
+ /** Configuration listener notification counter, must be incremented before each use of {@link #configurationUpdateListener}. */
private final AtomicLong notificationListenerCnt = new AtomicLong();
/** Lock for reading/updating the {@link #storageRoots}. Fair, to give a higher priority to external updates. */
@@ -112,8 +113,7 @@ public abstract class ConfigurationChanger implements DynamicConfigurationChange
* Closure interface to be used by the configuration changer. An instance of this closure is passed into the constructor and invoked
* every time when there's an update from any of the storages.
*/
- @FunctionalInterface
- public interface Notificator {
+ public interface ConfigurationUpdateListener {
/**
* Invoked every time when the configuration is updated.
*
@@ -123,7 +123,18 @@ public abstract class ConfigurationChanger implements DynamicConfigurationChange
* @param notificationNumber Configuration listener notification number.
* @return Future that must signify when processing is completed. Exceptional completion is not expected.
*/
- CompletableFuture<Void> notify(@Nullable SuperRoot oldRoot, SuperRoot newRoot, long storageRevision, long notificationNumber);
+ CompletableFuture<Void> onConfigurationUpdated(
+ @Nullable SuperRoot oldRoot, SuperRoot newRoot, long storageRevision, long notificationNumber
+ );
+
+ /**
+ * Invoked every time when a storage revision has been updated, but no data has been modified.
+ *
+ * @param storageRevision Configuration revision of the storage.
+ * @param notificationNumber Configuration listener notification number.
+ * @return Future that must signify when processing is completed. Exceptional completion is not expected.
+ */
+ CompletableFuture<Void> onRevisionUpdated(long storageRevision, long notificationNumber);
}
/**
@@ -185,21 +196,21 @@ public abstract class ConfigurationChanger implements DynamicConfigurationChange
/**
* Constructor.
*
- * @param notificator Closure to execute when update from the storage is received.
+ * @param configurationUpdateListener Closure to execute when update from the storage is received.
* @param rootKeys Configuration root keys.
* @param validators Validators.
* @param storage Configuration storage.
* @throws IllegalArgumentException If the configuration type of the root keys is not equal to the storage type.
*/
public ConfigurationChanger(
- Notificator notificator,
+ ConfigurationUpdateListener configurationUpdateListener,
Collection<RootKey<?, ?>> rootKeys,
Collection<Validator<?, ?>> validators,
ConfigurationStorage storage
) {
checkConfigurationType(rootKeys, storage);
- this.notificator = notificator;
+ this.configurationUpdateListener = configurationUpdateListener;
this.validators = List.copyOf(validators);
this.storage = storage;
@@ -265,7 +276,7 @@ public abstract class ConfigurationChanger implements DynamicConfigurationChange
storageRoots = new StorageRoots(superRoot, data.changeId());
- storage.registerConfigurationListener(this::updateFromListener);
+ storage.registerConfigurationListener(configurationStorageListener());
}
/**
@@ -588,6 +599,10 @@ public abstract class ConfigurationChanger implements DynamicConfigurationChange
throw new ConfigurationValidationException(validationIssues);
}
+ // "allChanges" map can be empty here in case the given update matches the current state of the local configuration. We
+ // still try to write the empty update, because local configuration can be obsolete. If this is the case, then the CAS will
+ // fail and the update will be recalculated and there is a chance that the new local configuration will produce a non-empty
+ // update.
return storage.write(allChanges, localRoots.version)
.thenCompose(casWroteSuccessfully -> {
if (casWroteSuccessfully) {
@@ -603,48 +618,62 @@ public abstract class ConfigurationChanger implements DynamicConfigurationChange
}
}
- /**
- * Updates configuration from storage listener.
- *
- * @param changedEntries Changed data.
- * @return Future that signifies update completion.
- */
- private CompletableFuture<Void> updateFromListener(Data changedEntries) {
- StorageRoots oldStorageRoots = storageRoots;
+ private ConfigurationStorageListener configurationStorageListener() {
+ return new ConfigurationStorageListener() {
+ @Override
+ public CompletableFuture<Void> onEntriesChanged(Data changedEntries) {
+ StorageRoots oldStorageRoots = storageRoots;
- Map<String, ?> dataValuesPrefixMap = toPrefixMap(changedEntries.values());
+ SuperRoot oldSuperRoot = oldStorageRoots.roots;
+ SuperRoot newSuperRoot = oldSuperRoot.copy();
- compressDeletedEntries(dataValuesPrefixMap);
+ Map<String, ?> dataValuesPrefixMap = toPrefixMap(changedEntries.values());
- SuperRoot oldSuperRoot = oldStorageRoots.roots;
- SuperRoot newSuperRoot = oldSuperRoot.copy();
+ compressDeletedEntries(dataValuesPrefixMap);
- fillFromPrefixMap(newSuperRoot, dataValuesPrefixMap);
+ fillFromPrefixMap(newSuperRoot, dataValuesPrefixMap);
- long newChangeId = changedEntries.changeId();
+ long newChangeId = changedEntries.changeId();
- rwLock.writeLock().lock();
+ var newStorageRoots = new StorageRoots(newSuperRoot, newChangeId);
- try {
- storageRoots = new StorageRoots(newSuperRoot, newChangeId);
- } finally {
- rwLock.writeLock().unlock();
- }
+ rwLock.writeLock().lock();
+
+ try {
+ storageRoots = newStorageRoots;
+ } finally {
+ rwLock.writeLock().unlock();
+ }
+
+ // Save revisions for recovery.
+ return storage.writeConfigurationRevision(oldStorageRoots.version, newStorageRoots.version)
+ .thenCompose(unused -> {
+ long notificationNumber = notificationListenerCnt.incrementAndGet();
+
+ CompletableFuture<Void> notificationFuture;
- // Save revisions for recovery.
- return storage.writeConfigurationRevision(oldStorageRoots.version, storageRoots.version)
- .thenCompose(unused -> {
- long notificationNumber = notificationListenerCnt.incrementAndGet();
+ if (dataValuesPrefixMap.isEmpty()) {
+ notificationFuture = configurationUpdateListener.onRevisionUpdated(newChangeId, notificationNumber);
+ } else {
+ notificationFuture = configurationUpdateListener
+ .onConfigurationUpdated(oldSuperRoot, newSuperRoot, newChangeId, notificationNumber);
+ }
- return notificator.notify(oldSuperRoot, newSuperRoot, newChangeId, notificationNumber)
- .whenComplete((v, t) -> {
+ return notificationFuture.whenComplete((v, t) -> {
if (t == null) {
oldStorageRoots.changeFuture.complete(null);
} else {
oldStorageRoots.changeFuture.completeExceptionally(t);
}
});
- });
+ });
+ }
+
+ @Override
+ public CompletableFuture<Void> onRevisionUpdated(long newRevision) {
+ return configurationUpdateListener.onRevisionUpdated(newRevision, notificationListenerCnt.incrementAndGet());
+ }
+ };
}
/**
@@ -655,7 +684,9 @@ public abstract class ConfigurationChanger implements DynamicConfigurationChange
CompletableFuture<Void> notifyCurrentConfigurationListeners() {
StorageRoots storageRoots = this.storageRoots;
- return notificator.notify(null, storageRoots.roots, storageRoots.version, notificationListenerCnt.incrementAndGet());
+ long notificationCount = notificationListenerCnt.incrementAndGet();
+
+ return configurationUpdateListener.onConfigurationUpdated(null, storageRoots.roots, storageRoots.version, notificationCount);
}
/** {@inheritDoc} */
diff --git a/modules/configuration/src/main/java/org/apache/ignite/internal/configuration/ConfigurationRegistry.java b/modules/configuration/src/main/java/org/apache/ignite/internal/configuration/ConfigurationRegistry.java
index 3e446a1d2b..bff666bffa 100644
--- a/modules/configuration/src/main/java/org/apache/ignite/internal/configuration/ConfigurationRegistry.java
+++ b/modules/configuration/src/main/java/org/apache/ignite/internal/configuration/ConfigurationRegistry.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.configuration;
+import static java.util.concurrent.CompletableFuture.completedFuture;
import static java.util.function.Predicate.not;
import static java.util.stream.Collectors.toList;
import static org.apache.ignite.internal.configuration.notifications.ConfigurationNotifier.notifyListeners;
@@ -46,7 +47,6 @@ import java.util.Objects;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.function.Consumer;
-import java.util.function.Function;
import org.apache.ignite.configuration.ConfigurationTree;
import org.apache.ignite.configuration.RootKey;
import org.apache.ignite.configuration.annotation.Config;
@@ -58,6 +58,7 @@ import org.apache.ignite.configuration.notifications.ConfigurationListener;
import org.apache.ignite.configuration.notifications.ConfigurationNamedListListener;
import org.apache.ignite.configuration.notifications.ConfigurationNotificationEvent;
import org.apache.ignite.configuration.validation.Validator;
+import org.apache.ignite.internal.configuration.ConfigurationChanger.ConfigurationUpdateListener;
import org.apache.ignite.internal.configuration.asm.ConfigurationAsmGenerator;
import org.apache.ignite.internal.configuration.notifications.ConfigurationStorageRevisionListener;
import org.apache.ignite.internal.configuration.notifications.ConfigurationStorageRevisionListenerHolder;
@@ -139,8 +140,7 @@ public class ConfigurationRegistry implements IgniteComponent, ConfigurationStor
validators0.add(new PowerOfTwoValidator());
validators0.add(new RangeValidator());
- changer = new ConfigurationChanger(this::notificator, rootKeys, validators0, storage) {
- /** {@inheritDoc} */
+ changer = new ConfigurationChanger(notificationUpdateListener(), rootKeys, validators0, storage) {
@Override
public InnerNode createRootNode(RootKey<?, ?> rootKey) {
return cgen.instantiateNode(rootKey.schemaClass());
@@ -303,63 +303,64 @@ public class ConfigurationRegistry implements IgniteComponent, ConfigurationStor
});
}
- /**
- * Configuration change notifier.
- *
- * @param oldSuperRoot Old roots values. All these roots always belong to a single storage.
- * @param newSuperRoot New values for the same roots as in {@code oldRoot}.
- * @param storageRevision Revision of the storage.
- * @param notificationNumber Current configuration listener notification number.
- * @return Future that must signify when processing is completed.
- */
- private CompletableFuture<Void> notificator(
- @Nullable SuperRoot oldSuperRoot,
- SuperRoot newSuperRoot,
- long storageRevision,
- long notificationNumber
- ) {
- Collection<CompletableFuture<?>> futures = new ArrayList<>();
-
- newSuperRoot.traverseChildren(new ConfigurationVisitor<Void>() {
- /** {@inheritDoc} */
+ private ConfigurationUpdateListener notificationUpdateListener() {
+ return new ConfigurationUpdateListener() {
@Override
- public Void visitInnerNode(String key, InnerNode newRoot) {
- DynamicConfiguration<InnerNode, ?> config = (DynamicConfiguration<InnerNode, ?>) configs.get(key);
+ public CompletableFuture<Void> onConfigurationUpdated(
+ @Nullable SuperRoot oldSuperRoot, SuperRoot newSuperRoot, long storageRevision, long notificationNumber
+ ) {
+ var futures = new ArrayList<CompletableFuture<?>>();
- assert config != null : key;
+ newSuperRoot.traverseChildren(new ConfigurationVisitor<Void>() {
+ @Override
+ public Void visitInnerNode(String key, InnerNode newRoot) {
+ DynamicConfiguration<InnerNode, ?> config = (DynamicConfiguration<InnerNode, ?>) configs.get(key);
- InnerNode oldRoot;
+ assert config != null : key;
- if (oldSuperRoot != null) {
- oldRoot = oldSuperRoot.traverseChild(key, innerNodeVisitor(), true);
+ InnerNode oldRoot;
- assert oldRoot != null : key;
- } else {
- oldRoot = null;
- }
+ if (oldSuperRoot != null) {
+ oldRoot = oldSuperRoot.traverseChild(key, innerNodeVisitor(), true);
- futures.addAll(notifyListeners(oldRoot, newRoot, config, storageRevision, notificationNumber));
+ assert oldRoot != null : key;
+ } else {
+ oldRoot = null;
+ }
- return null;
- }
- }, true);
+ futures.addAll(notifyListeners(oldRoot, newRoot, config, storageRevision, notificationNumber));
- futures.addAll(notifyStorageRevisionListeners(storageRevision, notificationNumber));
+ return null;
+ }
+ }, true);
- if (futures.isEmpty()) {
- return CompletableFuture.completedFuture(null);
- }
+ futures.addAll(notifyStorageRevisionListeners(storageRevision, notificationNumber));
- // Map futures is only for logging errors.
- Function<CompletableFuture<?>, CompletableFuture<?>> mapping = fut -> fut.whenComplete((res, throwable) -> {
- if (throwable != null) {
- LOG.info("Failed to notify configuration listener", throwable);
+ return combineFutures(futures);
}
- });
- CompletableFuture<?>[] resultFutures = futures.stream().map(mapping).toArray(CompletableFuture[]::new);
+ @Override
+ public CompletableFuture<Void> onRevisionUpdated(long storageRevision, long notificationNumber) {
+ return combineFutures(notifyStorageRevisionListeners(storageRevision, notificationNumber));
+ }
- return CompletableFuture.allOf(resultFutures);
+ private CompletableFuture<Void> combineFutures(Collection<CompletableFuture<?>> futures) {
+ if (futures.isEmpty()) {
+ return completedFuture(null);
+ }
+
+ CompletableFuture<?>[] resultFutures = futures.stream()
+ // Map futures is only for logging errors.
+ .map(fut -> fut.whenComplete((res, throwable) -> {
+ if (throwable != null) {
+ LOG.info("Failed to notify configuration listener", throwable);
+ }
+ }))
+ .toArray(CompletableFuture[]::new);
+
+ return CompletableFuture.allOf(resultFutures);
+ }
+ };
}
/** {@inheritDoc} */
diff --git a/modules/configuration/src/main/java/org/apache/ignite/internal/configuration/storage/ConfigurationStorageListener.java b/modules/configuration/src/main/java/org/apache/ignite/internal/configuration/storage/ConfigurationStorageListener.java
index 928050e9c1..a32df38a25 100644
--- a/modules/configuration/src/main/java/org/apache/ignite/internal/configuration/storage/ConfigurationStorageListener.java
+++ b/modules/configuration/src/main/java/org/apache/ignite/internal/configuration/storage/ConfigurationStorageListener.java
@@ -22,7 +22,6 @@ import java.util.concurrent.CompletableFuture;
/**
* Configuration storage listener for changes.
*/
-@FunctionalInterface
public interface ConfigurationStorageListener {
/**
* Method called when entries in storage change.
@@ -31,4 +30,12 @@ public interface ConfigurationStorageListener {
* @return Completable future that signifies the completion of all custom user listeners execution.
*/
CompletableFuture<Void> onEntriesChanged(Data changedEntries);
+
+ /**
+ * Method called when a revision of the underlying storage changed, but no entries were changed.
+ *
+ * @param newRevision New storage revision.
+ * @return Completable future that signifies the completion of all custom user listeners execution.
+ */
+ CompletableFuture<Void> onRevisionUpdated(long newRevision);
}
diff --git a/modules/configuration/src/testFixtures/java/org/apache/ignite/internal/configuration/TestConfigurationChanger.java b/modules/configuration/src/testFixtures/java/org/apache/ignite/internal/configuration/TestConfigurationChanger.java
index 86c83ced78..4b72a70d2a 100644
--- a/modules/configuration/src/testFixtures/java/org/apache/ignite/internal/configuration/TestConfigurationChanger.java
+++ b/modules/configuration/src/testFixtures/java/org/apache/ignite/internal/configuration/TestConfigurationChanger.java
@@ -24,6 +24,7 @@ import static org.apache.ignite.internal.configuration.util.ConfigurationUtil.po
import java.util.Collection;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.CompletableFuture;
import org.apache.ignite.configuration.RootKey;
import org.apache.ignite.configuration.annotation.Config;
import org.apache.ignite.configuration.annotation.ConfigurationRoot;
@@ -33,6 +34,7 @@ import org.apache.ignite.configuration.validation.Validator;
import org.apache.ignite.internal.configuration.asm.ConfigurationAsmGenerator;
import org.apache.ignite.internal.configuration.storage.ConfigurationStorage;
import org.apache.ignite.internal.configuration.tree.InnerNode;
+import org.jetbrains.annotations.Nullable;
/** Implementation of {@link ConfigurationChanger} to be used in tests. Has no support of listeners. */
public class TestConfigurationChanger extends ConfigurationChanger {
@@ -59,12 +61,7 @@ public class TestConfigurationChanger extends ConfigurationChanger {
Collection<Class<?>> internalSchemaExtensions,
Collection<Class<?>> polymorphicSchemaExtensions
) {
- super(
- (oldRoot, newRoot, revision, notificationNumber) -> completedFuture(null),
- rootKeys,
- validators,
- storage
- );
+ super(noOpListener(), rootKeys, validators, storage);
this.cgen = cgen;
@@ -74,6 +71,21 @@ public class TestConfigurationChanger extends ConfigurationChanger {
rootKeys.forEach(key -> cgen.compileRootSchema(key.schemaClass(), internalExtensions, polymorphicExtensions));
}
+ private static ConfigurationUpdateListener noOpListener() {
+ return new ConfigurationUpdateListener() {
+ @Override
+ public CompletableFuture<Void> onConfigurationUpdated(@Nullable SuperRoot oldRoot, SuperRoot newRoot, long storageRevision,
+ long notificationNumber) {
+ return completedFuture(null);
+ }
+
+ @Override
+ public CompletableFuture<Void> onRevisionUpdated(long storageRevision, long notificationNumber) {
+ return completedFuture(null);
+ }
+ };
+ }
+
/** {@inheritDoc} */
@Override
public InnerNode createRootNode(RootKey<?, ?> rootKey) {
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/causality/VersionedValue.java b/modules/core/src/main/java/org/apache/ignite/internal/causality/VersionedValue.java
index 75225ee0e7..efca7c5fa1 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/causality/VersionedValue.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/causality/VersionedValue.java
@@ -21,8 +21,10 @@ import static java.util.concurrent.CompletableFuture.completedFuture;
import static java.util.concurrent.CompletableFuture.failedFuture;
import java.util.ArrayList;
+import java.util.Iterator;
import java.util.List;
import java.util.Map.Entry;
+import java.util.NavigableMap;
import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
@@ -30,8 +32,6 @@ import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicReference;
-import java.util.concurrent.locks.ReadWriteLock;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.function.Function;
@@ -39,6 +39,7 @@ import java.util.function.Supplier;
import org.apache.ignite.lang.IgniteInternalException;
import org.apache.ignite.lang.IgniteStringFormatter;
import org.apache.ignite.lang.IgniteTriConsumer;
+import org.jetbrains.annotations.Nullable;
/**
* Parametrized type to store several versions of the value.
@@ -51,10 +52,10 @@ public class VersionedValue<T> {
private static final long NOT_INITIALIZED = -1L;
/** Default history size. */
- private static final int DEFAULT_HISTORY_SIZE = 2;
+ private static final int DEFAULT_MAX_HISTORY_SIZE = 10;
- /** Size of stored history. */
- private final int historySize;
+ /** Size of the history of changes to store, including last applied token. */
+ private final int maxHistorySize;
/** List of completion listeners, see {@link #whenComplete(IgniteTriConsumer)}. */
private final List<IgniteTriConsumer<Long, T, Throwable>> completionListeners = new CopyOnWriteArrayList<>();
@@ -63,15 +64,15 @@ public class VersionedValue<T> {
private final ConcurrentNavigableMap<Long, CompletableFuture<T>> history = new ConcurrentSkipListMap<>();
/**
- * This lock guarantees that the history is not trimming {@link #trimToSize(long)} during getting a value from versioned storage {@link
- * #get(long)}.
+ * Mutex that ensures that only one thread can be removing old history entries.
*/
- private final ReadWriteLock trimHistoryLock = new ReentrantReadWriteLock();
+ private final Object trimHistoryMutex = new Object();
/** Initial future. The future will be completed when {@link VersionedValue} sets a first value. */
private final CompletableFuture<T> initFut = new CompletableFuture<>();
/** The supplier may provide a value which will used as a default. */
+ @Nullable
private final Supplier<T> defaultValSupplier;
/** Update mutex. */
@@ -88,6 +89,7 @@ public class VersionedValue<T> {
* This {@code updaterFuture} is {@code null} if no updates in context of current causality token have been initiated.
* See {@link #update(long, BiFunction)}.
*/
+ @Nullable
private volatile CompletableFuture<T> updaterFuture = null;
/**
@@ -99,17 +101,17 @@ public class VersionedValue<T> {
* {@code Function<Long, CompletableFuture<?>>} that should be called on every update of
* storage revision as a listener. IMPORTANT: Revision update shouldn't happen
* concurrently with {@link #complete(long, T)} operations.
- * @param historySize Size of the history of changes to store, including last applied token.
+ * @param maxHistorySize Size of the history of changes to store, including last applied token.
* @param defaultVal Supplier of the default value, that is used on {@link #update(long, BiFunction)} to
* evaluate the default value if the value is not initialized yet. It is not guaranteed to
* execute only once.
*/
public VersionedValue(
- Consumer<Function<Long, CompletableFuture<?>>> observableRevisionUpdater,
- int historySize,
- Supplier<T> defaultVal
+ @Nullable Consumer<Function<Long, CompletableFuture<?>>> observableRevisionUpdater,
+ int maxHistorySize,
+ @Nullable Supplier<T> defaultVal
) {
- this.historySize = historySize;
+ this.maxHistorySize = maxHistorySize;
this.defaultValSupplier = defaultVal;
@@ -134,10 +136,10 @@ public class VersionedValue<T> {
* execute only once.
*/
public VersionedValue(
- Consumer<Function<Long, CompletableFuture<?>>> observableRevisionUpdater,
- Supplier<T> defaultVal
+ @Nullable Consumer<Function<Long, CompletableFuture<?>>> observableRevisionUpdater,
+ @Nullable Supplier<T> defaultVal
) {
- this(observableRevisionUpdater, DEFAULT_HISTORY_SIZE, defaultVal);
+ this(observableRevisionUpdater, DEFAULT_MAX_HISTORY_SIZE, defaultVal);
}
/**
@@ -150,8 +152,8 @@ public class VersionedValue<T> {
* storage revision as a listener. IMPORTANT: Revision update shouldn't happen
* concurrently with {@link #complete(long, T)} operations.
*/
- public VersionedValue(Consumer<Function<Long, CompletableFuture<?>>> observableRevisionUpdater) {
- this(observableRevisionUpdater, DEFAULT_HISTORY_SIZE, null);
+ public VersionedValue(@Nullable Consumer<Function<Long, CompletableFuture<?>>> observableRevisionUpdater) {
+ this(observableRevisionUpdater, DEFAULT_MAX_HISTORY_SIZE, null);
}
/**
@@ -187,34 +189,20 @@ public class VersionedValue<T> {
long actualToken0 = this.actualToken;
if (history.floorEntry(causalityToken) == null) {
- throw new OutdatedTokenException(causalityToken, actualToken0, historySize);
+ throw new OutdatedTokenException(causalityToken, actualToken0, maxHistorySize);
}
if (causalityToken <= actualToken0) {
return getValueForPreviousToken(causalityToken);
}
- trimHistoryLock.readLock().lock();
-
- try {
- if (causalityToken <= actualToken0) {
- return getValueForPreviousToken(causalityToken);
- }
-
- var fut = new CompletableFuture<T>();
-
- CompletableFuture<T> previousFut = history.putIfAbsent(causalityToken, fut);
-
- return previousFut == null ? fut : previousFut;
- } finally {
- trimHistoryLock.readLock().unlock();
- }
+ return history.computeIfAbsent(causalityToken, t -> new CompletableFuture<>());
}
/**
* Gets the latest value of completed future.
*/
- public T latest() {
+ public @Nullable T latest() {
for (CompletableFuture<T> fut : history.descendingMap().values()) {
if (fut.isDone()) {
return fut.join();
@@ -229,7 +217,7 @@ public class VersionedValue<T> {
*
* @return The value.
*/
- private T getDefault() {
+ private @Nullable T getDefault() {
if (defaultValSupplier != null && defaultValRef.get() == null) {
T defaultVal = defaultValSupplier.get();
@@ -252,7 +240,7 @@ public class VersionedValue<T> {
Entry<Long, CompletableFuture<T>> histEntry = history.floorEntry(causalityToken);
if (histEntry == null) {
- throw new OutdatedTokenException(causalityToken, actualToken, historySize);
+ throw new OutdatedTokenException(causalityToken, actualToken, maxHistorySize);
}
return histEntry.getValue();
@@ -317,7 +305,7 @@ public class VersionedValue<T> {
* @param value Value to set.
* @param throwable An exception.
*/
- private void completeInternal(long causalityToken, T value, Throwable throwable) {
+ private void completeInternal(long causalityToken, @Nullable T value, @Nullable Throwable throwable) {
CompletableFuture<T> res = history.putIfAbsent(
causalityToken,
throwable == null ? completedFuture(value) : failedFuture(throwable)
@@ -350,7 +338,9 @@ public class VersionedValue<T> {
* @param throwable Throwable.
* @return Error message.
*/
- private String completeInternalConflictErrorMessage(CompletableFuture<T> future, long token, T value, Throwable throwable) {
+ private String completeInternalConflictErrorMessage(
+ CompletableFuture<T> future, long token, @Nullable T value, @Nullable Throwable throwable
+ ) {
return future.handle(
(prevValue, prevThrowable) ->
IgniteStringFormatter.format(
@@ -408,7 +398,7 @@ public class VersionedValue<T> {
CompletableFuture<T> future = updaterFuture == null ? previousOrDefaultValueFuture(actualToken0) : updaterFuture;
CompletableFuture<CompletableFuture<T>> f0 = future
- .handle(updater::apply)
+ .handle(updater)
.handle((fut, e) -> e == null ? fut : failedFuture(e));
updaterFuture = f0.thenCompose(Function.identity());
@@ -446,7 +436,7 @@ public class VersionedValue<T> {
* @param value Value.
* @param throwable Throwable.
*/
- private void notifyCompletionListeners(long causalityToken, T value, Throwable throwable) {
+ private void notifyCompletionListeners(long causalityToken, @Nullable T value, @Nullable Throwable throwable) {
Throwable unpackedThrowable = throwable instanceof CompletionException ? throwable.getCause() : throwable;
List<Exception> exceptions = new ArrayList<>();
@@ -499,14 +489,12 @@ public class VersionedValue<T> {
return completeUpdatesFuture.thenRun(() -> {
completeRelatedFuture(causalityToken);
- if (history.size() > 1 && causalityToken - history.firstKey() >= historySize) {
- trimToSize(causalityToken);
- }
-
- Entry<Long, CompletableFuture<T>> entry = history.floorEntry(causalityToken);
+ // Delete older tokens if their amount exceeds the configured history size.
+ NavigableMap<Long, CompletableFuture<T>> oldTokensMap = history.headMap(causalityToken, true);
- assert entry != null && entry.getValue().isDone() : IgniteStringFormatter.format(
- "Future for the token is not completed [token={}]", causalityToken);
+ if (oldTokensMap.size() > maxHistorySize) {
+ trimToSize(oldTokensMap);
+ }
});
}
}
@@ -522,7 +510,7 @@ public class VersionedValue<T> {
CompletableFuture<T> future = entry.getValue();
if (!future.isDone()) {
- Entry<Long, CompletableFuture<T>> entryBefore = history.headMap(causalityToken).lastEntry();
+ Entry<Long, CompletableFuture<T>> entryBefore = history.lowerEntry(causalityToken);
CompletableFuture<T> previousFuture = entryBefore == null ? completedFuture(getDefault()) : entryBefore.getValue();
@@ -544,6 +532,8 @@ public class VersionedValue<T> {
// This future is previous, it is always done.
future.whenComplete((v, e) -> notifyCompletionListeners(causalityToken, v, e));
}
+
+ assert future.isDone() : IgniteStringFormatter.format("Future for the token is not completed [token={}]", causalityToken);
}
/**
@@ -568,22 +558,15 @@ public class VersionedValue<T> {
/**
* Trims the storage to history size.
- *
- * @param causalityToken Last token which is being applied.
*/
- private void trimToSize(long causalityToken) {
- Long lastToken = history.lastKey();
-
- trimHistoryLock.writeLock().lock();
+ private void trimToSize(NavigableMap<Long, CompletableFuture<T>> map) {
+ synchronized (trimHistoryMutex) {
+ Iterator<Long> it = map.keySet().iterator();
- try {
- for (Long token : history.keySet()) {
- if (!token.equals(lastToken) && causalityToken - token >= historySize) {
- history.remove(token);
- }
+ for (int i = map.size(); i > maxHistorySize; i--) {
+ it.next();
+ it.remove();
}
- } finally {
- trimHistoryLock.writeLock().unlock();
}
}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/causality/VersionedValueTest.java b/modules/core/src/test/java/org/apache/ignite/internal/causality/VersionedValueTest.java
index f12ca62248..9457722cb5 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/causality/VersionedValueTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/causality/VersionedValueTest.java
@@ -22,6 +22,7 @@ import static java.util.concurrent.CompletableFuture.failedFuture;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.assertThrowsWithCause;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.containsString;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
@@ -54,13 +55,13 @@ import org.junit.jupiter.api.Test;
*/
public class VersionedValueTest {
/** Test value. */
- public static final int TEST_VALUE = 1;
+ private static final int TEST_VALUE = 1;
/** Test exception is used for exceptionally completion Versioned value object. */
- public static final Exception TEST_EXCEPTION = new Exception("Test exception.");
+ private static final Exception TEST_EXCEPTION = new Exception("Test exception.");
/** The test revision register is used to move the revision forward. */
- public static final TestRevisionRegister REGISTER = new TestRevisionRegister();
+ private static final TestRevisionRegister REGISTER = new TestRevisionRegister();
@BeforeEach
public void clearRegister() {
@@ -157,7 +158,7 @@ public class VersionedValueTest {
/**
* The test reads a value with the specific token in which the value should not be updated.
- * The read happenes before the revision updated.
+ * The read happens before the revision updated.
*
* @throws OutdatedTokenException If failed.
*/
@@ -208,14 +209,34 @@ public class VersionedValueTest {
*/
@Test
public void testObsoleteToken() {
- VersionedValue<Integer> longVersionedValue = new VersionedValue<>(null);
+ VersionedValue<Integer> versionedValue = new VersionedValue<>(null, 2, null);
+
+ versionedValue.complete(0, TEST_VALUE);
+ versionedValue.complete(1, TEST_VALUE);
+ // Internal map size is now 3: one actual and two old tokens. One old token must be removed.
+ versionedValue.complete(2, TEST_VALUE);
+
+ assertThrowsExactly(OutdatedTokenException.class, () -> versionedValue.get(0));
+ }
+
+ /**
+ * Tests that tokens that are newer then the current token do not contribute to the history size calculation.
+ */
+ @Test
+ public void testNewTokensNotGetRemoved() {
+ VersionedValue<Integer> longVersionedValue = new VersionedValue<>(null, 2, null);
longVersionedValue.complete(0, TEST_VALUE);
longVersionedValue.complete(1, TEST_VALUE);
- longVersionedValue.complete(2);
+ // History size is now 2, try adding a token from the future.
+ assertDoesNotThrow(() -> longVersionedValue.get(10));
+
+ // Internal map size is now 3, this should trigger history trimming.
+ longVersionedValue.complete(2, TEST_VALUE);
assertThrowsExactly(OutdatedTokenException.class, () -> longVersionedValue.get(0));
+ assertDoesNotThrow(() -> longVersionedValue.get(10));
}
/**
@@ -473,7 +494,7 @@ public class VersionedValueTest {
/**
* Tests a case when there is no default value supplier.
*/
- public void checkDefaultValue(VersionedValue<Integer> vv, Integer expectedDefault) {
+ private static void checkDefaultValue(VersionedValue<Integer> vv, Integer expectedDefault) {
assertEquals(expectedDefault, vv.latest());
vv.update(0, (a, e) -> {
@@ -597,7 +618,7 @@ public class VersionedValueTest {
/**
* Clear list.
*/
- public void clear() {
+ void clear() {
moveRevisionList.clear();
}
@@ -607,7 +628,7 @@ public class VersionedValueTest {
* @param revision Revision.
* @return Future for all listeners.
*/
- public CompletableFuture<?> moveRevision(long revision) {
+ CompletableFuture<?> moveRevision(long revision) {
List<CompletableFuture<?>> futures = new ArrayList<>();
moveRevisionList.forEach(m -> futures.add(m.apply(revision)));
diff --git a/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/DistributionZoneManager.java b/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/DistributionZoneManager.java
index d43177020b..35b3db88e7 100644
--- a/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/DistributionZoneManager.java
+++ b/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/DistributionZoneManager.java
@@ -120,6 +120,8 @@ public class DistributionZoneManager implements IgniteComponent {
private static final String DISTRIBUTION_ZONE_MANAGER_POOL_NAME = "dst-zones-scheduler";
+ private static final String META_STORAGE_WATCH_ID = "dst-zones-watch";
+
/** Id of the default distribution zone. */
public static final int DEFAULT_ZONE_ID = 0;
@@ -862,10 +864,8 @@ public class DistributionZoneManager implements IgniteComponent {
}
try {
- long appliedRevision = metaStorageManager.appliedRevision();
-
vaultMgr.get(zonesLogicalTopologyKey())
- .thenAccept(vaultEntry -> {
+ .thenAcceptBoth(metaStorageManager.appliedRevision(META_STORAGE_WATCH_ID), (vaultEntry, appliedRevision) -> {
if (!busyLock.enterBusy()) {
throw new IgniteInternalException(NODE_STOPPING_ERR, new NodeStoppingException());
}
@@ -904,9 +904,14 @@ public class DistributionZoneManager implements IgniteComponent {
private WatchListener createMetastorageListener() {
return new WatchListener() {
@Override
- public void onUpdate(WatchEvent evt) {
+ public String id() {
+ return META_STORAGE_WATCH_ID;
+ }
+
+ @Override
+ public CompletableFuture<Void> onUpdate(WatchEvent evt) {
if (!busyLock.enterBusy()) {
- throw new IgniteInternalException(NODE_STOPPING_ERR, new NodeStoppingException());
+ return failedFuture(new NodeStoppingException());
}
try {
@@ -942,6 +947,8 @@ public class DistributionZoneManager implements IgniteComponent {
DistributionZoneView defaultZoneView = zonesConfiguration.value().defaultDistributionZone();
scheduleTimers(defaultZoneView, addedNodes, removedNodes, revision);
+
+ return completedFuture(null);
} finally {
busyLock.leaveBusy();
}
diff --git a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerConfigurationChangesTest.java b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerConfigurationChangesTest.java
index 521dab900f..4fa103ab21 100644
--- a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerConfigurationChangesTest.java
+++ b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerConfigurationChangesTest.java
@@ -51,7 +51,7 @@ import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
import org.apache.ignite.configuration.NamedConfigurationTree;
import org.apache.ignite.configuration.NamedListView;
-import org.apache.ignite.internal.cluster.management.topology.LogicalTopologyServiceImpl;
+import org.apache.ignite.internal.cluster.management.topology.api.LogicalTopologyService;
import org.apache.ignite.internal.cluster.management.topology.api.LogicalTopologySnapshot;
import org.apache.ignite.internal.configuration.ConfigurationManager;
import org.apache.ignite.internal.configuration.storage.TestConfigurationStorage;
@@ -83,7 +83,6 @@ import org.jetbrains.annotations.Nullable;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
-import org.mockito.Mock;
/**
* Tests distribution zones configuration changes and reaction to that changes.
@@ -101,9 +100,6 @@ public class DistributionZoneManagerConfigurationChangesTest extends IgniteAbstr
private ConfigurationManager clusterCfgMgr;
- @Mock
- private LogicalTopologyServiceImpl logicalTopologyService;
-
private VaultManager vaultMgr;
@BeforeEach
@@ -121,7 +117,9 @@ public class DistributionZoneManagerConfigurationChangesTest extends IgniteAbstr
MetaStorageManager metaStorageManager = mock(MetaStorageManager.class);
- logicalTopologyService = mock(LogicalTopologyServiceImpl.class);
+ when(metaStorageManager.appliedRevision(any())).thenReturn(completedFuture(0L));
+
+ LogicalTopologyService logicalTopologyService = mock(LogicalTopologyService.class);
vaultMgr = mock(VaultManager.class);
diff --git a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerLogicalTopologyEventsTest.java b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerLogicalTopologyEventsTest.java
index d0b57d3cf3..82185aa6bd 100644
--- a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerLogicalTopologyEventsTest.java
+++ b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerLogicalTopologyEventsTest.java
@@ -77,13 +77,11 @@ import org.apache.ignite.network.NetworkAddress;
import org.jetbrains.annotations.Nullable;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
-import org.mockito.Mock;
/**
* Tests reactions to topology changes in accordance with distribution zones logic.
*/
public class DistributionZoneManagerLogicalTopologyEventsTest {
- @Mock
private ClusterManagementGroupManager cmgManager;
private DistributionZoneManager distributionZoneManager;
@@ -110,6 +108,8 @@ public class DistributionZoneManagerLogicalTopologyEventsTest {
MetaStorageManager metaStorageManager = mock(MetaStorageManager.class);
+ when(metaStorageManager.appliedRevision(any())).thenReturn(completedFuture(0L));
+
cmgManager = mock(ClusterManagementGroupManager.class);
clusterStateStorage = new TestClusterStateStorage();
diff --git a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerScaleUpTest.java b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerScaleUpTest.java
index 624f9617e7..2a03fcac9f 100644
--- a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerScaleUpTest.java
+++ b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerScaleUpTest.java
@@ -109,9 +109,6 @@ public class DistributionZoneManagerScaleUpTest {
private ConfigurationManager clusterCfgMgr;
- @Mock
- private LogicalTopologyServiceImpl logicalTopologyService;
-
private LogicalTopology topology;
private ClusterStateStorage clusterStateStorage;
@@ -1573,7 +1570,7 @@ public class DistributionZoneManagerScaleUpTest {
}
private void mockVaultAppliedRevision(long revision) {
- when(metaStorageManager.appliedRevision()).thenReturn(revision);
+ when(metaStorageManager.appliedRevision(any())).thenReturn(completedFuture(revision));
}
private void watchListenerOnUpdate(Set<String> nodes, long rev) {
diff --git a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerWatchListenerTest.java b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerWatchListenerTest.java
index 16413cf9ac..84f1a807e6 100644
--- a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerWatchListenerTest.java
+++ b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerWatchListenerTest.java
@@ -73,7 +73,6 @@ import org.apache.ignite.internal.vault.VaultManager;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
-import org.mockito.Mock;
/**
* Tests distribution zones logical topology changes and reaction to that changes.
@@ -86,9 +85,6 @@ public class DistributionZoneManagerWatchListenerTest extends IgniteAbstractTest
private ConfigurationManager clusterCfgMgr;
- @Mock
- private LogicalTopologyServiceImpl logicalTopologyService;
-
private LogicalTopology topology;
private ClusterStateStorage clusterStateStorage;
@@ -101,11 +97,9 @@ public class DistributionZoneManagerWatchListenerTest extends IgniteAbstractTest
private DistributionZonesConfiguration zonesConfiguration;
- @Mock
private ClusterManagementGroupManager cmgManager;
- @Mock
- RaftGroupService metaStorageService;
+ private RaftGroupService metaStorageService;
@BeforeEach
public void setUp() {
@@ -339,7 +333,7 @@ public class DistributionZoneManagerWatchListenerTest extends IgniteAbstractTest
}
private void mockVaultAppliedRevision(long revision) {
- when(metaStorageManager.appliedRevision()).thenReturn(revision);
+ when(metaStorageManager.appliedRevision(any())).thenReturn(completedFuture(revision));
}
private void mockCmgLocalNodes() {
diff --git a/modules/metastorage-api/src/main/java/org/apache/ignite/internal/metastorage/MetaStorageManager.java b/modules/metastorage-api/src/main/java/org/apache/ignite/internal/metastorage/MetaStorageManager.java
index 018c1596c8..f4fa35bb2c 100644
--- a/modules/metastorage-api/src/main/java/org/apache/ignite/internal/metastorage/MetaStorageManager.java
+++ b/modules/metastorage-api/src/main/java/org/apache/ignite/internal/metastorage/MetaStorageManager.java
@@ -40,9 +40,11 @@ import org.jetbrains.annotations.Nullable;
public interface MetaStorageManager extends IgniteComponent {
/**
* Returns the current <em>applied revision</em> of the Meta Storage, that is, the most recent revision of updates that have been
- * applied on this node.
+ * processed by a particular Watch on this node.
+ *
+ * @param watchId ID of the watch that this revision corresponds to.
*/
- long appliedRevision();
+ CompletableFuture<Long> appliedRevision(String watchId);
/**
* Retrieves an entry for the given key.
diff --git a/modules/metastorage-api/src/main/java/org/apache/ignite/internal/metastorage/WatchEvent.java b/modules/metastorage-api/src/main/java/org/apache/ignite/internal/metastorage/WatchEvent.java
index 6a2b19d592..8dc49fcd98 100644
--- a/modules/metastorage-api/src/main/java/org/apache/ignite/internal/metastorage/WatchEvent.java
+++ b/modules/metastorage-api/src/main/java/org/apache/ignite/internal/metastorage/WatchEvent.java
@@ -19,6 +19,7 @@ package org.apache.ignite.internal.metastorage;
import java.util.Collection;
import java.util.List;
+import org.apache.ignite.internal.tostring.IgniteToStringInclude;
import org.apache.ignite.internal.tostring.S;
/**
@@ -27,6 +28,7 @@ import org.apache.ignite.internal.tostring.S;
*/
public class WatchEvent {
/** Events about each entry update in the revision. */
+ @IgniteToStringInclude
private final List<EntryEvent> entryEvts;
private final long revision;
@@ -38,8 +40,6 @@ public class WatchEvent {
* @param revision Revision of the updated entries.
*/
public WatchEvent(List<EntryEvent> entryEvts, long revision) {
- assert entryEvts != null && !entryEvts.isEmpty();
-
this.entryEvts = entryEvts;
this.revision = revision;
}
@@ -89,6 +89,30 @@ public class WatchEvent {
return revision;
}
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ WatchEvent event = (WatchEvent) o;
+
+ if (revision != event.revision) {
+ return false;
+ }
+ return entryEvts.equals(event.entryEvts);
+ }
+
+ @Override
+ public int hashCode() {
+ int result = entryEvts.hashCode();
+ result = 31 * result + (int) (revision ^ (revision >>> 32));
+ return result;
+ }
+
@Override
public String toString() {
return S.toString(this);
diff --git a/modules/metastorage-api/src/main/java/org/apache/ignite/internal/metastorage/WatchListener.java b/modules/metastorage-api/src/main/java/org/apache/ignite/internal/metastorage/WatchListener.java
index a8a499b4c6..665ac765a0 100644
--- a/modules/metastorage-api/src/main/java/org/apache/ignite/internal/metastorage/WatchListener.java
+++ b/modules/metastorage-api/src/main/java/org/apache/ignite/internal/metastorage/WatchListener.java
@@ -17,16 +17,35 @@
package org.apache.ignite.internal.metastorage;
+import java.util.concurrent.CompletableFuture;
+
/**
* The listener which receives and handles watch updates.
*/
public interface WatchListener {
+ /**
+ * Returns a unique identifier for this Watch. This identifier should never change between node restarts and must uniquely identify
+ * a Watch among all Watches on a local node.
+ */
+ String id();
+
/**
* The method will be called on each meta storage update.
*
* @param event A single event or a batch. The batch always contains updates for specific revision.
+ * @return Future that will be completed when the event is processed.
+ */
+ CompletableFuture<Void> onUpdate(WatchEvent event);
+
+ /**
+ * Callback that will be invoked if a Meta Storage update has been received, but the modified entries do not match the given Watch.
+ *
+ * @param revision Meta Storage revision.
+ * @return Future that will be completed when the event is processed.
*/
- void onUpdate(WatchEvent event);
+ default CompletableFuture<Void> onRevisionUpdated(long revision) {
+ return CompletableFuture.completedFuture(null);
+ }
/**
* The method will be called in case of an error occurred. The listener and corresponding watch will be unregistered.
diff --git a/modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/impl/ItMetaStorageManagerImplTest.java b/modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/impl/ItMetaStorageManagerImplTest.java
index c2411ee654..e07995cb97 100644
--- a/modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/impl/ItMetaStorageManagerImplTest.java
+++ b/modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/impl/ItMetaStorageManagerImplTest.java
@@ -188,7 +188,7 @@ public class ItMetaStorageManagerImplTest extends IgniteAbstractTest {
assertThat(vaultManager.get(key1), willBe(nullValue()));
assertThat(vaultManager.get(key2), willBe(nullValue()));
- metaStorageManager.registerExactWatch(key1, noOpListener());
+ metaStorageManager.registerExactWatch(key1, new NoOpListener("test1"));
invokeFuture = metaStorageManager.invoke(
Conditions.exists(new ByteArray("foo")),
@@ -201,13 +201,16 @@ public class ItMetaStorageManagerImplTest extends IgniteAbstractTest {
assertThat(invokeFuture, willBe(true));
- assertTrue(waitForCondition(() -> metaStorageManager.appliedRevision() == 2, 10_000));
+ assertTrue(waitForCondition(() -> metaStorageManager.appliedRevision("test1").join() == 2, 10_000));
// Expect that only the watched key is persisted.
assertThat(vaultManager.get(key1).thenApply(VaultEntry::value), willBe(value));
assertThat(vaultManager.get(key2), willBe(nullValue()));
- metaStorageManager.registerExactWatch(key2, noOpListener());
+ metaStorageManager.registerExactWatch(key2, new NoOpListener("test2"));
+
+ assertThat(metaStorageManager.appliedRevision("test1"), willBe(2L));
+ assertThat(metaStorageManager.appliedRevision("test2"), willBe(0L));
byte[] newValue = "newValue".getBytes(StandardCharsets.UTF_8);
@@ -222,20 +225,32 @@ public class ItMetaStorageManagerImplTest extends IgniteAbstractTest {
assertThat(invokeFuture, willBe(true));
- assertTrue(waitForCondition(() -> metaStorageManager.appliedRevision() == 3, 10_000));
+ assertTrue(waitForCondition(() -> metaStorageManager.appliedRevision("test1").join() == 3, 10_000));
+ assertTrue(waitForCondition(() -> metaStorageManager.appliedRevision("test2").join() == 3, 10_000));
assertThat(vaultManager.get(key1).thenApply(VaultEntry::value), willBe(newValue));
assertThat(vaultManager.get(key2).thenApply(VaultEntry::value), willBe(newValue));
}
- private static WatchListener noOpListener() {
- return new WatchListener() {
- @Override
- public void onUpdate(WatchEvent event) {}
+ private static class NoOpListener implements WatchListener {
+ private final String id;
+
+ NoOpListener(String id) {
+ this.id = id;
+ }
+
+ @Override
+ public String id() {
+ return id;
+ }
+
+ @Override
+ public CompletableFuture<Void> onUpdate(WatchEvent event) {
+ return completedFuture(null);
+ }
- @Override
- public void onError(Throwable e) {}
- };
+ @Override
+ public void onError(Throwable e) {}
}
@Test
diff --git a/modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/impl/ItMetaStorageMultipleNodesTest.java b/modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/impl/ItMetaStorageMultipleNodesTest.java
index 9dec1c3974..9a90c3d8cf 100644
--- a/modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/impl/ItMetaStorageMultipleNodesTest.java
+++ b/modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/impl/ItMetaStorageMultipleNodesTest.java
@@ -18,6 +18,7 @@
package org.apache.ignite.internal.metastorage.impl;
import static java.util.concurrent.CompletableFuture.allOf;
+import static java.util.concurrent.CompletableFuture.completedFuture;
import static java.util.stream.Collectors.toSet;
import static org.apache.ignite.internal.metastorage.dsl.Conditions.notExists;
import static org.apache.ignite.internal.metastorage.dsl.Conditions.revision;
@@ -249,12 +250,19 @@ public class ItMetaStorageMultipleNodesTest extends IgniteAbstractTest {
secondNode.metaStorageManager.registerExactWatch(key, new WatchListener() {
@Override
- public void onUpdate(WatchEvent event) {
+ public String id() {
+ return "test";
+ }
+
+ @Override
+ public CompletableFuture<Void> onUpdate(WatchEvent event) {
// Skip the first update event, because it's not guaranteed to arrive here (insert may have happened before the watch was
// registered).
if (event.revision() != 1) {
awaitFuture.complete(event.entryEvent());
}
+
+ return completedFuture(null);
}
@Override
diff --git a/modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/impl/ItMetaStorageWatchTest.java b/modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/impl/ItMetaStorageWatchTest.java
index bdadfcbafa..e5e758f862 100644
--- a/modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/impl/ItMetaStorageWatchTest.java
+++ b/modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/impl/ItMetaStorageWatchTest.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.metastorage.impl;
+import static java.util.concurrent.CompletableFuture.completedFuture;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willBe;
import static org.apache.ignite.utils.ClusterServiceTestUtils.findLocalAddresses;
import static org.hamcrest.MatcherAssert.assertThat;
@@ -205,11 +206,18 @@ public class ItMetaStorageWatchTest extends IgniteAbstractTest {
void testExactWatch() throws Exception {
testWatches((node, latch) -> node.metaStorageManager.registerExactWatch(new ByteArray("foo"), new WatchListener() {
@Override
- public void onUpdate(WatchEvent event) {
+ public String id() {
+ return "test";
+ }
+
+ @Override
+ public CompletableFuture<Void> onUpdate(WatchEvent event) {
assertThat(event.entryEvent().newEntry().key(), is("foo".getBytes(StandardCharsets.UTF_8)));
assertThat(event.entryEvent().newEntry().value(), is("bar".getBytes(StandardCharsets.UTF_8)));
latch.countDown();
+
+ return completedFuture(null);
}
@Override
@@ -223,11 +231,18 @@ public class ItMetaStorageWatchTest extends IgniteAbstractTest {
void testPrefixWatch() throws Exception {
testWatches((node, latch) -> node.metaStorageManager.registerPrefixWatch(new ByteArray("fo"), new WatchListener() {
@Override
- public void onUpdate(WatchEvent event) {
+ public String id() {
+ return "test";
+ }
+
+ @Override
+ public CompletableFuture<Void> onUpdate(WatchEvent event) {
assertThat(event.entryEvent().newEntry().key(), is("foo".getBytes(StandardCharsets.UTF_8)));
assertThat(event.entryEvent().newEntry().value(), is("bar".getBytes(StandardCharsets.UTF_8)));
latch.countDown();
+
+ return completedFuture(null);
}
@Override
@@ -245,11 +260,18 @@ public class ItMetaStorageWatchTest extends IgniteAbstractTest {
node.metaStorageManager.registerRangeWatch(startRange, endRange, new WatchListener() {
@Override
- public void onUpdate(WatchEvent event) {
+ public String id() {
+ return "test";
+ }
+
+ @Override
+ public CompletableFuture<Void> onUpdate(WatchEvent event) {
assertThat(event.entryEvent().newEntry().key(), is("foo".getBytes(StandardCharsets.UTF_8)));
assertThat(event.entryEvent().newEntry().value(), is("bar".getBytes(StandardCharsets.UTF_8)));
latch.countDown();
+
+ return completedFuture(null);
}
@Override
@@ -301,11 +323,18 @@ public class ItMetaStorageWatchTest extends IgniteAbstractTest {
for (Node node : nodes) {
node.metaStorageManager.registerExactWatch(new ByteArray("foo"), new WatchListener() {
@Override
- public void onUpdate(WatchEvent event) {
+ public String id() {
+ return "test1";
+ }
+
+ @Override
+ public CompletableFuture<Void> onUpdate(WatchEvent event) {
assertThat(event.entryEvent().newEntry().key(), is("foo".getBytes(StandardCharsets.UTF_8)));
assertThat(event.entryEvent().newEntry().value(), is("bar".getBytes(StandardCharsets.UTF_8)));
exactLatch.countDown();
+
+ return completedFuture(null);
}
@Override
@@ -316,7 +345,12 @@ public class ItMetaStorageWatchTest extends IgniteAbstractTest {
node.metaStorageManager.registerPrefixWatch(new ByteArray("ba"), new WatchListener() {
@Override
- public void onUpdate(WatchEvent event) {
+ public String id() {
+ return "test2";
+ }
+
+ @Override
+ public CompletableFuture<Void> onUpdate(WatchEvent event) {
List<String> keys = event.entryEvents().stream()
.map(e -> new String(e.newEntry().key(), StandardCharsets.UTF_8))
.collect(Collectors.toList());
@@ -329,6 +363,8 @@ public class ItMetaStorageWatchTest extends IgniteAbstractTest {
assertThat(values, containsInAnyOrder("one", "two"));
prefixLatch.countDown();
+
+ return completedFuture(null);
}
@Override
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/command/MetastorageCommandsMessageGroup.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/command/MetastorageCommandsMessageGroup.java
index 13df10e7fc..8856166a9d 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/command/MetastorageCommandsMessageGroup.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/command/MetastorageCommandsMessageGroup.java
@@ -55,6 +55,9 @@ public interface MetastorageCommandsMessageGroup {
/** Message type for {@link GetAndRemoveAllCommand}. */
short GET_AND_REMOVE_ALL = 32;
+ /** Message type for {@link GetRevisionCommand}. */
+ short GET_REVISION = 33;
+
/** Message type for {@link PutCommand}. */
short PUT = 40;
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/impl/MetaStorageManagerImpl.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/impl/MetaStorageManagerImpl.java
index 897444d14c..23452efed6 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/impl/MetaStorageManagerImpl.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/impl/MetaStorageManagerImpl.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.metastorage.impl;
+import static java.util.concurrent.CompletableFuture.completedFuture;
import static org.apache.ignite.internal.util.ByteUtils.bytesToLong;
import static org.apache.ignite.internal.util.ByteUtils.longToBytes;
import static org.apache.ignite.internal.util.IgniteUtils.cancelOrConsume;
@@ -34,13 +35,12 @@ import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.internal.metastorage.Entry;
import org.apache.ignite.internal.metastorage.MetaStorageManager;
+import org.apache.ignite.internal.metastorage.WatchEvent;
import org.apache.ignite.internal.metastorage.WatchListener;
import org.apache.ignite.internal.metastorage.dsl.Condition;
import org.apache.ignite.internal.metastorage.dsl.Iif;
import org.apache.ignite.internal.metastorage.dsl.Operation;
import org.apache.ignite.internal.metastorage.dsl.StatementResult;
-import org.apache.ignite.internal.metastorage.exceptions.CompactedException;
-import org.apache.ignite.internal.metastorage.exceptions.OperationTimeoutException;
import org.apache.ignite.internal.metastorage.server.KeyValueStorage;
import org.apache.ignite.internal.metastorage.server.raft.MetaStorageLearnerListener;
import org.apache.ignite.internal.metastorage.server.raft.MetaStorageListener;
@@ -77,7 +77,7 @@ public class MetaStorageManagerImpl implements MetaStorageManager {
/**
* Special key for the Vault where the applied revision is stored.
*/
- private static final ByteArray APPLIED_REV = ByteArray.fromString("applied_revision");
+ private static final String APPLIED_REV_PREFIX = "applied_revision_";
private final ClusterService clusterService;
@@ -103,11 +103,6 @@ public class MetaStorageManagerImpl implements MetaStorageManager {
/** Prevents double stopping of the component. */
private final AtomicBoolean isStopped = new AtomicBoolean();
- /**
- * Applied revision of Meta storage, that is, the most recent revision that has been processed on this node.
- */
- private volatile long appliedRevision;
-
/**
* The constructor.
*
@@ -177,8 +172,6 @@ public class MetaStorageManagerImpl implements MetaStorageManager {
@Override
public void start() {
- this.appliedRevision = readAppliedRevision().join();
-
storage.start();
cmgMgr.metaStorageNodes()
@@ -219,23 +212,32 @@ public class MetaStorageManagerImpl implements MetaStorageManager {
}
@Override
- public long appliedRevision() {
- return appliedRevision;
+ public CompletableFuture<Long> appliedRevision(String watchId) {
+ return vaultMgr.get(appliedRevisionKey(watchId))
+ .thenApply(appliedRevision -> appliedRevision == null ? 0L : bytesToLong(appliedRevision.value()));
+ }
+
+ private long appliedRevision(WatchListener lsnr) {
+ return appliedRevision(lsnr.id()).join();
+ }
+
+ private static ByteArray appliedRevisionKey(String watchId) {
+ return ByteArray.fromString(APPLIED_REV_PREFIX + watchId);
}
@Override
- public void registerPrefixWatch(ByteArray key, WatchListener lsnr) {
- storage.watchPrefix(key.bytes(), appliedRevision + 1, lsnr);
+ public void registerPrefixWatch(ByteArray key, WatchListener listener) {
+ storage.watchPrefix(key.bytes(), appliedRevision(listener) + 1, listener);
}
@Override
public void registerExactWatch(ByteArray key, WatchListener listener) {
- storage.watchExact(key.bytes(), appliedRevision + 1, listener);
+ storage.watchExact(key.bytes(), appliedRevision(listener) + 1, listener);
}
@Override
public void registerRangeWatch(ByteArray keyFrom, @Nullable ByteArray keyTo, WatchListener listener) {
- storage.watchRange(keyFrom.bytes(), keyTo == null ? null : keyTo.bytes(), appliedRevision + 1, listener);
+ storage.watchRange(keyFrom.bytes(), keyTo == null ? null : keyTo.bytes(), appliedRevision(listener) + 1, listener);
}
@Override
@@ -528,30 +530,6 @@ public class MetaStorageManagerImpl implements MetaStorageManager {
}
}
- /**
- * Retrieves entries for the given key range in lexicographic order. Entries will be filtered out by the current applied revision as an
- * upper bound. Applied revision is a revision of the last successful vault update.
- *
- * @param keyFrom Start key of range (inclusive). Couldn't be {@code null}.
- * @param keyTo End key of range (exclusive). Could be {@code null}.
- * @return Cursor built upon entries corresponding to the given range and applied revision.
- * @throws OperationTimeoutException If the operation is timed out.
- * @throws CompactedException If the desired revisions are removed from the storage due to a compaction.
- * @see ByteArray
- * @see Entry
- */
- public Publisher<Entry> rangeWithAppliedRevision(ByteArray keyFrom, @Nullable ByteArray keyTo) {
- if (!busyLock.enterBusy()) {
- return new NodeStoppingPublisher<>();
- }
-
- try {
- return new CompletableFuturePublisher<>(metaStorageSvcFut.thenApply(svc -> svc.range(keyFrom, keyTo, appliedRevision)));
- } finally {
- busyLock.leaveBusy();
- }
- }
-
@Override
public Publisher<Entry> prefix(ByteArray keyPrefix) {
return prefix(keyPrefix, -1);
@@ -587,36 +565,28 @@ public class MetaStorageManagerImpl implements MetaStorageManager {
}
}
- /**
- * Returns applied revision from the local storage.
- */
- private CompletableFuture<Long> readAppliedRevision() {
- return vaultMgr.get(APPLIED_REV)
- .thenApply(appliedRevision -> appliedRevision == null ? 0L : bytesToLong(appliedRevision.value()));
- }
-
/**
* Saves processed Meta Storage revision and corresponding entries to the Vault.
*/
- private void saveUpdatedEntriesToVault(long revision, Collection<Entry> updatedEntries) {
- assert revision > appliedRevision : "Remote revision " + revision + " is greater than applied revision " + appliedRevision;
-
+ private CompletableFuture<Void> saveUpdatedEntriesToVault(String watchId, WatchEvent watchEvent) {
if (!busyLock.enterBusy()) {
LOG.info("Skipping applying MetaStorage revision because the node is stopping");
- return;
+ return completedFuture(null);
}
try {
- Map<ByteArray, byte[]> batch = IgniteUtils.newHashMap(updatedEntries.size() + 1);
-
- batch.put(APPLIED_REV, longToBytes(revision));
+ if (watchEvent.entryEvents().isEmpty()) {
+ return vaultMgr.put(appliedRevisionKey(watchId), longToBytes(watchEvent.revision()));
+ } else {
+ Map<ByteArray, byte[]> batch = IgniteUtils.newHashMap(watchEvent.entryEvents().size() + 1);
- updatedEntries.forEach(e -> batch.put(new ByteArray(e.key()), e.value()));
+ batch.put(appliedRevisionKey(watchId), longToBytes(watchEvent.revision()));
- vaultMgr.putAll(batch).join();
+ watchEvent.entryEvents().forEach(e -> batch.put(new ByteArray(e.newEntry().key()), e.newEntry().value()));
- appliedRevision = revision;
+ return vaultMgr.putAll(batch);
+ }
} finally {
busyLock.leaveBusy();
}
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/KeyValueStorage.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/KeyValueStorage.java
index a1f3c8511b..1b49a067a5 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/KeyValueStorage.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/KeyValueStorage.java
@@ -286,10 +286,4 @@ public interface KeyValueStorage extends ManuallyCloseable {
* @param snapshotPath Path to the snapshot's directory.
*/
void restoreSnapshot(Path snapshotPath);
-
- /**
- * Closes the storage.
- */
- @Override
- void close();
}
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/OnRevisionAppliedCallback.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/OnRevisionAppliedCallback.java
index 436233abfc..155ffb5b69 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/OnRevisionAppliedCallback.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/OnRevisionAppliedCallback.java
@@ -17,8 +17,8 @@
package org.apache.ignite.internal.metastorage.server;
-import java.util.Collection;
-import org.apache.ignite.internal.metastorage.Entry;
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.internal.metastorage.WatchEvent;
/**
* Interface for declaring callbacks that get called after all Meta Storage watches have been notified of a particular revision.
@@ -26,10 +26,11 @@ import org.apache.ignite.internal.metastorage.Entry;
@FunctionalInterface
public interface OnRevisionAppliedCallback {
/**
- * Notifies of completion of processing of Meta Storage watches for a particular revision.
+ * Notifies of completion of processing of Meta Storage watches for a particular Watch and revision.
*
- * @param revision Revision.
- * @param updatedEntries Entries that have been modified under this revision and processed by at least one watch.
+ * @param watchId ID of the Watch that has finished processing the revision.
+ * @param watchEvent Event with modified Meta Storage entries processed by the Watch.
+ * @return Future that represents the state of the execution of the callback.
*/
- void onRevisionApplied(long revision, Collection<Entry> updatedEntries);
+ CompletableFuture<Void> onRevisionApplied(String watchId, WatchEvent watchEvent);
}
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/Watch.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/Watch.java
index 489de7c898..42c35ec9e3 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/Watch.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/Watch.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.metastorage.server;
+import java.util.concurrent.CompletableFuture;
import java.util.function.Predicate;
import org.apache.ignite.internal.metastorage.WatchEvent;
import org.apache.ignite.internal.metastorage.WatchListener;
@@ -59,11 +60,24 @@ public class Watch {
/**
* Notifies the event listener about a Meta Storage event.
+ *
+ * @see WatchListener#onUpdate
*/
- public void onUpdate(WatchEvent event) {
- listener.onUpdate(event);
-
+ public CompletableFuture<Void> onUpdate(WatchEvent event) {
targetRevision = event.revision() + 1;
+
+ return listener.onUpdate(event);
+ }
+
+ /**
+ * Notifies the event listener about a Meta Storage revision update.
+ *
+ * @see WatchListener#onRevisionUpdated
+ */
+ public CompletableFuture<Void> onRevisionUpdated(long revision) {
+ targetRevision = revision + 1;
+
+ return listener.onRevisionUpdated(revision);
}
/**
@@ -73,6 +87,13 @@ public class Watch {
listener.onError(e);
}
+ /**
+ * Returns the ID of the Watch.
+ */
+ public String id() {
+ return listener.id();
+ }
+
/**
* Returns the event listener.
*/
@@ -86,4 +107,23 @@ public class Watch {
public long targetRevision() {
return targetRevision;
}
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ Watch watch = (Watch) o;
+
+ return id().equals(watch.id());
+ }
+
+ @Override
+ public int hashCode() {
+ return id().hashCode();
+ }
}
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/WatchProcessor.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/WatchProcessor.java
index fb8337c0b2..6189c12cee 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/WatchProcessor.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/WatchProcessor.java
@@ -17,22 +17,35 @@
package org.apache.ignite.internal.metastorage.server;
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.concurrent.CompletableFuture.failedFuture;
+
import java.util.ArrayList;
-import java.util.HashSet;
import java.util.List;
import java.util.OptionalLong;
-import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import org.apache.ignite.internal.close.ManuallyCloseable;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.internal.metastorage.Entry;
import org.apache.ignite.internal.metastorage.EntryEvent;
import org.apache.ignite.internal.metastorage.WatchEvent;
import org.apache.ignite.internal.metastorage.WatchListener;
+import org.apache.ignite.internal.thread.NamedThreadFactory;
+import org.apache.ignite.internal.util.IgniteUtils;
/**
* Class for storing and notifying Meta Storage Watches.
+ *
+ * <p>Every Meta Storage update is processed by each registered Watch in parallel, however notifications for a single Watch are
+ * linearised (Watches are always notified of one event at a time and in increasing order of revisions).
*/
-public class WatchProcessor {
+public class WatchProcessor implements ManuallyCloseable {
/** Reads an entry from the storage using a given key and revision. */
@FunctionalInterface
public interface EntryReader {
@@ -41,38 +54,45 @@ public class WatchProcessor {
private static final IgniteLogger LOG = Loggers.forClass(WatchProcessor.class);
- /** List of watches that listen to storage events. */
- private final List<Watch> watches = new CopyOnWriteArrayList<>();
+ /** Map that contains Watches and corresponding Watch notification process (represented as a CompletableFuture). */
+ private final ConcurrentMap<Watch, CompletableFuture<Void>> watches = new ConcurrentHashMap<>();
private final EntryReader entryReader;
- /** Callback that gets notified after a {@link WatchEvent} has been processed by all registered watches. */
+ /** Callback that gets notified after a {@link WatchEvent} has been processed by a registered watch. */
private volatile OnRevisionAppliedCallback revisionCallback;
+ /** Executor for processing watch events. */
+ private final ExecutorService watchExecutor;
+
/**
* Creates a new instance.
*
* @param entryReader Function for reading an entry from the storage using a given key and revision.
*/
- public WatchProcessor(EntryReader entryReader) {
+ public WatchProcessor(String nodeName, EntryReader entryReader) {
this.entryReader = entryReader;
+
+ this.watchExecutor = Executors.newFixedThreadPool(4, NamedThreadFactory.create(nodeName, "metastorage-watch-executor", LOG));
}
/** Adds a watch. */
public void addWatch(Watch watch) {
- watches.add(watch);
+ assert !watches.containsKey(watch) : "Watch with id \"" + watch.id() + "\" already exists";
+
+ watches.put(watch, completedFuture(null));
}
/** Removes a watch (identified by its listener). */
public void removeWatch(WatchListener listener) {
- watches.removeIf(watch -> watch.listener() == listener);
+ watches.keySet().removeIf(watch -> watch.listener() == listener);
}
/**
* Returns the minimal target revision of all registered watches.
*/
public OptionalLong minWatchRevision() {
- return watches.stream()
+ return watches.keySet().stream()
.mapToLong(Watch::targetRevision)
.min();
}
@@ -90,47 +110,62 @@ public class WatchProcessor {
* Notifies registered watch about an update event.
*/
public void notifyWatches(List<Entry> updatedEntries) {
+ watches.replaceAll((watch, watchOperation) ->
+ watchOperation.thenComposeAsync(v -> notifyWatch(watch, updatedEntries), watchExecutor));
+ }
+
+ private CompletableFuture<Void> notifyWatch(Watch watch, List<Entry> updatedEntries) {
// Revision must be the same for all entries.
long newRevision = updatedEntries.get(0).revision();
- // Set of entries which keys match at least one watch.
- var matchedEntries = new HashSet<Entry>();
-
- for (Watch watch : watches) {
- var entryEvents = new ArrayList<EntryEvent>();
+ List<EntryEvent> entryEvents = List.of();
- for (Entry newEntry : updatedEntries) {
- byte[] newKey = newEntry.key();
+ for (Entry newEntry : updatedEntries) {
+ byte[] newKey = newEntry.key();
- assert newEntry.revision() == newRevision;
+ assert newEntry.revision() == newRevision;
- if (watch.matches(newKey, newRevision)) {
- Entry oldEntry = entryReader.get(newKey, newRevision - 1);
+ if (watch.matches(newKey, newRevision)) {
+ Entry oldEntry = entryReader.get(newKey, newRevision - 1);
- entryEvents.add(new EntryEvent(oldEntry, newEntry));
-
- matchedEntries.add(newEntry);
+ if (entryEvents.isEmpty()) {
+ entryEvents = new ArrayList<>();
}
- }
- if (!entryEvents.isEmpty()) {
- var event = new WatchEvent(entryEvents, newRevision);
+ entryEvents.add(new EntryEvent(oldEntry, newEntry));
+ }
+ }
- try {
- watch.onUpdate(event);
- } catch (Exception e) {
- // TODO: IGNITE-14693 Implement Meta storage exception handling logic.
- LOG.error("Error occurred when processing a watch event {}, watch processing will be stopped", e, event);
+ var event = new WatchEvent(entryEvents, newRevision);
- watch.onError(e);
+ CompletableFuture<Void> eventNotificationFuture;
- throw e;
- }
- }
+ try {
+ eventNotificationFuture = entryEvents.isEmpty()
+ ? watch.onRevisionUpdated(newRevision)
+ : watch.onUpdate(event);
+ } catch (Throwable e) {
+ eventNotificationFuture = failedFuture(e);
}
- if (!matchedEntries.isEmpty()) {
- revisionCallback.onRevisionApplied(newRevision, matchedEntries);
- }
+ return eventNotificationFuture
+ .whenComplete((v, e) -> {
+ if (e != null) {
+ // TODO: IGNITE-14693 Implement Meta storage exception handling logic.
+ LOG.error("Error occurred when processing a watch event {}, watch {} is going to be stopped", e, event, watch.id());
+
+ watch.onError(e);
+
+ watches.remove(watch);
+ }
+ })
+ .thenCompose(v -> revisionCallback.onRevisionApplied(watch.id(), event));
+ }
+
+ @Override
+ public void close() {
+ watches.values().forEach(f -> f.cancel(true));
+
+ IgniteUtils.shutdownAndAwaitTermination(watchExecutor, 10, TimeUnit.SECONDS);
}
}
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/persistence/RocksDbKeyValueStorage.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/persistence/RocksDbKeyValueStorage.java
index 31c4331e3e..f993e87f27 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/persistence/RocksDbKeyValueStorage.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/persistence/RocksDbKeyValueStorage.java
@@ -36,7 +36,6 @@ import static org.apache.ignite.lang.ErrorGroups.MetaStorage.COMPACTION_ERR;
import static org.apache.ignite.lang.ErrorGroups.MetaStorage.OP_EXECUTION_ERR;
import static org.apache.ignite.lang.ErrorGroups.MetaStorage.RESTORING_STORAGE_ERR;
import static org.apache.ignite.lang.ErrorGroups.MetaStorage.STARTING_STORAGE_ERR;
-import static org.apache.ignite.lang.ErrorGroups.MetaStorage.WATCH_EXECUTION_ERR;
import java.nio.charset.StandardCharsets;
import java.nio.file.Path;
@@ -47,12 +46,9 @@ import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.TreeSet;
-import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
-import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReadWriteLock;
@@ -170,17 +166,8 @@ public class RocksDbKeyValueStorage implements KeyValueStorage {
/** Update counter. Will be incremented for each update of any particular entry. */
private volatile long updCntr;
- /** Executor for processing watch events. */
- private final ExecutorService watchExecutor;
-
- /**
- * Future representing the task of polling the {@code eventQueue}. Needed for cancelling the task on storage stop.
- */
- @Nullable
- private volatile Future<?> watchExecutorFuture;
-
/** Watch processor. */
- private final WatchProcessor watchProcessor = new WatchProcessor(this::get);
+ private final WatchProcessor watchProcessor;
/** Status of the watch recovery process. */
private enum RecoveryStatus {
@@ -196,8 +183,14 @@ public class RocksDbKeyValueStorage implements KeyValueStorage {
*/
private final AtomicReference<RecoveryStatus> recoveryStatus = new AtomicReference<>(RecoveryStatus.INITIAL);
- /** Queue of update events, consumed by the {@link #watchExecutor}. */
- private final BlockingQueue<List<Entry>> eventQueue = new LinkedBlockingQueue<>();
+ /**
+ * Buffer used to cache new events while an event replay is in progress. After replay finishes, the cache gets drained and is never
+ * used again.
+ *
+ * <p>Multi-threaded access is guarded by {@link #rwLock}.
+ */
+ @Nullable
+ private List<List<Entry>> eventCache;
/**
* Current list of updated entries.
@@ -214,8 +207,9 @@ public class RocksDbKeyValueStorage implements KeyValueStorage {
public RocksDbKeyValueStorage(String nodeName, Path dbPath) {
this.dbPath = dbPath;
+ this.watchProcessor = new WatchProcessor(nodeName, this::get);
+
this.snapshotExecutor = Executors.newFixedThreadPool(2, NamedThreadFactory.create(nodeName, "metastorage-snapshot-executor", LOG));
- this.watchExecutor = Executors.newSingleThreadExecutor(NamedThreadFactory.create(nodeName, "metastorage-watch-executor", LOG));
}
/** {@inheritDoc} */
@@ -281,16 +275,9 @@ public class RocksDbKeyValueStorage implements KeyValueStorage {
}
}
- /** {@inheritDoc} */
@Override
public void close() {
- Future<?> f = watchExecutorFuture;
-
- if (f != null) {
- f.cancel(true);
- }
-
- IgniteUtils.shutdownAndAwaitTermination(watchExecutor, 10, TimeUnit.SECONDS);
+ watchProcessor.close();
IgniteUtils.shutdownAndAwaitTermination(snapshotExecutor, 10, TimeUnit.SECONDS);
@@ -1192,32 +1179,36 @@ public class RocksDbKeyValueStorage implements KeyValueStorage {
case PENDING:
// Watches have been enabled, but no event replay happened because there was no data in the
- // storage. Since we are adding some data here, we need to start the watch processor thread.
- recoveryStatus.set(RecoveryStatus.IN_PROGRESS);
+ // storage. We need to finish the recovery process (by simply updating the status) and process the events as usual.
+ recoveryStatus.set(RecoveryStatus.DONE);
+
+ notifyWatches();
+
+ break;
- startWatchExecutor();
+ case IN_PROGRESS:
+ // Buffer the event while event replay is still in progress.
+ if (eventCache == null) {
+ eventCache = new ArrayList<>();
+ }
- publishWatchEvent();
+ eventCache.add(List.copyOf(updatedEntries));
+
+ updatedEntries.clear();
break;
default:
- publishWatchEvent();
+ notifyWatches();
break;
}
}
- private void publishWatchEvent() {
- List<Entry> updatedEntriesCopy = List.copyOf(updatedEntries);
+ private void notifyWatches() {
+ watchProcessor.notifyWatches(List.copyOf(updatedEntries));
updatedEntries.clear();
-
- try {
- eventQueue.put(updatedEntriesCopy);
- } catch (InterruptedException e) {
- throw new MetaStorageException(WATCH_EXECUTION_ERR, "Interrupted when publishing a watch event", e);
- }
}
private void replayUpdates(long upperRevision) {
@@ -1225,7 +1216,8 @@ public class RocksDbKeyValueStorage implements KeyValueStorage {
if (minWatchRevision == -1 || minWatchRevision > upperRevision) {
// No events to replay, we can start processing more recent events from the event queue.
- startWatchExecutor();
+ finishReplay();
+
return;
}
@@ -1250,7 +1242,7 @@ public class RocksDbKeyValueStorage implements KeyValueStorage {
if (!updatedEntries.isEmpty()) {
var updatedEntriesCopy = List.copyOf(updatedEntries);
- watchExecutor.execute(() -> watchProcessor.notifyWatches(updatedEntriesCopy));
+ watchProcessor.notifyWatches(updatedEntriesCopy);
updatedEntries.clear();
}
@@ -1265,27 +1257,28 @@ public class RocksDbKeyValueStorage implements KeyValueStorage {
// Notify about the events left after finishing the cycle above.
if (!updatedEntries.isEmpty()) {
- watchExecutor.execute(() -> watchProcessor.notifyWatches(updatedEntries));
+ watchProcessor.notifyWatches(updatedEntries);
}
}
- // Events replay is finished, now we can start processing more recent events from the event queue.
- startWatchExecutor();
+ finishReplay();
}
- private void startWatchExecutor() {
- if (recoveryStatus.compareAndSet(RecoveryStatus.IN_PROGRESS, RecoveryStatus.DONE)) {
- watchExecutorFuture = watchExecutor.submit(() -> {
- while (!Thread.currentThread().isInterrupted()) {
- try {
- watchProcessor.notifyWatches(eventQueue.take());
- } catch (InterruptedException e) {
- LOG.info("Watch Executor interrupted, watches stopped");
+ private void finishReplay() {
+ // Take the lock to drain the event cache and prevent new events from being cached. Since event notification is asynchronous,
+ // this lock shouldn't be held for long.
+ rwLock.writeLock().lock();
- return;
- }
- }
- });
+ try {
+ if (eventCache != null) {
+ eventCache.forEach(watchProcessor::notifyWatches);
+
+ eventCache = null;
+ }
+
+ recoveryStatus.set(RecoveryStatus.DONE);
+ } finally {
+ rwLock.writeLock().unlock();
}
}
diff --git a/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/AbstractKeyValueStorageTest.java b/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/AbstractKeyValueStorageTest.java
index f767f800a7..4cd848247b 100644
--- a/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/AbstractKeyValueStorageTest.java
+++ b/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/AbstractKeyValueStorageTest.java
@@ -18,6 +18,7 @@
package org.apache.ignite.internal.metastorage.server;
import static java.nio.charset.StandardCharsets.UTF_8;
+import static java.util.concurrent.CompletableFuture.completedFuture;
import static java.util.function.Function.identity;
import static org.apache.ignite.internal.metastorage.dsl.Operations.ops;
import static org.apache.ignite.internal.metastorage.dsl.Operations.put;
@@ -33,12 +34,12 @@ import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.never;
import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
import java.util.Collection;
import java.util.List;
@@ -83,7 +84,7 @@ public abstract class AbstractKeyValueStorageTest {
* After each.
*/
@AfterEach
- void tearDown() {
+ void tearDown() throws Exception {
storage.close();
}
@@ -2086,16 +2087,21 @@ public abstract class AbstractKeyValueStorageTest {
long appliedRevision = storage.revision();
- storage.startWatches((revision, updatedEntries) -> {
- // No-op.
- });
+ storage.startWatches((revision, updatedEntries) -> completedFuture(null));
CompletableFuture<byte[]> fut = new CompletableFuture<>();
storage.watchExact(key(0), appliedRevision + 1, new WatchListener() {
@Override
- public void onUpdate(WatchEvent event) {
+ public String id() {
+ return "test";
+ }
+
+ @Override
+ public CompletableFuture<Void> onUpdate(WatchEvent event) {
fut.complete(event.entryEvent().newEntry().value());
+
+ return completedFuture(null);
}
@Override
@@ -2397,7 +2403,7 @@ public abstract class AbstractKeyValueStorageTest {
}
/**
- * Tests that, if a watch throws an exception, its {@code onError} method is invoked and all other watches do not get executed.
+ * Tests that, if a watch throws an exception, its {@code onError} method is invoked.
*/
@Test
void testWatchErrorHandling() {
@@ -2409,6 +2415,15 @@ public abstract class AbstractKeyValueStorageTest {
WatchListener mockListener2 = mock(WatchListener.class);
WatchListener mockListener3 = mock(WatchListener.class);
+ when(mockListener1.id()).thenReturn("test1");
+ when(mockListener1.onUpdate(any())).thenReturn(completedFuture(null));
+
+ when(mockListener2.id()).thenReturn("test2");
+ when(mockListener2.onUpdate(any())).thenReturn(completedFuture(null));
+
+ when(mockListener3.id()).thenReturn("test3");
+ when(mockListener3.onUpdate(any())).thenReturn(completedFuture(null));
+
var exception = new IllegalStateException();
doThrow(exception).when(mockListener2).onUpdate(any());
@@ -2419,6 +2434,8 @@ public abstract class AbstractKeyValueStorageTest {
OnRevisionAppliedCallback mockCallback = mock(OnRevisionAppliedCallback.class);
+ when(mockCallback.onRevisionApplied(any(), any())).thenReturn(completedFuture(null));
+
storage.startWatches(mockCallback);
storage.put(key, value);
@@ -2427,9 +2444,9 @@ public abstract class AbstractKeyValueStorageTest {
verify(mockListener2, timeout(10_000)).onError(exception);
- verify(mockListener3, never()).onUpdate(any());
- verify(mockListener3, never()).onError(any());
- verify(mockCallback, never()).onRevisionApplied(anyLong(), any());
+ verify(mockListener3, timeout(10_000)).onUpdate(any());
+
+ verify(mockCallback, times(2)).onRevisionApplied(any(), any());
}
private static void fill(KeyValueStorage storage, int keySuffix, int num) {
@@ -2473,7 +2490,12 @@ public abstract class AbstractKeyValueStorageTest {
watchMethod.accept(new WatchListener() {
@Override
- public void onUpdate(WatchEvent event) {
+ public String id() {
+ return "test";
+ }
+
+ @Override
+ public CompletableFuture<Void> onUpdate(WatchEvent event) {
try {
var curState = state.incrementAndGet();
@@ -2482,9 +2504,13 @@ public abstract class AbstractKeyValueStorageTest {
if (curState == expectedNumCalls) {
resultFuture.complete(null);
}
+
+ return completedFuture(null);
} catch (Exception e) {
resultFuture.completeExceptionally(e);
}
+
+ return completedFuture(null);
}
@Override
@@ -2493,7 +2519,7 @@ public abstract class AbstractKeyValueStorageTest {
}
});
- storage.startWatches((revision, updatedEntries) -> {});
+ storage.startWatches((revision, updatedEntries) -> completedFuture(null));
return resultFuture;
}
diff --git a/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/RocksDbKeyValueStorageTest.java b/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/RocksDbKeyValueStorageTest.java
index 64ca76230e..51cc7dac6d 100644
--- a/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/RocksDbKeyValueStorageTest.java
+++ b/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/RocksDbKeyValueStorageTest.java
@@ -25,6 +25,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
import java.nio.file.Path;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.apache.ignite.internal.metastorage.WatchEvent;
@@ -50,7 +51,7 @@ public class RocksDbKeyValueStorageTest extends AbstractKeyValueStorageTest {
}
@Test
- void testWatchReplayOnSnapshotLoad() throws InterruptedException {
+ void testWatchReplayOnSnapshotLoad() throws Exception {
storage.put("foo".getBytes(UTF_8), "bar".getBytes(UTF_8));
storage.put("baz".getBytes(UTF_8), "quux".getBytes(UTF_8));
@@ -70,10 +71,17 @@ public class RocksDbKeyValueStorageTest extends AbstractKeyValueStorageTest {
storage.watchExact("foo".getBytes(UTF_8), 1, new WatchListener() {
@Override
- public void onUpdate(WatchEvent event) {
+ public String id() {
+ return "test1";
+ }
+
+ @Override
+ public CompletableFuture<Void> onUpdate(WatchEvent event) {
assertThat(event.entryEvent().newEntry().value(), is("bar".getBytes(UTF_8)));
latch.countDown();
+
+ return CompletableFuture.completedFuture(null);
}
@Override
@@ -84,10 +92,17 @@ public class RocksDbKeyValueStorageTest extends AbstractKeyValueStorageTest {
storage.watchExact("baz".getBytes(UTF_8), 1, new WatchListener() {
@Override
- public void onUpdate(WatchEvent event) {
+ public String id() {
+ return "test2";
+ }
+
+ @Override
+ public CompletableFuture<Void> onUpdate(WatchEvent event) {
assertThat(event.entryEvent().newEntry().value(), is("quux".getBytes(UTF_8)));
latch.countDown();
+
+ return CompletableFuture.completedFuture(null);
}
@Override
@@ -96,7 +111,7 @@ public class RocksDbKeyValueStorageTest extends AbstractKeyValueStorageTest {
}
});
- storage.startWatches((revision, updatedEntries) -> {});
+ storage.startWatches((revision, updatedEntries) -> CompletableFuture.completedFuture(null));
storage.restoreSnapshot(snapshotPath);
diff --git a/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/WatchProcessorTest.java b/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/WatchProcessorTest.java
new file mode 100644
index 0000000000..9f43d28e53
--- /dev/null
+++ b/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/WatchProcessorTest.java
@@ -0,0 +1,225 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.metastorage.server;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.clearInvocations;
+import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.internal.metastorage.Entry;
+import org.apache.ignite.internal.metastorage.EntryEvent;
+import org.apache.ignite.internal.metastorage.WatchEvent;
+import org.apache.ignite.internal.metastorage.WatchListener;
+import org.apache.ignite.internal.metastorage.impl.EntryImpl;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.InOrder;
+
+/**
+ * Tests for {@link WatchProcessor}.
+ */
+public class WatchProcessorTest {
+ private final WatchProcessor watchProcessor = new WatchProcessor("test", WatchProcessorTest::oldEntry);
+
+ private final OnRevisionAppliedCallback revisionCallback = mock(OnRevisionAppliedCallback.class);
+
+ @BeforeEach
+ void setUp() {
+ when(revisionCallback.onRevisionApplied(any(), any())).thenReturn(completedFuture(null));
+
+ watchProcessor.setRevisionCallback(revisionCallback);
+ }
+
+ @AfterEach
+ void tearDown() {
+ watchProcessor.close();
+ }
+
+ /**
+ * Tests a scenario when updates for different listeners come in a single batch.
+ */
+ @Test
+ void testGroupEventNotification() {
+ WatchListener listener1 = mockListener();
+ WatchListener listener2 = mockListener();
+
+ watchProcessor.addWatch(new Watch(0, listener1, key -> Arrays.equals(key, "foo".getBytes(UTF_8))));
+ watchProcessor.addWatch(new Watch(0, listener2, key -> Arrays.equals(key, "bar".getBytes(UTF_8))));
+
+ var entry1 = new EntryImpl("foo".getBytes(UTF_8), null, 1, 0);
+ var entry2 = new EntryImpl("bar".getBytes(UTF_8), null, 1, 0);
+
+ watchProcessor.notifyWatches(List.of(entry1, entry2));
+
+ verify(listener1, timeout(1_000)).onUpdate(new WatchEvent(new EntryEvent(oldEntry(entry1), entry1)));
+ verify(listener2, timeout(1_000)).onUpdate(new WatchEvent(new EntryEvent(oldEntry(entry2), entry2)));
+
+ verify(listener1, never()).onRevisionUpdated(anyLong());
+ verify(listener2, never()).onRevisionUpdated(anyLong());
+
+ // Revision callback should be called for every listener update.
+ verify(revisionCallback, timeout(1_000).times(2)).onRevisionApplied(any(), any());
+ }
+
+ /**
+ * Tests a scenario when updates for different listeners come sequentially.
+ */
+ @Test
+ void testSequentialEventNotification() {
+ WatchListener listener1 = mockListener();
+ WatchListener listener2 = mockListener();
+
+ watchProcessor.addWatch(new Watch(0, listener1, key -> Arrays.equals(key, "foo".getBytes(UTF_8))));
+ watchProcessor.addWatch(new Watch(0, listener2, key -> Arrays.equals(key, "bar".getBytes(UTF_8))));
+
+ var entry1 = new EntryImpl("foo".getBytes(UTF_8), null, 1, 0);
+ var entry2 = new EntryImpl("bar".getBytes(UTF_8), null, 2, 0);
+
+ watchProcessor.notifyWatches(List.of(entry1));
+
+ verify(listener1, timeout(1_000)).onUpdate(new WatchEvent(new EntryEvent(oldEntry(entry1), entry1)));
+ verify(listener2, timeout(1_000)).onRevisionUpdated(1);
+
+ watchProcessor.notifyWatches(List.of(entry2));
+
+ verify(listener1, timeout(1_000)).onRevisionUpdated(2);
+ verify(listener2, timeout(1_000)).onUpdate(new WatchEvent(new EntryEvent(oldEntry(entry2), entry2)));
+
+ // Revision callback should be called for every listener update.
+ verify(revisionCallback, timeout(1_000).times(4)).onRevisionApplied(any(), any());
+ }
+
+ /**
+ * Tests a scenario that, when a watch throws an exception, it gets disabled and other watches continue working.
+ */
+ @Test
+ void testWatchFailure() {
+ WatchListener listener1 = mockListener();
+
+ WatchListener listener2 = mock(WatchListener.class);
+
+ when(listener2.id()).thenReturn("error");
+ when(listener2.onUpdate(any())).thenThrow(new IllegalStateException());
+
+ watchProcessor.addWatch(new Watch(0, listener1, key -> Arrays.equals(key, "foo".getBytes(UTF_8))));
+ watchProcessor.addWatch(new Watch(0, listener2, key -> Arrays.equals(key, "bar".getBytes(UTF_8))));
+
+ var entry1 = new EntryImpl("foo".getBytes(UTF_8), null, 1, 0);
+ var entry2 = new EntryImpl("bar".getBytes(UTF_8), null, 1, 0);
+
+ watchProcessor.notifyWatches(List.of(entry1, entry2));
+
+ verify(listener1, timeout(1_000)).onUpdate(new WatchEvent(new EntryEvent(oldEntry(entry1), entry1)));
+ verify(listener2, timeout(1_000)).onUpdate(new WatchEvent(new EntryEvent(oldEntry(entry2), entry2)));
+ verify(listener2, timeout(1_000)).onError(any(IllegalStateException.class));
+
+ var entry3 = new EntryImpl("bar".getBytes(UTF_8), null, 2, 0);
+
+ watchProcessor.notifyWatches(List.of(entry3));
+
+ clearInvocations(listener1, listener2, revisionCallback);
+
+ verify(listener1, timeout(1_000)).onRevisionUpdated(2);
+
+ verify(listener2, never()).onUpdate(any());
+ verify(listener2, never()).onError(any());
+ verify(listener2, never()).onRevisionUpdated(anyLong());
+
+ verify(revisionCallback, timeout(1_000)).onRevisionApplied(any(), any());
+ }
+
+ /**
+ * Tests watch notification independence (two watches can be processed simultaneously) and linearizability
+ * (one watch can process one event at a time).
+ */
+ @Test
+ void testNotificationParallelism() {
+ WatchListener listener1 = mockListener();
+
+ WatchListener listener2 = mock(WatchListener.class);
+
+ when(listener2.id()).thenReturn("blocking");
+
+ var blockingFuture = new CompletableFuture<Void>();
+
+ when(listener2.onUpdate(any()))
+ // Block the first call, the second call should work as usual.
+ .thenReturn(blockingFuture)
+ .thenReturn(completedFuture(null));
+
+ watchProcessor.addWatch(new Watch(0, listener1, key -> Arrays.equals(key, "foo".getBytes(UTF_8))));
+ watchProcessor.addWatch(new Watch(0, listener2, key -> Arrays.equals(key, "bar".getBytes(UTF_8))));
+
+ var entry1 = new EntryImpl("foo".getBytes(UTF_8), null, 1, 0);
+ var entry2 = new EntryImpl("bar".getBytes(UTF_8), null, 1, 0);
+
+ watchProcessor.notifyWatches(List.of(entry1, entry2));
+
+ verify(listener1, timeout(1_000)).onUpdate(new WatchEvent(new EntryEvent(oldEntry(entry1), entry1)));
+ verify(listener2, timeout(1_000)).onUpdate(new WatchEvent(new EntryEvent(oldEntry(entry2), entry2)));
+
+ var entry3 = new EntryImpl("foo".getBytes(UTF_8), null, 2, 0);
+ var entry4 = new EntryImpl("bar".getBytes(UTF_8), null, 2, 0);
+
+ watchProcessor.notifyWatches(List.of(entry3, entry4));
+
+ verify(listener1, timeout(1_000)).onUpdate(new WatchEvent(new EntryEvent(oldEntry(entry3), entry3)));
+ verify(listener2, never()).onUpdate(new WatchEvent(new EntryEvent(oldEntry(entry4), entry4)));
+
+ blockingFuture.complete(null);
+
+ InOrder inOrder = inOrder(listener2);
+
+ inOrder.verify(listener2, timeout(1_000)).onUpdate(new WatchEvent(new EntryEvent(oldEntry(entry2), entry2)));
+ inOrder.verify(listener2, timeout(1_000)).onUpdate(new WatchEvent(new EntryEvent(oldEntry(entry4), entry4)));
+ }
+
+ private static WatchListener mockListener() {
+ var listener = mock(WatchListener.class);
+
+ String id = UUID.randomUUID().toString();
+
+ when(listener.id()).thenReturn(id);
+
+ when(listener.onUpdate(any())).thenReturn(completedFuture(null));
+ when(listener.onRevisionUpdated(anyLong())).thenReturn(completedFuture(null));
+
+ return listener;
+ }
+
+ private static Entry oldEntry(byte[] key, long revision) {
+ return new EntryImpl(key, null, revision, 0);
+ }
+
+ private static Entry oldEntry(Entry entry) {
+ return new EntryImpl(entry.key(), null, entry.revision() - 1, 0);
+ }
+}
diff --git a/modules/metastorage/src/testFixtures/java/org/apache/ignite/internal/metastorage/server/SimpleInMemoryKeyValueStorage.java b/modules/metastorage/src/testFixtures/java/org/apache/ignite/internal/metastorage/server/SimpleInMemoryKeyValueStorage.java
index be26f84e0d..b83bf3d338 100644
--- a/modules/metastorage/src/testFixtures/java/org/apache/ignite/internal/metastorage/server/SimpleInMemoryKeyValueStorage.java
+++ b/modules/metastorage/src/testFixtures/java/org/apache/ignite/internal/metastorage/server/SimpleInMemoryKeyValueStorage.java
@@ -34,9 +34,6 @@ import java.util.OptionalLong;
import java.util.TreeMap;
import java.util.TreeSet;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
import java.util.function.Predicate;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
@@ -46,9 +43,7 @@ import org.apache.ignite.internal.metastorage.dsl.Operation;
import org.apache.ignite.internal.metastorage.dsl.StatementResult;
import org.apache.ignite.internal.metastorage.exceptions.MetaStorageException;
import org.apache.ignite.internal.metastorage.impl.EntryImpl;
-import org.apache.ignite.internal.thread.NamedThreadFactory;
import org.apache.ignite.internal.util.Cursor;
-import org.apache.ignite.internal.util.IgniteUtils;
import org.jetbrains.annotations.Nullable;
/**
@@ -82,14 +77,12 @@ public class SimpleInMemoryKeyValueStorage implements KeyValueStorage {
private boolean areWatchesEnabled = false;
- private final WatchProcessor watchProcessor = new WatchProcessor(this::get);
-
- private final ExecutorService watchExecutor;
+ private final WatchProcessor watchProcessor;
private final List<Entry> updatedEntries = new ArrayList<>();
public SimpleInMemoryKeyValueStorage(String nodeName) {
- this.watchExecutor = Executors.newSingleThreadExecutor(NamedThreadFactory.create(nodeName, "watch-executor", LOG));
+ this.watchProcessor = new WatchProcessor(nodeName, this::get);
}
/** {@inheritDoc} */
@@ -487,11 +480,9 @@ public class SimpleInMemoryKeyValueStorage implements KeyValueStorage {
return;
}
- var updatedEntriesCopy = List.copyOf(updatedEntries);
+ watchProcessor.notifyWatches(List.copyOf(updatedEntries));
updatedEntries.clear();
-
- watchExecutor.execute(() -> watchProcessor.notifyWatches(updatedEntriesCopy));
}
@Override
@@ -516,7 +507,7 @@ public class SimpleInMemoryKeyValueStorage implements KeyValueStorage {
@Override
public void close() {
- IgniteUtils.shutdownAndAwaitTermination(watchExecutor, 10, TimeUnit.SECONDS);
+ watchProcessor.close();
}
@Override
diff --git a/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/AssignmentsTracker.java b/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/AssignmentsTracker.java
index 72c8aaa867..04f6a51e4e 100644
--- a/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/AssignmentsTracker.java
+++ b/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/AssignmentsTracker.java
@@ -64,7 +64,7 @@ public class AssignmentsTracker {
private final TablesConfiguration tablesCfg;
/** Map replication group id to assignment nodes. */
- private Map<ReplicationGroupId, Set<Assignment>> groupAssignments;
+ private final Map<ReplicationGroupId, Set<Assignment>> groupAssignments;
/** Assignment configuration listener. */
private final AssignmentsCfgListener assignmentsCfgListener;
@@ -184,7 +184,12 @@ public class AssignmentsTracker {
*/
private class AssignmentsListener implements WatchListener {
@Override
- public void onUpdate(WatchEvent event) {
+ public String id() {
+ return STABLE_ASSIGNMENTS_PREFIX + "watch";
+ }
+
+ @Override
+ public CompletableFuture<Void> onUpdate(WatchEvent event) {
assert !event.entryEvent().newEntry().empty() : "New assignments are empty";
LOG.debug("Assignment update [revision={}, key={}]", event.revision(),
@@ -210,6 +215,8 @@ public class AssignmentsTracker {
if (leaseRenewalRequired) {
triggerToRenewLeases();
}
+
+ return completedFuture(null);
}
@Override
diff --git a/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/LeaseTracker.java b/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/LeaseTracker.java
index 4feb632178..ac937f3f83 100644
--- a/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/LeaseTracker.java
+++ b/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/LeaseTracker.java
@@ -17,9 +17,11 @@
package org.apache.ignite.internal.placementdriver;
+import static java.util.concurrent.CompletableFuture.completedFuture;
import static org.apache.ignite.internal.placementdriver.PlacementDriverManager.PLACEMENTDRIVER_PREFIX;
import java.util.Map;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
@@ -123,7 +125,12 @@ public class LeaseTracker {
*/
private class UpdateListener implements WatchListener {
@Override
- public void onUpdate(WatchEvent event) {
+ public String id() {
+ return PLACEMENTDRIVER_PREFIX + "watch";
+ }
+
+ @Override
+ public CompletableFuture<Void> onUpdate(WatchEvent event) {
for (EntryEvent entry : event.entryEvents()) {
Entry msEntry = entry.newEntry();
@@ -141,6 +148,8 @@ public class LeaseTracker {
leases.put(grpId, lease);
}
}
+
+ return completedFuture(null);
}
@Override
diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/ItDistributedConfigurationPropertiesTest.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/ItDistributedConfigurationPropertiesTest.java
index 0be9e12055..426ba80295 100644
--- a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/ItDistributedConfigurationPropertiesTest.java
+++ b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/ItDistributedConfigurationPropertiesTest.java
@@ -44,6 +44,7 @@ import org.apache.ignite.internal.cluster.management.raft.TestClusterStateStorag
import org.apache.ignite.internal.cluster.management.topology.LogicalTopologyImpl;
import org.apache.ignite.internal.cluster.management.topology.LogicalTopologyServiceImpl;
import org.apache.ignite.internal.configuration.storage.ConfigurationStorageListener;
+import org.apache.ignite.internal.configuration.storage.Data;
import org.apache.ignite.internal.configuration.storage.DistributedConfigurationStorage;
import org.apache.ignite.internal.configuration.testframework.ConfigurationExtension;
import org.apache.ignite.internal.configuration.testframework.InjectConfiguration;
@@ -162,11 +163,23 @@ public class ItDistributedConfigurationPropertiesTest {
/** {@inheritDoc} */
@Override
public synchronized void registerConfigurationListener(ConfigurationStorageListener listener) {
- super.registerConfigurationListener(changedEntries -> {
- if (receivesUpdates) {
- return listener.onEntriesChanged(changedEntries);
- } else {
- return CompletableFuture.completedFuture(null);
+ super.registerConfigurationListener(new ConfigurationStorageListener() {
+ @Override
+ public CompletableFuture<Void> onEntriesChanged(Data changedEntries) {
+ if (receivesUpdates) {
+ return listener.onEntriesChanged(changedEntries);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }
+
+ @Override
+ public CompletableFuture<Void> onRevisionUpdated(long newRevision) {
+ if (receivesUpdates) {
+ return listener.onRevisionUpdated(newRevision);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
}
});
}
diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/storage/ItDistributedConfigurationStorageTest.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/storage/ItDistributedConfigurationStorageTest.java
index 3576a62403..4db3c7f5d2 100644
--- a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/storage/ItDistributedConfigurationStorageTest.java
+++ b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/storage/ItDistributedConfigurationStorageTest.java
@@ -145,7 +145,17 @@ public class ItDistributedConfigurationStorageTest {
.forEach(IgniteComponent::start);
// this is needed to avoid assertion errors
- cfgStorage.registerConfigurationListener(changedEntries -> completedFuture(null));
+ cfgStorage.registerConfigurationListener(new ConfigurationStorageListener() {
+ @Override
+ public CompletableFuture<Void> onEntriesChanged(Data changedEntries) {
+ return completedFuture(null);
+ }
+
+ @Override
+ public CompletableFuture<Void> onRevisionUpdated(long newRevision) {
+ return completedFuture(null);
+ }
+ });
// deploy watches to propagate data from the metastore into the vault
metaStorageManager.deployWatches();
@@ -192,7 +202,10 @@ public class ItDistributedConfigurationStorageTest {
assertThat(node.cfgStorage.write(data, 0), willBe(equalTo(true)));
assertThat(node.cfgStorage.writeConfigurationRevision(0, 1), willCompleteSuccessfully());
- assertTrue(waitForCondition(() -> node.metaStorageManager.appliedRevision() != 0, 3000));
+ assertTrue(waitForCondition(
+ () -> node.metaStorageManager.appliedRevision(DistributedConfigurationStorage.WATCH_ID).join() != 0,
+ 3000
+ ));
} finally {
node.stop();
}
diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/storage/ItRebalanceDistributedTest.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/storage/ItRebalanceDistributedTest.java
index 36a2a75e34..933592c6ea 100644
--- a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/storage/ItRebalanceDistributedTest.java
+++ b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/storage/ItRebalanceDistributedTest.java
@@ -757,22 +757,27 @@ public class ItRebalanceDistributedTest {
}
@Override
- protected void handleChangeStableAssignmentEvent(WatchEvent evt) {
+ protected CompletableFuture<Void> handleChangeStableAssignmentEvent(WatchEvent evt) {
TablePartitionId tablePartitionId = getTablePartitionId(evt);
- try {
- super.handleChangeStableAssignmentEvent(evt);
+ return super.handleChangeStableAssignmentEvent(evt)
+ .whenComplete((v, e) -> {
+ if (tablePartitionId == null) {
+ return;
+ }
- if (tablePartitionId != null) {
- finishHandleChangeStableAssignmentEventFutures.get(tablePartitionId).complete(null);
- }
- } catch (Throwable t) {
- if (tablePartitionId != null) {
- finishHandleChangeStableAssignmentEventFutures.get(tablePartitionId).completeExceptionally(t);
- }
+ CompletableFuture<Void> finishFuture = finishHandleChangeStableAssignmentEventFutures.get(tablePartitionId);
- throw t;
- }
+ if (finishFuture == null) {
+ return;
+ }
+
+ if (e == null) {
+ finishFuture.complete(null);
+ } else {
+ finishFuture.completeExceptionally(e);
+ }
+ });
}
};
}
diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/test/WatchListenerInhibitor.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/test/WatchListenerInhibitor.java
index 95426b9b30..d58cd800de 100644
--- a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/test/WatchListenerInhibitor.java
+++ b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/test/WatchListenerInhibitor.java
@@ -19,26 +19,24 @@ package org.apache.ignite.internal.test;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.getFieldValue;
-import java.util.List;
-import java.util.concurrent.CountDownLatch;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
import org.apache.ignite.Ignite;
import org.apache.ignite.internal.app.IgniteImpl;
-import org.apache.ignite.internal.metastorage.WatchEvent;
-import org.apache.ignite.internal.metastorage.WatchListener;
import org.apache.ignite.internal.metastorage.impl.MetaStorageManagerImpl;
import org.apache.ignite.internal.metastorage.server.Watch;
import org.apache.ignite.internal.metastorage.server.WatchProcessor;
import org.apache.ignite.internal.metastorage.server.persistence.RocksDbKeyValueStorage;
/**
- * Listener which wraps another one to inhibit events.
+ * Class for blocking Watch processing on a given Ignite node.
*/
public class WatchListenerInhibitor {
/** "watches" field captured from the {@link RocksDbKeyValueStorage} instance. */
- private final List<Watch> watches;
+ private final Map<Watch, CompletableFuture<Void>> watches;
- /** Latch used to block the watch notification thread. */
- private final CountDownLatch inhibitLatch = new CountDownLatch(1);
+ /** Future used to block the watch notification thread. */
+ private final CompletableFuture<Void> inhibitFuture = new CompletableFuture<>();
/**
* Creates the specific listener which can inhibit events for real metastorage listener.
@@ -54,12 +52,12 @@ public class WatchListenerInhibitor {
var watchProcessor = (WatchProcessor) getFieldValue(storage, RocksDbKeyValueStorage.class, "watchProcessor");
- var watches = (List<Watch>) getFieldValue(watchProcessor, WatchProcessor.class, "watches");
+ var watches = (Map<Watch, CompletableFuture<Void>>) getFieldValue(watchProcessor, WatchProcessor.class, "watches");
return new WatchListenerInhibitor(watches);
}
- private WatchListenerInhibitor(List<Watch> watches) {
+ private WatchListenerInhibitor(Map<Watch, CompletableFuture<Void>> watches) {
this.watches = watches;
}
@@ -67,35 +65,13 @@ public class WatchListenerInhibitor {
* Starts inhibiting events.
*/
public void startInhibit() {
- // Inject a watch that matches all keys and revisions and blocks the watch notification thread until the latch is released.
- var blockingWatch = new Watch(
- 0,
- new WatchListener() {
- @Override
- public void onUpdate(WatchEvent event) {
- try {
- inhibitLatch.await();
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
-
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public void onError(Throwable e) {
- }
- },
- key -> true
- );
-
- watches.add(0, blockingWatch);
+ watches.replaceAll((watch, watchOperation) -> watchOperation.thenCompose(v -> inhibitFuture));
}
/**
* Stops inhibiting events.
*/
public void stopInhibit() {
- inhibitLatch.countDown();
+ inhibitFuture.complete(null);
}
}
diff --git a/modules/runner/src/main/java/org/apache/ignite/internal/configuration/storage/DistributedConfigurationStorage.java b/modules/runner/src/main/java/org/apache/ignite/internal/configuration/storage/DistributedConfigurationStorage.java
index 006c5729b7..6961512d8a 100644
--- a/modules/runner/src/main/java/org/apache/ignite/internal/configuration/storage/DistributedConfigurationStorage.java
+++ b/modules/runner/src/main/java/org/apache/ignite/internal/configuration/storage/DistributedConfigurationStorage.java
@@ -26,13 +26,11 @@ import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Flow.Subscriber;
import java.util.concurrent.Flow.Subscription;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicLong;
import org.apache.ignite.configuration.annotation.ConfigurationType;
import org.apache.ignite.internal.configuration.util.ConfigurationSerializationUtil;
import org.apache.ignite.internal.future.InFlightFutures;
@@ -67,6 +65,8 @@ public class DistributedConfigurationStorage implements ConfigurationStorage {
/** Prefix added to configuration keys to distinguish them in the meta storage. Must end with a dot. */
private static final String DISTRIBUTED_PREFIX = "dst-cfg.";
+ public static final String WATCH_ID = DISTRIBUTED_PREFIX + "watch";
+
/**
* Key for CAS-ing configuration keys to meta storage.
*/
@@ -85,8 +85,8 @@ public class DistributedConfigurationStorage implements ConfigurationStorage {
/**
* This key is expected to be the last key in lexicographical order of distributed configuration keys. It is possible because keys are
- * in lexicographical order in meta storage and adding {@code (char)('.' + 1)} to the end will produce all keys with prefix {@link
- * DistributedConfigurationStorage#DISTRIBUTED_PREFIX}
+ * in lexicographical order in meta storage and adding {@code (char)('.' + 1)} to the end will produce all keys with prefix
+ * {@link DistributedConfigurationStorage#DISTRIBUTED_PREFIX}
*/
private static final ByteArray DST_KEYS_END_RANGE = new ByteArray(incrementLastChar(DISTRIBUTED_PREFIX));
@@ -109,13 +109,13 @@ public class DistributedConfigurationStorage implements ConfigurationStorage {
*
* <p>This is true for all cases except for node restart. Key-specific revision values are lost on local vault copy after restart, so
* stored {@link MetaStorageManager#appliedRevision} value is used instead. This fact has very important side effect: it's no longer
- * possible to use {@link ConditionType#REV_EQUAL} on {@link #MASTER_KEY}
- * in {@link DistributedConfigurationStorage#write(Map, long)}. {@link ConditionType#REV_LESS_OR_EQUAL} must be used instead.
+ * possible to use {@link ConditionType#REV_EQUAL} on {@link #MASTER_KEY} in {@link DistributedConfigurationStorage#write(Map, long)}.
+ * {@link ConditionType#REV_LESS_OR_EQUAL} must be used instead.
*
* @see #MASTER_KEY
* @see #write(Map, long)
*/
- private final AtomicLong changeId = new AtomicLong(0L);
+ private volatile long changeId;
private final ExecutorService threadPool = Executors.newFixedThreadPool(4, new NamedThreadFactory("dst-cfg", LOG));
@@ -125,7 +125,7 @@ public class DistributedConfigurationStorage implements ConfigurationStorage {
* Constructor.
*
* @param metaStorageMgr Meta storage manager.
- * @param vaultMgr Vault manager.
+ * @param vaultMgr Vault manager.
*/
public DistributedConfigurationStorage(MetaStorageManager metaStorageMgr, VaultManager vaultMgr) {
this.metaStorageMgr = metaStorageMgr;
@@ -206,19 +206,18 @@ public class DistributedConfigurationStorage implements ConfigurationStorage {
});
}
- /** {@inheritDoc} */
@Override
public CompletableFuture<Data> readDataOnRecovery() throws StorageException {
- CompletableFuture<Data> future = vaultMgr.get(CONFIGURATION_REVISIONS_KEY)
- .thenApply(configurationRevision -> resolveRevision(metaStorageMgr.appliedRevision(), configurationRevision))
+ CompletableFuture<Data> future = metaStorageMgr.appliedRevision(WATCH_ID)
+ .thenCombine(vaultMgr.get(CONFIGURATION_REVISIONS_KEY), DistributedConfigurationStorage::resolveRevision)
.thenApplyAsync(this::readDataOnRecovery0, threadPool);
return registerFuture(future);
}
/**
- * Resolves current configuration revision based on the saved in the Vault revision of the metastorage and also
- * previous and current revisions of the configuration saved in the Vault.
+ * Resolves current configuration revision based on the saved in the Vault revision of the metastorage and also previous and current
+ * revisions of the configuration saved in the Vault.
*
* @param metaStorageRevision Meta Storage revision.
* @param revisionsEntry Configuration revisions entry.
@@ -264,7 +263,7 @@ public class DistributedConfigurationStorage implements ConfigurationStorage {
assert data.isEmpty() || cfgRevision > 0;
- changeId.set(cfgRevision);
+ changeId = cfgRevision;
return new Data(data, cfgRevision);
}
@@ -272,10 +271,10 @@ public class DistributedConfigurationStorage implements ConfigurationStorage {
/** {@inheritDoc} */
@Override
public CompletableFuture<Boolean> write(Map<String, ? extends Serializable> newValues, long curChangeId) {
- assert curChangeId <= changeId.get();
+ assert curChangeId <= changeId;
assert lsnr != null : "Configuration listener must be initialized before write.";
- if (curChangeId < changeId.get()) {
+ if (curChangeId < changeId) {
// This means that curChangeId is less than version and other node has already updated configuration and
// write should be retried. Actual version will be set when watch and corresponding configuration listener
// updates configuration.
@@ -303,67 +302,68 @@ public class DistributedConfigurationStorage implements ConfigurationStorage {
return metaStorageMgr.invoke(condition, operations, Set.of(Operations.noop()));
}
- /** {@inheritDoc} */
@Override
- public synchronized void registerConfigurationListener(ConfigurationStorageListener lsnr) {
- if (this.lsnr == null) {
- this.lsnr = lsnr;
+ public void registerConfigurationListener(ConfigurationStorageListener lsnr) {
+ assert this.lsnr == null;
- // TODO: registerPrefixWatch could throw OperationTimeoutException and CompactedException and we should
- // TODO: properly handle such cases https://issues.apache.org/jira/browse/IGNITE-14604
- metaStorageMgr.registerPrefixWatch(DST_KEYS_START_RANGE, new WatchListener() {
- @Override
- public void onUpdate(WatchEvent events) {
- var data = new HashMap<String, Serializable>();
+ this.lsnr = lsnr;
- Entry masterKeyEntry = null;
+ // TODO: registerPrefixWatch could throw OperationTimeoutException and CompactedException and we should
+ // TODO: properly handle such cases https://issues.apache.org/jira/browse/IGNITE-14604
+ metaStorageMgr.registerPrefixWatch(DST_KEYS_START_RANGE, new WatchListener() {
+ @Override
+ public String id() {
+ return WATCH_ID;
+ }
- for (EntryEvent event : events.entryEvents()) {
- Entry e = event.newEntry();
+ @Override
+ public CompletableFuture<Void> onUpdate(WatchEvent events) {
+ Map<String, Serializable> data = IgniteUtils.newHashMap(events.entryEvents().size() - 1);
- if (Arrays.equals(e.key(), MASTER_KEY.bytes())) {
- masterKeyEntry = e;
- } else {
- String key = new String(e.key(), UTF_8).substring(DISTRIBUTED_PREFIX.length());
+ Entry masterKeyEntry = null;
- Serializable value = e.value() == null ? null : ConfigurationSerializationUtil.fromBytes(e.value());
+ for (EntryEvent event : events.entryEvents()) {
+ Entry e = event.newEntry();
- data.put(key, value);
- }
+ if (Arrays.equals(e.key(), MASTER_KEY.bytes())) {
+ masterKeyEntry = e;
+ } else {
+ String key = new String(e.key(), UTF_8).substring(DISTRIBUTED_PREFIX.length());
+
+ Serializable value = e.value() == null ? null : ConfigurationSerializationUtil.fromBytes(e.value());
+
+ data.put(key, value);
}
+ }
- // Contract of meta storage ensures that all updates of one revision will come in one batch.
- // Also masterKey should be updated every time when we update cfg.
- // That means that masterKey update must be included in the batch.
- assert masterKeyEntry != null;
+ // Contract of meta storage ensures that all updates of one revision will come in one batch.
+ // Also masterKey should be updated every time when we update cfg.
+ // That means that masterKey update must be included in the batch.
+ assert masterKeyEntry != null;
- long newChangeId = masterKeyEntry.revision();
+ long newChangeId = masterKeyEntry.revision();
- assert newChangeId > changeId.get();
+ assert newChangeId > changeId;
- changeId.set(newChangeId);
+ changeId = newChangeId;
- try {
- lsnr.onEntriesChanged(new Data(data, newChangeId)).get();
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
+ return lsnr.onEntriesChanged(new Data(data, newChangeId));
+ }
- throw new StorageException("Interrupted when processing Meta Storage watch", e);
- } catch (ExecutionException e) {
- throw new StorageException("Error when processing Meta Storage watch", e);
- }
- }
+ @Override
+ public CompletableFuture<Void> onRevisionUpdated(long revision) {
+ assert revision > changeId;
- @Override
- public void onError(Throwable e) {
- // TODO: need to handle this case and there should some mechanism for registering new watch as far as
- // TODO: onError unregisters failed watch https://issues.apache.org/jira/browse/IGNITE-14604
- LOG.warn("Meta storage listener issue", e);
- }
- });
- } else {
- LOG.info("Configuration listener has already been set");
- }
+ return lsnr.onRevisionUpdated(revision);
+ }
+
+ @Override
+ public void onError(Throwable e) {
+ // TODO: need to handle this case and there should some mechanism for registering new watch as far as
+ // TODO: onError unregisters failed watch https://issues.apache.org/jira/browse/IGNITE-14604
+ LOG.warn("Meta storage listener issue", e);
+ }
+ });
}
/** {@inheritDoc} */
diff --git a/modules/runner/src/test/java/org/apache/ignite/internal/configuration/storage/ConfigurationStorageTest.java b/modules/runner/src/test/java/org/apache/ignite/internal/configuration/storage/ConfigurationStorageTest.java
index 84d66d3bd3..8142b8041c 100644
--- a/modules/runner/src/test/java/org/apache/ignite/internal/configuration/storage/ConfigurationStorageTest.java
+++ b/modules/runner/src/test/java/org/apache/ignite/internal/configuration/storage/ConfigurationStorageTest.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.configuration.storage;
+import static java.util.concurrent.CompletableFuture.completedFuture;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willBe;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.anEmptyMap;
@@ -46,7 +47,17 @@ public abstract class ConfigurationStorageTest {
void setUp() {
storage = getStorage();
- storage.registerConfigurationListener(data -> CompletableFuture.completedFuture(null));
+ storage.registerConfigurationListener(new ConfigurationStorageListener() {
+ @Override
+ public CompletableFuture<Void> onEntriesChanged(Data changedEntries) {
+ return completedFuture(null);
+ }
+
+ @Override
+ public CompletableFuture<Void> onRevisionUpdated(long newRevision) {
+ return completedFuture(null);
+ }
+ });
}
/**
diff --git a/modules/runner/src/test/java/org/apache/ignite/internal/configuration/storage/DistributedConfigurationCatchUpTest.java b/modules/runner/src/test/java/org/apache/ignite/internal/configuration/storage/DistributedConfigurationCatchUpTest.java
index 4ef2cc87aa..675fc85695 100644
--- a/modules/runner/src/test/java/org/apache/ignite/internal/configuration/storage/DistributedConfigurationCatchUpTest.java
+++ b/modules/runner/src/test/java/org/apache/ignite/internal/configuration/storage/DistributedConfigurationCatchUpTest.java
@@ -17,8 +17,12 @@
package org.apache.ignite.internal.configuration.storage;
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static java.util.concurrent.CompletableFuture.completedFuture;
import static org.apache.ignite.configuration.annotation.ConfigurationType.DISTRIBUTED;
-import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willBe;
+import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
+import static org.hamcrest.MatcherAssert.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyCollection;
import static org.mockito.Mockito.doAnswer;
@@ -29,6 +33,7 @@ import java.util.Collections;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;
import org.apache.ignite.configuration.RootKey;
import org.apache.ignite.configuration.annotation.ConfigurationRoot;
@@ -95,6 +100,8 @@ public class DistributedConfigurationCatchUpTest {
MetaStorageMockWrapper wrapper = new MetaStorageMockWrapper();
+ when(wrapper.mock.appliedRevision(DistributedConfigurationStorage.WATCH_ID)).thenReturn(completedFuture(0L));
+
DistributedConfigurationStorage storage = storage(wrapper);
try {
@@ -111,7 +118,7 @@ public class DistributedConfigurationCatchUpTest {
CompletableFuture<Void> change = changer.change(source);
- change.get();
+ assertThat(change, willCompleteSuccessfully());
} finally {
changer.stop();
}
@@ -123,7 +130,7 @@ public class DistributedConfigurationCatchUpTest {
vaultManager.put(MetaStorageMockWrapper.TEST_KEY, new byte[]{4, 1, 2, 3, 4}).get();
// This emulates a change in MetaStorage that is not related to the configuration.
- when(wrapper.mock.appliedRevision()).thenReturn(1L);
+ when(wrapper.mock.appliedRevision(DistributedConfigurationStorage.WATCH_ID)).thenReturn(completedFuture(2L));
storage = storage(wrapper);
@@ -136,10 +143,7 @@ public class DistributedConfigurationCatchUpTest {
changer.start();
// Should return last configuration change, not last MetaStorage change.
- Long lastConfigurationChangeRevision = storage.lastRevision().get();
-
- // Should be one, because we only changed configuration once.
- assertEquals(1, lastConfigurationChangeRevision);
+ assertThat(storage.lastRevision(), willBe(1L));
} finally {
changer.stop();
}
@@ -173,7 +177,7 @@ public class DistributedConfigurationCatchUpTest {
private WatchListener lsnr;
/** Current master key revision. */
- private long masterKeyRevision;
+ private final AtomicLong masterKeyRevision = new AtomicLong();
private MetaStorageMockWrapper() {
mock = mock(MetaStorageManager.class);
@@ -184,7 +188,7 @@ public class DistributedConfigurationCatchUpTest {
private void setup() {
// Returns current master key revision.
when(mock.get(MASTER_KEY)).then(invocation -> {
- return CompletableFuture.completedFuture(new EntryImpl(MASTER_KEY.bytes(), null, masterKeyRevision, -1));
+ return completedFuture(new EntryImpl(MASTER_KEY.bytes(), null, masterKeyRevision.get(), -1));
});
// On any invocation - trigger storage listener.
@@ -207,8 +211,13 @@ public class DistributedConfigurationCatchUpTest {
*/
private CompletableFuture<Boolean> triggerStorageListener() {
return CompletableFuture.supplyAsync(() -> {
- EntryEvent entryEvent = new EntryEvent(null, new EntryImpl(MASTER_KEY.bytes(), null, ++masterKeyRevision, -1));
- lsnr.onUpdate(new WatchEvent(entryEvent));
+ long newRevision = masterKeyRevision.incrementAndGet();
+
+ lsnr.onUpdate(new WatchEvent(List.of(
+ new EntryEvent(null, new EntryImpl(MASTER_KEY.bytes(), null, newRevision, -1)),
+ // Add a mock entry to simulate a configuration update.
+ new EntryEvent(null, new EntryImpl((DISTRIBUTED_PREFIX + "foobar").getBytes(UTF_8), null, newRevision, -1))
+ ), newRevision));
return true;
});
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
index 7315fa3a12..33ddae80fc 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
@@ -75,6 +75,7 @@ import java.util.stream.Stream;
import org.apache.ignite.configuration.ConfigurationChangeException;
import org.apache.ignite.configuration.ConfigurationProperty;
import org.apache.ignite.configuration.NamedConfigurationTree;
+import org.apache.ignite.configuration.NamedListView;
import org.apache.ignite.configuration.notifications.ConfigurationNamedListListener;
import org.apache.ignite.configuration.notifications.ConfigurationNotificationEvent;
import org.apache.ignite.internal.affinity.AffinityUtils;
@@ -100,7 +101,6 @@ import org.apache.ignite.internal.raft.RaftGroupEventsListener;
import org.apache.ignite.internal.raft.RaftManager;
import org.apache.ignite.internal.raft.RaftNodeId;
import org.apache.ignite.internal.raft.server.RaftGroupOptions;
-import org.apache.ignite.internal.raft.service.LeaderWithTerm;
import org.apache.ignite.internal.raft.service.RaftGroupListener;
import org.apache.ignite.internal.raft.service.RaftGroupService;
import org.apache.ignite.internal.raft.storage.impl.LogStorageFactoryCreator;
@@ -114,6 +114,7 @@ import org.apache.ignite.internal.schema.configuration.TableChange;
import org.apache.ignite.internal.schema.configuration.TableConfiguration;
import org.apache.ignite.internal.schema.configuration.TableView;
import org.apache.ignite.internal.schema.configuration.TablesConfiguration;
+import org.apache.ignite.internal.schema.configuration.index.TableIndexView;
import org.apache.ignite.internal.schema.event.SchemaEvent;
import org.apache.ignite.internal.schema.event.SchemaEventParameters;
import org.apache.ignite.internal.storage.DataStorageManager;
@@ -196,8 +197,8 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
private static final IntSupplier TX_STATE_STORAGE_FLUSH_DELAY_SUPPLIER = () -> TX_STATE_STORAGE_FLUSH_DELAY;
/**
- * If this property is set to {@code true} then an attempt to get the configuration property directly from Meta storage will be
- * skipped, and the local property will be returned.
+ * If this property is set to {@code true} then an attempt to get the configuration property directly from Meta storage will be skipped,
+ * and the local property will be returned.
* TODO: IGNITE-16774 This property and overall approach, access configuration directly through Meta storage,
* TODO: will be removed after fix of the issue.
*/
@@ -373,48 +374,56 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
tablesByIdVv = new VersionedValue<>(null, HashMap::new);
registry.accept(token -> {
- List<CompletableFuture<?>> futures = new ArrayList<>(beforeTablesVvComplete);
-
- beforeTablesVvComplete.clear();
-
- return allOf(futures.toArray(new CompletableFuture[]{}))
- .orTimeout(TABLES_COMPLETE_TIMEOUT, TimeUnit.SECONDS)
- .whenComplete((v, e) -> {
- if (!busyLock.enterBusy()) {
- if (e != null) {
- LOG.warn("Error occurred while updating tables and stopping components.", e);
- // Stop of the components has been started, so we do nothing and resources of tablesByIdVv will be
- // freed in the logic of TableManager stop. We cannot complete tablesByIdVv exceptionally because
- // we will lose a context of tables.
+ CompletableFuture<Void> beforeTablesVvCompleteFuture;
+
+ if (beforeTablesVvComplete.isEmpty()) {
+ beforeTablesVvCompleteFuture = completedFuture(null);
+ } else {
+ CompletableFuture<?>[] futures = beforeTablesVvComplete.toArray(new CompletableFuture[0]);
+
+ beforeTablesVvComplete.clear();
+
+ beforeTablesVvCompleteFuture = allOf(futures).orTimeout(TABLES_COMPLETE_TIMEOUT, TimeUnit.SECONDS);
+ }
+
+ return beforeTablesVvCompleteFuture.whenComplete((v, e) -> {
+ if (!busyLock.enterBusy()) {
+ if (e != null) {
+ LOG.warn("Error occurred while updating tables and stopping components.", e);
+ // Stop of the components has been started, so we do nothing and resources of tablesByIdVv will be
+ // freed in the logic of TableManager stop. We cannot complete tablesByIdVv exceptionally because
+ // we will lose a context of tables.
+ }
+ return;
+ }
+ try {
+ if (e != null) {
+ LOG.warn("Error occurred while updating tables.", e);
+ if (e instanceof CompletionException) {
+ Throwable th = e.getCause();
+ // Case when stopping of the previous component has been started and related futures completed
+ // exceptionally
+ if (th instanceof NodeStoppingException || (th.getCause() != null
+ && th.getCause() instanceof NodeStoppingException)) {
+ // Stop of the components has been started so we do nothing and resources will be freed in the
+ // logic of TableManager stop
+ return;
}
- return;
}
- try {
- if (e != null) {
- LOG.warn("Error occurred while updating tables.", e);
- if (e instanceof CompletionException) {
- Throwable th = e.getCause();
- // Case when stopping of the previous component has been started and related futures completed
- // exceptionally
- if (th instanceof NodeStoppingException || (th.getCause() != null
- && th.getCause() instanceof NodeStoppingException)) {
- // Stop of the components has been started so we do nothing and resources will be freed in the
- // logic of TableManager stop
- return;
- }
- }
- // TODO: https://issues.apache.org/jira/browse/IGNITE-17515
- tablesByIdVv.completeExceptionally(token, e);
- }
+ // TODO: https://issues.apache.org/jira/browse/IGNITE-17515
+ tablesByIdVv.completeExceptionally(token, e);
+
+ return;
+ }
- //Normal scenario, when all related futures for tablesByIdVv are completed and we can complete tablesByIdVv
- tablesByIdVv.complete(token);
+ //Normal scenario, when all related futures for tablesByIdVv are completed and we can complete tablesByIdVv
+ tablesByIdVv.complete(token);
- tablesToStopInCaseOfError.clear();
- } finally {
- busyLock.leaveBusy();
- }
- });
+ tablesToStopInCaseOfError.clear();
+ } finally {
+ busyLock.leaveBusy();
+ }
+ });
});
txStateStorageScheduledPool = Executors.newSingleThreadScheduledExecutor(
@@ -727,6 +736,7 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
Set<Assignment> newPartAssignment = newAssignments.get(partId);
TableImpl table = tablesById.get(tblId);
+
InternalTable internalTbl = table.internalTable();
Assignment localMemberAssignment = newPartAssignment.stream()
@@ -1513,48 +1523,41 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
/** See {@link #dropTableAsync(String)} for details. */
private CompletableFuture<Void> dropTableAsyncInternal(String name) {
- CompletableFuture<Void> dropTblFut = new CompletableFuture<>();
-
- tableAsyncInternal(name).thenAccept(tbl -> {
+ return tableAsyncInternal(name).thenCompose(tbl -> {
// In case of drop it's an optimization that allows not to fire drop-change-closure if there's no such
// distributed table and the local config has lagged behind.
if (tbl == null) {
- dropTblFut.completeExceptionally(new TableNotFoundException(DEFAULT_SCHEMA_NAME, name));
- } else {
- tablesCfg.change(chg ->
- chg.changeTables(tblChg -> {
- TableView tableCfg = tblChg.get(name);
+ return failedFuture(new TableNotFoundException(DEFAULT_SCHEMA_NAME, name));
+ }
- if (tableCfg == null) {
- throw new TableNotFoundException(DEFAULT_SCHEMA_NAME, name);
- }
+ return tablesCfg
+ .change(chg -> chg
+ .changeTables(tblChg -> {
+ if (tblChg.get(name) == null) {
+ throw new TableNotFoundException(DEFAULT_SCHEMA_NAME, name);
+ }
- tblChg.delete(name);
- }).changeIndexes(idxChg -> {
- List<String> indicesNames = tablesCfg.indexes().value().namedListKeys();
+ tblChg.delete(name);
+ })
+ .changeIndexes(idxChg -> {
+ NamedListView<? extends TableIndexView> indexes = chg.indexes();
- indicesNames.stream().filter(idx ->
- tablesCfg.indexes().get(idx).tableId().value().equals(tbl.tableId())).forEach(idxChg::delete);
- }))
- .whenComplete((res, t) -> {
- if (t != null) {
- Throwable ex = getRootCause(t);
+ for (String indexName : indexes.namedListKeys()) {
+ if (indexes.get(indexName).tableId().equals(tbl.tableId())) {
+ idxChg.delete(indexName);
+ }
+ }
+ }))
+ .exceptionally(t -> {
+ Throwable ex = getRootCause(t);
- if (ex instanceof TableNotFoundException) {
- dropTblFut.completeExceptionally(ex);
- } else {
- LOG.debug("Unable to drop table [name={}]", ex, name);
+ if (!(ex instanceof TableNotFoundException)) {
+ LOG.debug("Unable to drop table [name={}]", ex, name);
+ }
- dropTblFut.completeExceptionally(ex);
- }
- } else {
- dropTblFut.complete(res);
- }
- });
- }
+ throw new CompletionException(ex);
+ });
});
-
- return dropTblFut;
}
/** {@inheritDoc} */
@@ -1857,9 +1860,14 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
private WatchListener createDistributionZonesDataNodesListener() {
return new WatchListener() {
@Override
- public void onUpdate(WatchEvent evt) {
+ public String id() {
+ return "dst-zones-data-nodes-watch";
+ }
+
+ @Override
+ public CompletableFuture<Void> onUpdate(WatchEvent evt) {
if (!busyLock.enterBusy()) {
- throw new IgniteInternalException(new NodeStoppingException());
+ return failedFuture(new NodeStoppingException());
}
try {
@@ -1867,7 +1875,7 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
if (dataNodesBytes == null) {
//The zone was removed so data nodes was removed too.
- return;
+ return completedFuture(null);
}
NamedConfigurationTree<TableConfiguration, TableView, TableChange> tables = tablesCfg.tables();
@@ -1911,6 +1919,8 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
}
}
}
+
+ return completedFuture(null);
} finally {
busyLock.leaveBusy();
}
@@ -1931,74 +1941,116 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
private WatchListener createPendingAssignmentsRebalanceListener() {
return new WatchListener() {
@Override
- public void onUpdate(WatchEvent evt) {
+ public String id() {
+ return "pending-assignments-watch";
+ }
+
+ @Override
+ public CompletableFuture<Void> onUpdate(WatchEvent evt) {
if (!busyLock.enterBusy()) {
- throw new IgniteInternalException(new NodeStoppingException());
+ return failedFuture(new NodeStoppingException());
}
try {
- assert evt.single();
+ return handleChangePendingAssignmentEvent(evt);
+ } finally {
+ busyLock.leaveBusy();
+ }
+ }
+
+ @Override
+ public void onError(Throwable e) {
+ LOG.warn("Unable to process pending assignments event", e);
+ }
+ };
+ }
- Entry pendingAssignmentsWatchEvent = evt.entryEvent().newEntry();
+ private CompletableFuture<Void> handleChangePendingAssignmentEvent(WatchEvent evt) {
+ assert evt.single();
- if (pendingAssignmentsWatchEvent.value() == null) {
- return;
- }
+ Entry pendingAssignmentsWatchEntry = evt.entryEvent().newEntry();
+
+ if (pendingAssignmentsWatchEntry.value() == null) {
+ return completedFuture(null);
+ }
+
+ int partId = extractPartitionNumber(pendingAssignmentsWatchEntry.key());
+ UUID tblId = extractTableId(pendingAssignmentsWatchEntry.key(), PENDING_ASSIGNMENTS_PREFIX);
- int partId = extractPartitionNumber(pendingAssignmentsWatchEvent.key());
- UUID tblId = extractTableId(pendingAssignmentsWatchEvent.key(), PENDING_ASSIGNMENTS_PREFIX);
+ var replicaGrpId = new TablePartitionId(tblId, partId);
- TablePartitionId replicaGrpId = new TablePartitionId(tblId, partId);
+ // Stable assignments from the meta store, which revision is bounded by the current pending event.
+ CompletableFuture<Entry> stableAssignmentsFuture = metaStorageMgr.get(stablePartAssignmentsKey(replicaGrpId), evt.revision());
- Entry pendingAssignmentsEntry = metaStorageMgr.get(pendingPartAssignmentsKey(replicaGrpId)).join();
+ return tablesByIdVv.get(evt.revision())
+ .thenCombineAsync(stableAssignmentsFuture, (tables, stableAssignmentsEntry) -> {
+ if (!busyLock.enterBusy()) {
+ return CompletableFuture.<Void>failedFuture(new NodeStoppingException());
+ }
- assert pendingAssignmentsWatchEvent.revision() <= pendingAssignmentsEntry.revision()
- : "Meta Storage watch cannot notify about an event with the revision that is more than the actual revision.";
+ try {
+ return handleChangePendingAssignmentEvent(
+ replicaGrpId,
+ tables.get(tblId),
+ pendingAssignmentsWatchEntry,
+ stableAssignmentsEntry
+ );
+ } finally {
+ busyLock.leaveBusy();
+ }
+ }, ioExecutor)
+ .thenCompose(Function.identity());
+ }
- // Assignments of the pending rebalance that we received through Meta storage watch mechanism.
- Set<Assignment> pendingAssignments = ByteUtils.fromBytes(pendingAssignmentsWatchEvent.value());
+ private CompletableFuture<Void> handleChangePendingAssignmentEvent(
+ TablePartitionId replicaGrpId,
+ TableImpl tbl,
+ Entry pendingAssignmentsEntry,
+ Entry stableAssignmentsEntry
+ ) {
+ // Assignments of the pending rebalance that we received through the Meta storage watch mechanism.
+ Set<Assignment> pendingAssignments = ByteUtils.fromBytes(pendingAssignmentsEntry.value());
- PeersAndLearners pendingConfiguration = configurationFromAssignments(pendingAssignments);
+ PeersAndLearners pendingConfiguration = configurationFromAssignments(pendingAssignments);
- TableImpl tbl = tablesByIdVv.latest().get(tblId);
+ ExtendedTableConfiguration tblCfg = (ExtendedTableConfiguration) tablesCfg.tables().get(tbl.name());
- ExtendedTableConfiguration tblCfg = (ExtendedTableConfiguration) tablesCfg.tables().get(tbl.name());
+ int partId = replicaGrpId.partitionId();
- // Stable assignments from the meta store, which revision is bounded by the current pending event.
- byte[] stableAssignmentsBytes = metaStorageMgr.get(stablePartAssignmentsKey(replicaGrpId),
- pendingAssignmentsWatchEvent.revision()).join().value();
+ byte[] stableAssignmentsBytes = stableAssignmentsEntry.value();
- Set<Assignment> stableAssignments = stableAssignmentsBytes == null
- // This is for the case when the first rebalance occurs.
- ? ((List<Set<Assignment>>) ByteUtils.fromBytes(tblCfg.assignments().value())).get(partId)
- : ByteUtils.fromBytes(stableAssignmentsBytes);
+ Set<Assignment> stableAssignments = stableAssignmentsBytes == null
+ // This is for the case when the first rebalance occurs.
+ ? ((List<Set<Assignment>>) ByteUtils.fromBytes(tblCfg.assignments().value())).get(partId)
+ : ByteUtils.fromBytes(stableAssignmentsBytes);
- PeersAndLearners stableConfiguration = configurationFromAssignments(stableAssignments);
+ PeersAndLearners stableConfiguration = configurationFromAssignments(stableAssignments);
- placementDriver.updateAssignment(
- replicaGrpId,
- stableConfiguration.peers().stream().map(Peer::consistentId).collect(toList())
- );
+ placementDriver.updateAssignment(
+ replicaGrpId,
+ stableConfiguration.peers().stream().map(Peer::consistentId).collect(toList())
+ );
- ClusterNode localMember = clusterService.topologyService().localMember();
+ ClusterNode localMember = clusterService.topologyService().localMember();
- // Start a new Raft node and Replica if this node has appeared in the new assignments.
- boolean shouldStartLocalServices = pendingAssignments.stream()
- .filter(assignment -> localMember.name().equals(assignment.consistentId()))
- .anyMatch(assignment -> !stableAssignments.contains(assignment));
+ // Start a new Raft node and Replica if this node has appeared in the new assignments.
+ boolean shouldStartLocalServices = pendingAssignments.stream()
+ .filter(assignment -> localMember.name().equals(assignment.consistentId()))
+ .anyMatch(assignment -> !stableAssignments.contains(assignment));
- PendingComparableValuesTracker<HybridTimestamp> safeTime =
- new PendingComparableValuesTracker<>(new HybridTimestamp(1, 0));
+ var safeTime = new PendingComparableValuesTracker<>(new HybridTimestamp(1, 0));
- InternalTable internalTable = tbl.internalTable();
+ InternalTable internalTable = tbl.internalTable();
- LOG.info("Received update on pending assignments. Check if new raft group should be started"
- + " [key={}, partition={}, table={}, localMemberAddress={}]",
- pendingAssignmentsWatchEvent.key(), partId, tbl.name(), localMember.address());
+ LOG.info("Received update on pending assignments. Check if new raft group should be started"
+ + " [key={}, partition={}, table={}, localMemberAddress={}]",
+ pendingAssignmentsEntry.key(), partId, tbl.name(), localMember.address());
- if (shouldStartLocalServices) {
- PartitionStorages partitionStorages = getOrCreatePartitionStorages(tbl, partId).join();
+ CompletableFuture<Void> localServicesStartFuture;
+ if (shouldStartLocalServices) {
+ localServicesStartFuture = getOrCreatePartitionStorages(tbl, partId)
+ .thenAcceptAsync(partitionStorages -> {
MvPartitionStorage mvPartitionStorage = partitionStorages.getMvPartitionStorage();
TxStateStorage txStatePartitionStorage = partitionStorages.getTxStateStorage();
@@ -2031,7 +2083,7 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
partId,
busyLock,
createPartitionMover(internalTable, partId),
- TableManager.this::calculateAssignments,
+ this::calculateAssignments,
rebalanceScheduler
);
@@ -2049,6 +2101,8 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
groupOptions
);
+ UUID tblId = tbl.tableId();
+
replicaMgr.startReplica(replicaGrpId,
new PartitionReplicaListener(
mvPartitionStorage,
@@ -2066,43 +2120,43 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
txStatePartitionStorage,
placementDriver,
storageUpdateHandler,
- TableManager.this::isLocalPeer,
+ this::isLocalPeer,
completedFuture(schemaManager.schemaRegistry(tblId))
)
);
- } catch (NodeStoppingException e) {
+ } catch (NodeStoppingException ignored) {
// no-op
}
- }
+ }, ioExecutor);
+ } else {
+ localServicesStartFuture = completedFuture(null);
+ }
+ return localServicesStartFuture
+ .thenCompose(v -> metaStorageMgr.get(pendingPartAssignmentsKey(replicaGrpId)))
+ .thenCompose(latestPendingAssignmentsEntry -> {
// Do not change peers of the raft group if this is a stale event.
// Note that we start raft node before for the sake of the consistency in a starting and stopping raft nodes.
- if (pendingAssignmentsWatchEvent.revision() < pendingAssignmentsEntry.revision()) {
- return;
+ if (pendingAssignmentsEntry.revision() < latestPendingAssignmentsEntry.revision()) {
+ return completedFuture(null);
}
RaftGroupService partGrpSvc = internalTable.partitionRaftGroupService(partId);
- LeaderWithTerm leaderWithTerm = partGrpSvc.refreshAndGetLeaderWithTerm().join();
-
- // run update of raft configuration if this node is a leader
- if (isLocalPeer(leaderWithTerm.leader())) {
- LOG.info("Current node={} is the leader of partition raft group={}. "
- + "Initiate rebalance process for partition={}, table={}",
- localMember.address(), replicaGrpId, partId, tbl.name());
-
- partGrpSvc.changePeersAsync(pendingConfiguration, leaderWithTerm.term()).join();
- }
- } finally {
- busyLock.leaveBusy();
- }
- }
+ return partGrpSvc.refreshAndGetLeaderWithTerm()
+ .thenCompose(leaderWithTerm -> {
+ // run update of raft configuration if this node is a leader
+ if (isLocalPeer(leaderWithTerm.leader())) {
+ LOG.info("Current node={} is the leader of partition raft group={}. "
+ + "Initiate rebalance process for partition={}, table={}",
+ localMember.address(), replicaGrpId, partId, tbl.name());
- @Override
- public void onError(Throwable e) {
- LOG.warn("Unable to process pending assignments event", e);
- }
- };
+ return partGrpSvc.changePeersAsync(pendingConfiguration, leaderWithTerm.term());
+ } else {
+ return completedFuture(null);
+ }
+ });
+ });
}
/**
@@ -2113,8 +2167,21 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
private WatchListener createStableAssignmentsRebalanceListener() {
return new WatchListener() {
@Override
- public void onUpdate(WatchEvent evt) {
- handleChangeStableAssignmentEvent(evt);
+ public String id() {
+ return "stable-assignments-watch";
+ }
+
+ @Override
+ public CompletableFuture<Void> onUpdate(WatchEvent evt) {
+ if (!busyLock.enterBusy()) {
+ return failedFuture(new NodeStoppingException());
+ }
+
+ try {
+ return handleChangeStableAssignmentEvent(evt);
+ } finally {
+ busyLock.leaveBusy();
+ }
}
@Override
@@ -2132,26 +2199,50 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
private WatchListener createAssignmentsSwitchRebalanceListener() {
return new WatchListener() {
@Override
- public void onUpdate(WatchEvent evt) {
- byte[] key = evt.entryEvent().newEntry().key();
+ public String id() {
+ return "assignments-switch-watch";
+ }
- int partitionNumber = extractPartitionNumber(key);
- UUID tblId = extractTableId(key, ASSIGNMENTS_SWITCH_REDUCE_PREFIX);
+ @Override
+ public CompletableFuture<Void> onUpdate(WatchEvent evt) {
+ if (!busyLock.enterBusy()) {
+ return failedFuture(new NodeStoppingException());
+ }
- TablePartitionId replicaGrpId = new TablePartitionId(tblId, partitionNumber);
+ try {
+ byte[] key = evt.entryEvent().newEntry().key();
- TableImpl tbl = tablesByIdVv.latest().get(tblId);
+ int partitionNumber = extractPartitionNumber(key);
+ UUID tblId = extractTableId(key, ASSIGNMENTS_SWITCH_REDUCE_PREFIX);
- TableConfiguration tblCfg = tablesCfg.tables().get(tbl.name());
+ TablePartitionId replicaGrpId = new TablePartitionId(tblId, partitionNumber);
- RebalanceUtil.handleReduceChanged(
- metaStorageMgr,
- baselineMgr.nodes().stream().map(ClusterNode::name).collect(toList()),
- tblCfg.value().replicas(),
- partitionNumber,
- replicaGrpId,
- evt
- );
+ return tablesByIdVv.get(evt.revision())
+ .thenCompose(tables -> {
+ if (!busyLock.enterBusy()) {
+ return failedFuture(new NodeStoppingException());
+ }
+
+ try {
+ TableImpl tbl = tables.get(tblId);
+
+ TableConfiguration tblCfg = tablesCfg.tables().get(tbl.name());
+
+ return RebalanceUtil.handleReduceChanged(
+ metaStorageMgr,
+ baselineMgr.nodes().stream().map(ClusterNode::name).collect(toList()),
+ tblCfg.value().replicas(),
+ partitionNumber,
+ replicaGrpId,
+ evt
+ );
+ } finally {
+ busyLock.leaveBusy();
+ }
+ });
+ } finally {
+ busyLock.leaveBusy();
+ }
}
@Override
@@ -2228,59 +2319,59 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
*
* @param evt Event.
*/
- protected void handleChangeStableAssignmentEvent(WatchEvent evt) {
- inBusyLock(busyLock, () -> {
- assert evt.single() : evt;
-
- Entry stableAssignmentsWatchEvent = evt.entryEvent().newEntry();
+ protected CompletableFuture<Void> handleChangeStableAssignmentEvent(WatchEvent evt) {
+ assert evt.single() : evt;
- if (stableAssignmentsWatchEvent.value() == null) {
- return;
- }
+ Entry stableAssignmentsWatchEvent = evt.entryEvent().newEntry();
- int partitionId = extractPartitionNumber(stableAssignmentsWatchEvent.key());
- UUID tableId = extractTableId(stableAssignmentsWatchEvent.key(), STABLE_ASSIGNMENTS_PREFIX);
+ if (stableAssignmentsWatchEvent.value() == null) {
+ return completedFuture(null);
+ }
- TablePartitionId tablePartitionId = new TablePartitionId(tableId, partitionId);
+ int partitionId = extractPartitionNumber(stableAssignmentsWatchEvent.key());
+ UUID tableId = extractTableId(stableAssignmentsWatchEvent.key(), STABLE_ASSIGNMENTS_PREFIX);
- Set<Assignment> stableAssignments = ByteUtils.fromBytes(stableAssignmentsWatchEvent.value());
+ TablePartitionId tablePartitionId = new TablePartitionId(tableId, partitionId);
- byte[] pendingAssignmentsFromMetaStorage = metaStorageMgr.get(
- pendingPartAssignmentsKey(tablePartitionId),
- stableAssignmentsWatchEvent.revision()
- ).join().value();
+ Set<Assignment> stableAssignments = ByteUtils.fromBytes(stableAssignmentsWatchEvent.value());
- Set<Assignment> pendingAssignments = pendingAssignmentsFromMetaStorage == null
- ? Set.of()
- : ByteUtils.fromBytes(pendingAssignmentsFromMetaStorage);
+ return metaStorageMgr.get(pendingPartAssignmentsKey(tablePartitionId), stableAssignmentsWatchEvent.revision())
+ .thenComposeAsync(pendingAssignmentsEntry -> {
+ byte[] pendingAssignmentsFromMetaStorage = pendingAssignmentsEntry.value();
- String localMemberName = clusterService.topologyService().localMember().name();
+ Set<Assignment> pendingAssignments = pendingAssignmentsFromMetaStorage == null
+ ? Set.of()
+ : ByteUtils.fromBytes(pendingAssignmentsFromMetaStorage);
- boolean shouldStopLocalServices = Stream.concat(stableAssignments.stream(), pendingAssignments.stream())
- .noneMatch(assignment -> assignment.consistentId().equals(localMemberName));
+ String localMemberName = clusterService.topologyService().localMember().name();
- if (shouldStopLocalServices) {
- try {
- raftMgr.stopRaftNodes(tablePartitionId);
+ boolean shouldStopLocalServices = Stream.concat(stableAssignments.stream(), pendingAssignments.stream())
+ .noneMatch(assignment -> assignment.consistentId().equals(localMemberName));
- replicaMgr.stopReplica(tablePartitionId);
- } catch (NodeStoppingException e) {
- // no-op
- }
+ if (!shouldStopLocalServices) {
+ return completedFuture(null);
+ }
- InternalTable internalTable = tablesByIdVv.latest().get(tableId).internalTable();
+ try {
+ raftMgr.stopRaftNodes(tablePartitionId);
- // TODO: IGNITE-18703 Destroy raft log and meta
+ replicaMgr.stopReplica(tablePartitionId);
+ } catch (NodeStoppingException e) {
+ // no-op
+ }
- // Should be done fairly quickly.
- mvGc.removeStorage(tablePartitionId)
- .thenCompose(unused -> allOf(
- internalTable.storage().destroyPartition(partitionId),
- runAsync(() -> internalTable.txStateStorage().destroyTxStateStorage(partitionId), ioExecutor)
- ))
- .join();
- }
- });
+ return tablesByIdVv.get(evt.revision())
+ // TODO: IGNITE-18703 Destroy raft log and meta
+ .thenCombine(mvGc.removeStorage(tablePartitionId), (tables, unused) -> {
+ InternalTable internalTable = tables.get(tableId).internalTable();
+
+ return allOf(
+ internalTable.storage().destroyPartition(partitionId),
+ runAsync(() -> internalTable.txStateStorage().destroyTxStateStorage(partitionId), ioExecutor)
+ );
+ })
+ .thenCompose(Function.identity());
+ }, ioExecutor);
}
private static void handleExceptionOnCleanUpTablesResources(