sashapolo commented on code in PR #2266:
URL: https://github.com/apache/ignite-3/pull/2266#discussion_r1251868940
##########
modules/runner/src/main/java/org/apache/ignite/internal/configuration/storage/DistributedConfigurationStorage.java:
##########
@@ -206,60 +194,44 @@ public CompletableFuture<Serializable> readLatest(String
key) {
@Override
public CompletableFuture<Data> readDataOnRecovery() throws
StorageException {
- CompletableFuture<Data> future =
vaultMgr.get(CONFIGURATION_REVISIONS_KEY)
- .thenApplyAsync(entry -> {
- long revision =
resolveRevision(metaStorageMgr.appliedRevision(), entry);
-
- return readDataOnRecovery0(revision);
- }, threadPool);
+ CompletableFuture<Data> future =
metaStorageMgr.recoveryFinishedFuture()
+ .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.
- *
- * @param metaStorageRevision Meta Storage revision.
- * @param revisionsEntry Configuration revisions entry.
- * @return Configuration revision.
- */
- private static long resolveRevision(long metaStorageRevision, @Nullable
VaultEntry revisionsEntry) {
- if (revisionsEntry != null) {
- byte[] value = revisionsEntry.value();
- long prevMasterKeyRevision = ByteUtils.bytesToLong(value, 0);
- long curMasterKeyRevision = ByteUtils.bytesToLong(value,
Long.BYTES);
-
- // If current master key revision is higher than applied revision,
then node failed
- // before applied revision changed, so we have to use previous
master key revision
- return curMasterKeyRevision <= metaStorageRevision ?
curMasterKeyRevision : prevMasterKeyRevision;
- } else {
- // Configuration has not been updated yet, so it is safe to return
0 as the revision for the master key.
- return 0L;
- }
- }
-
private Data readDataOnRecovery0(long cfgRevision) {
var data = new HashMap<String, Serializable>();
- try (Cursor<VaultEntry> entries = storedDistributedConfigKeys()) {
- for (VaultEntry entry : entries) {
- ByteArray key = entry.key();
+ byte[] masterKey = MASTER_KEY.bytes();
+ boolean sawMasterKey = false;
+
+ try (Cursor<Entry> cursor =
metaStorageMgr.getLocally(DST_KEYS_START_RANGE, DST_KEYS_END_RANGE,
cfgRevision)) {
+ for (Entry entry : cursor) {
+ if (entry.tombstone()) {
+ continue;
+ }
+
+ byte[] key = entry.key();
byte[] value = entry.value();
- // vault iterator should not return nulls as values
+ // MetaStorage iterator should not return nulls as values.
assert value != null;
- if (key.equals(MASTER_KEY)) {
+ if (!sawMasterKey && Arrays.equals(masterKey, key)) {
+ sawMasterKey = true;
+
continue;
}
- String dataKey =
key.toString().substring(DISTRIBUTED_PREFIX.length());
+ byte[] keyWithoutPrefix = Arrays.copyOfRange(key,
DST_KEYS_START_RANGE.length(), key.length);
Review Comment:
Wasn't there a `String` constructor that allows creating strings using an
offset and length?
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/storage/ItRebalanceDistributedTest.java:
##########
@@ -256,6 +261,37 @@ void after() {
}
}
+ // TODO: https://issues.apache.org/jira/browse/IGNITE-19506 Remove.
Review Comment:
This test is disabled under the same ticket, why is this code needed?
##########
modules/runner/src/main/java/org/apache/ignite/internal/configuration/storage/DistributedConfigurationStorage.java:
##########
@@ -296,9 +268,9 @@ public CompletableFuture<Boolean> write(Map<String, ?
extends Serializable> newV
operations.add(Operations.put(MASTER_KEY,
ByteUtils.longToBytes(curChangeId)));
- SimpleCondition condition = curChangeId == 0L
- ? Conditions.notExists(MASTER_KEY)
- : Conditions.revision(MASTER_KEY).eq(curChangeId);
+ Condition condition = curChangeId == 0L
+ ? notExists(MASTER_KEY)
+ : or(notExists(MASTER_KEY),
revision(MASTER_KEY).le(curChangeId));
Review Comment:
Why did the condition change here?
##########
modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java:
##########
@@ -960,44 +957,16 @@ public void init(
* and deploying watches after that.
*/
private CompletableFuture<?> recoverComponentsStateOnStart(ExecutorService
startupExecutor) {
- // Recovery future must be created before configuration listeners are
triggered.
- CompletableFuture<?> recoveryFuture =
RecoveryCompletionFutureFactory.create(
- clusterCfgMgr,
- fut -> new ConfigurationCatchUpListener(cfgStorage, fut, LOG)
- );
-
- //TODO https://issues.apache.org/jira/browse/IGNITE-19778
- // The order of these two lines matter, the first method relies on the
second one not being called yet.
- // After the fix, the order will most likely have to be reversed.
- CompletableFuture<Void> startupRevisionUpdate =
notifyRevisionUpdateListenerOnStart();
CompletableFuture<Void> startupConfigurationUpdate =
notifyConfigurationListeners();
+ CompletableFuture<Void> startupRevisionUpdate =
((MetaStorageManagerImpl) metaStorageMgr).notifyRevisionUpdateListenerOnStart();
Review Comment:
Why do you need to cast to `MetaStorageManagerImpl`?
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/ItDistributedConfigurationPropertiesTest.java:
##########
@@ -228,7 +228,7 @@ void start() {
Stream.of(clusterService, raftManager, cmgManager,
metaStorageManager)
.forEach(IgniteComponent::start);
- distributedCfgManager.start();
+ CompletableFuture.runAsync(distributedCfgManager::start);
Review Comment:
What's this for?
--
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]
