ibessonov commented on code in PR #2266:
URL: https://github.com/apache/ignite-3/pull/2266#discussion_r1252140674
##########
modules/metastorage-api/src/main/java/org/apache/ignite/internal/metastorage/MetaStorageManager.java:
##########
@@ -84,7 +85,18 @@ public interface MetaStorageManager extends IgniteComponent {
* @param revUpperBound The upper bound of revision.
* @return Value corresponding to the given key.
*/
- Entry getLocally(byte[] key, long revUpperBound);
+ Entry getLocally(ByteArray key, long revUpperBound);
+
+ /**
+ * Returns cursor by entries which correspond to the given keys range and
bounded by revision number. The entries in the cursor
+ * are obtained from the local storage.
+ *
+ * @param startKey Start key of range (inclusive).
+ * @param endKey Last key of range (exclusive).
+ * @param revision Upper bound of revision.
+ * @return Cursor by entries which correspond to the given keys range.
+ */
+ Cursor<Entry> getLocally(ByteArray startKey, ByteArray endKey, long
revision);
Review Comment:
What happens if revision is too big? Will we just ignore it and read
(potentially) invalid data, or will we wait for the corresponding update?
Please document how it works, it'll only take a line (here and in previous
method).
##########
modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/impl/ItMetaStorageWatchTest.java:
##########
@@ -218,6 +219,12 @@ private void startCluster(int size) throws
NodeStoppingException {
String name = nodes.get(0).name();
nodes.get(0).cmgManager.initCluster(List.of(name), List.of(name),
"test");
+
+ CompletableFuture<Void> startFut = allOf(
Review Comment:
Why not simply make a loop? You don't have to create new future. But that's
just a question
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/storage/ItRebalanceDistributedTest.java:
##########
@@ -863,40 +900,51 @@ protected CompletableFuture<Void>
handleChangeStableAssignmentEvent(WatchEvent e
* Starts the created components.
*/
void start() {
- nodeComponents = List.of(
+ nodeComponents = new CopyOnWriteArrayList<>();
+
+ List<IgniteComponent> firstComponents = List.of(
vaultManager,
nodeCfgMgr,
clusterService,
- raftManager,
- cmgManager,
- metaStorageManager,
- clusterCfgMgr,
- clockWaiter,
- catalogManager,
- distributionZoneManager,
- replicaManager,
- txManager,
- baselineMgr,
- dataStorageMgr,
- schemaManager,
- tableManager
+ raftManager
);
- nodeComponents.forEach(IgniteComponent::start);
+ firstComponents.forEach(IgniteComponent::start);
+
+ nodeComponents.addAll(firstComponents);
+
+ deployWatchesFut = CompletableFuture.supplyAsync(() -> {
+ List<IgniteComponent> secondComponents = List.of(
+ cmgManager,
+ metaStorageManager,
+ clusterCfgMgr,
+ clockWaiter,
+ catalogManager,
+ distributionZoneManager,
+ replicaManager,
+ txManager,
+ baselineMgr,
+ dataStorageMgr,
+ schemaManager,
+ tableManager
+ );
- assertThat(
- allOf(
+ secondComponents.forEach(IgniteComponent::start);
+
+ nodeComponents.addAll(secondComponents);
+
+ var configurationNotificationFut =
metaStorageManager.recoveryFinishedFuture().thenCompose(rev -> {
+ return allOf(
nodeCfgMgr.configurationRegistry().notifyCurrentConfigurationListeners(),
clusterCfgMgr.configurationRegistry().notifyCurrentConfigurationListeners(),
- // Why "-1"? I don't know, it just works like that.
- ((MetaStorageManagerImpl)
metaStorageManager).notifyRevisionUpdateListenerOnStart(
- metaStorageManager.appliedRevision() - 1
- )
- ),
- willSucceedIn(1, TimeUnit.MINUTES)
- );
+ ((MetaStorageManagerImpl)
metaStorageManager).notifyRevisionUpdateListenerOnStart()
+ );
+ });
+
+ assertThat(configurationNotificationFut, willSucceedIn(1,
TimeUnit.MINUTES));
- deployWatchesFut = metaStorageManager.deployWatches();
+ return metaStorageManager.deployWatches();
+ }).thenCompose(fut -> fut);
Review Comment:
Please use `Function.identity()`, maybe with a static import
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/BaseIgniteRestartTest.java:
##########
@@ -241,32 +236,17 @@ public static PartialNode partialNode(
) {
AtomicLong lastRevision = new AtomicLong();
- Consumer<Long> revisionCallback0 = rev -> {
- if (revisionCallback != null) {
- revisionCallback.accept(rev);
- }
-
- lastRevision.set(rev);
- };
-
- CompletableFuture<Void> configurationCatchUpFuture =
RecoveryCompletionFutureFactory.create(
- clusterCfgMgr,
- fut -> new TestConfigurationCatchUpListener(cfgStorage, fut,
revisionCallback0)
- );
-
CompletableFuture<?> startFuture = CompletableFuture.allOf(
nodeCfgMgr.configurationRegistry().notifyCurrentConfigurationListeners(),
clusterConfigRegistry.notifyCurrentConfigurationListeners()
).thenCompose(unused ->
// Deploy all registered watches because all components are
ready and have registered their listeners.
- CompletableFuture.allOf(metaStorageMgr.deployWatches(),
configurationCatchUpFuture)
+ metaStorageMgr.deployWatches()
);
assertThat("Partial node was not started", startFuture,
willCompleteSuccessfully());
- log.info("Completed recovery on partially started node, last revision
applied: " + lastRevision.get()
- + ", acceptableDifference: " +
IgniteSystemProperties.getInteger(CONFIGURATION_CATCH_UP_DIFFERENCE_PROPERTY,
100)
- );
+ log.info("Completed recovery on partially started node, last revision
applied: " + lastRevision.get());
Review Comment:
I would change the "last applied revision" part, and use a different
terminology
##########
modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/impl/MetaStorageManagerImpl.java:
##########
@@ -824,19 +830,7 @@ private CompletableFuture<Void>
onRevisionApplied(WatchEvent watchEvent, HybridT
clusterTime.updateSafeTime(time);
try {
- CompletableFuture<Void> saveToVaultFuture;
-
- if (watchEvent.entryEvents().isEmpty()) {
- saveToVaultFuture = vaultMgr.put(APPLIED_REV_KEY,
longToBytes(watchEvent.revision()));
- } else {
- Map<ByteArray, byte[]> batch =
IgniteUtils.newHashMap(watchEvent.entryEvents().size() + 1);
-
- batch.put(APPLIED_REV_KEY, longToBytes(watchEvent.revision()));
-
- watchEvent.entryEvents().forEach(e -> batch.put(new
ByteArray(e.newEntry().key()), e.newEntry().value()));
-
- saveToVaultFuture = vaultMgr.putAll(batch);
- }
+ CompletableFuture<Void> saveToVaultFuture =
vaultMgr.put(APPLIED_REV_KEY, longToBytes(watchEvent.revision()));
Review Comment:
Is there anyone who reads the value? Why would they need it?
Can we remove it completely?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
