You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by GitBox <gi...@apache.org> on 2022/11/11 10:58:10 UTC
[GitHub] [ignite-3] sashapolo commented on a diff in pull request #1331: IGNITE-18122 Track last applied term and group config in storages
sashapolo commented on code in PR #1331:
URL: https://github.com/apache/ignite-3/pull/1331#discussion_r1020042717
##########
modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/PartitionMetaTest.java:
##########
@@ -41,15 +44,45 @@ void testLastAppliedIndex() {
assertEquals(0, meta.lastAppliedIndex());
- assertDoesNotThrow(() -> meta.lastAppliedIndex(null, 100));
+ assertDoesNotThrow(() -> meta.lastApplied(null, 100, 10));
assertEquals(100, meta.lastAppliedIndex());
- assertDoesNotThrow(() -> meta.lastAppliedIndex(UUID.randomUUID(), 500));
+ assertDoesNotThrow(() -> meta.lastApplied(UUID.randomUUID(), 500, 50));
assertEquals(500, meta.lastAppliedIndex());
}
+ @Test
+ void testLastAppliedTerm() {
+ PartitionMeta meta = new PartitionMeta();
+
+ assertEquals(0, meta.lastAppliedTerm());
+
+ assertDoesNotThrow(() -> meta.lastApplied(null, 100, 10));
Review Comment:
I think that `assertDoesNotThrow` is a bit redundant in these tests. This method is only useful when you don't have any other assertions in a test
##########
modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/PartitionMeta.java:
##########
@@ -341,6 +399,7 @@ public int pageCount() {
*/
void writeTo(PartitionMetaIo metaIo, long pageAddr) {
metaIo.setLastAppliedIndex(pageAddr, lastAppliedIndex);
+ metaIo.setLastAppliedTerm(pageAddr, lastAppliedTerm);
Review Comment:
Do we also need to write `lastGroupConfig` here?
##########
modules/raft/src/main/java/org/apache/ignite/raft/jraft/StateMachine.java:
##########
@@ -96,6 +97,16 @@ public interface StateMachine {
*/
void onConfigurationCommitted(final Configuration conf);
+ /**
+ * Invoked when a configuration has been committed to the group. This is different from
+ * {@link #onConfigurationCommitted( Configuration)} as full configuration entry is provided.
Review Comment:
```suggestion
* {@link #onConfigurationCommitted(Configuration)} as full configuration entry is provided.
```
##########
modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/CommittedConfiguration.java:
##########
@@ -0,0 +1,117 @@
+/*
+ * 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.raft.client.service;
+
+import java.util.List;
+import org.apache.ignite.internal.tostring.IgniteToStringInclude;
+import org.apache.ignite.internal.tostring.S;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * A committed RAFT configuration with index and term at which it was committed.
+ */
+public class CommittedConfiguration {
+ private final long index;
+ private final long term;
+
+ @IgniteToStringInclude
+ private final List<String> peers;
+ @IgniteToStringInclude
+ private final List<String> learners;
+
+ @Nullable
+ @IgniteToStringInclude
+ private final List<String> oldPeers;
+ @Nullable
+ @IgniteToStringInclude
+ private final List<String> oldLearners;
+
+ /**
+ * Creates a new instance.
+ */
+ public CommittedConfiguration(long index, long term, List<String> peers, List<String> learners, @Nullable List<String> oldPeers,
+ @Nullable List<String> oldLearners) {
+ this.index = index;
+ this.term = term;
+ this.peers = List.copyOf(peers);
+ this.learners = List.copyOf(learners);
+ this.oldPeers = oldPeers == null ? null : List.copyOf(oldPeers);
+ this.oldLearners = oldLearners == null ? oldLearners : List.copyOf(oldLearners);
+ }
+
+ /**
+ * Returns RAFT index corresponding to this configuration entry. Never 0.
Review Comment:
I guess that `Never 0` should be replaced with `always positive`
##########
modules/raft/src/main/java/org/apache/ignite/raft/jraft/core/FSMCallerImpl.java:
##########
@@ -665,6 +676,27 @@ private void doSnapshotLoad(final LoadSnapshotClosure done) {
setError(e);
return;
}
+
+ // Tests use such strange metas, so we have to protect... in production, these are never null.
Review Comment:
What tests? Are they ours or from Jraft?
##########
modules/storage-api/src/main/java/org/apache/ignite/internal/storage/GroupConfiguration.java:
##########
@@ -0,0 +1,128 @@
+/*
+ * 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.storage;
+
+import java.io.Serializable;
+import java.util.Collection;
+import java.util.List;
+import java.util.Objects;
+import org.apache.ignite.internal.tostring.IgniteToStringInclude;
+import org.apache.ignite.internal.tostring.S;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * A RAFT group configuration at which it was committed.
+ */
+public class GroupConfiguration implements Serializable {
Review Comment:
Shall we rename it to `RaftGroupConfiguration`?
##########
modules/table/src/test/java/org/apache/ignite/internal/table/distributed/raft/snapshot/outgoing/OutgoingSnapshotMvDataStreamingTest.java:
##########
@@ -281,6 +280,43 @@ void sendsOutOfOrderRowsWithHighestPriority() {
assertThat(response.rows().get(1).rowId(), is(rowId1.uuid()));
}
+ @Test
+ void doesNotEnqueueMissingRows() {
+ when(mvPartitionStorage.scanVersions(rowIdOutOfOrder)).thenReturn(Cursor.fromIterator(emptyIterator()));
Review Comment:
There is a `CursorUtils#emptyCursor` method
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/storage/state/rocksdb/TxStateRocksDbStorage.java:
##########
@@ -115,11 +118,13 @@ public TxStateRocksDbStorage(
this.persistedTierReadOptions = persistedTierReadOptions;
this.partitionId = partitionId;
this.tableStorage = tableStorage;
- this.lastAppliedIndexKey = ByteBuffer.allocate(Short.BYTES).order(ByteOrder.BIG_ENDIAN)
+ this.lastAppliedIndexAndTermKey = ByteBuffer.allocate(Short.BYTES).order(ByteOrder.BIG_ENDIAN)
.putShort((short) partitionId)
.array();
- lastAppliedIndex = readLastAppliedIndex(readOptions);
+ byte[] indexAndTermBytes = readLastAppliedIndexAndTerm(readOptions);
Review Comment:
That's actually interesting: why do you store index and term together here, but not in the Partition Storage?
##########
modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/CommittedConfiguration.java:
##########
@@ -0,0 +1,117 @@
+/*
+ * 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.raft.client.service;
+
+import java.util.List;
+import org.apache.ignite.internal.tostring.IgniteToStringInclude;
+import org.apache.ignite.internal.tostring.S;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * A committed RAFT configuration with index and term at which it was committed.
+ */
+public class CommittedConfiguration {
+ private final long index;
+ private final long term;
+
+ @IgniteToStringInclude
+ private final List<String> peers;
+ @IgniteToStringInclude
+ private final List<String> learners;
+
+ @Nullable
+ @IgniteToStringInclude
+ private final List<String> oldPeers;
+ @Nullable
+ @IgniteToStringInclude
+ private final List<String> oldLearners;
+
+ /**
+ * Creates a new instance.
+ */
+ public CommittedConfiguration(long index, long term, List<String> peers, List<String> learners, @Nullable List<String> oldPeers,
+ @Nullable List<String> oldLearners) {
+ this.index = index;
+ this.term = term;
+ this.peers = List.copyOf(peers);
+ this.learners = List.copyOf(learners);
+ this.oldPeers = oldPeers == null ? null : List.copyOf(oldPeers);
+ this.oldLearners = oldLearners == null ? oldLearners : List.copyOf(oldLearners);
Review Comment:
```suggestion
this.oldLearners = oldLearners == null ? null : List.copyOf(oldLearners);
```
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/server/impl/JraftServerImpl.java:
##########
@@ -588,6 +599,26 @@ public void result(Serializable res) {
}
}
+ @Override
+ public void onRawConfigurationCommitted(ConfigurationEntry entry) {
+ boolean hasOldConf = entry.getOldConf() != null && entry.getOldConf().getPeers() != null;
+
+ CommittedConfiguration committedConf = new CommittedConfiguration(
+ entry.getId().getIndex(),
+ entry.getId().getTerm(),
+ peersIdsToStrings(entry.getConf().getPeers()),
+ peersIdsToStrings(entry.getConf().getLearners()),
+ hasOldConf ? peersIdsToStrings(entry.getOldConf().getPeers()) : null,
+ hasOldConf ? peersIdsToStrings(entry.getOldConf().getLearners()) : null
+ );
+
+ listener.onConfigurationCommitted(committedConf);
+ }
+
+ private static List<String> peersIdsToStrings(Collection<PeerId> peerIds) {
+ return peerIds.stream().map(PeerId::toString).collect(toList());
Review Comment:
I would suggest using `toUnmodifiableList` here, because you will copy this list inside `ComittedConfiguration` otherwise
##########
modules/raft/src/main/java/org/apache/ignite/raft/jraft/core/FSMCallerImpl.java:
##########
@@ -497,6 +497,17 @@ private void doCommitted(final long committedIndex) {
final LogEntry logEntry = iterImpl.entry();
if (logEntry.getType() != EnumOutter.EntryType.ENTRY_TYPE_DATA) {
if (logEntry.getType() == EnumOutter.EntryType.ENTRY_TYPE_CONFIGURATION) {
+ ConfigurationEntry configurationEntry = new ConfigurationEntry(
+ logEntry.getId().copy(),
+ new Configuration(logEntry.getPeers(), logEntry.getLearners()),
+ null
+ );
+ if (logEntry.getOldPeers() != null && !logEntry.getOldPeers().isEmpty()) {
+ configurationEntry.setOldConf(new Configuration(logEntry.getOldPeers(), logEntry.getOldLearners()));
+ }
+
+ this.fsm.onRawConfigurationCommitted(configurationEntry);
+
if (logEntry.getOldPeers() != null && !logEntry.getOldPeers().isEmpty()) {
// Joint stage is not supposed to be noticeable by end users.
this.fsm.onConfigurationCommitted(new Configuration(iterImpl.entry().getPeers()));
Review Comment:
Not related to this PR, but why do we only provide peers configuration here and not learners configuration?
##########
modules/raft/src/main/java/org/apache/ignite/raft/jraft/core/FSMCallerImpl.java:
##########
@@ -497,6 +497,17 @@ private void doCommitted(final long committedIndex) {
final LogEntry logEntry = iterImpl.entry();
if (logEntry.getType() != EnumOutter.EntryType.ENTRY_TYPE_DATA) {
if (logEntry.getType() == EnumOutter.EntryType.ENTRY_TYPE_CONFIGURATION) {
+ ConfigurationEntry configurationEntry = new ConfigurationEntry(
+ logEntry.getId().copy(),
+ new Configuration(logEntry.getPeers(), logEntry.getLearners()),
+ null
Review Comment:
Looks like `ConfigurationEntry` does not expect to be provided with `null` here, you should use an empty configuration instead
##########
modules/storage-api/src/main/java/org/apache/ignite/internal/storage/MvPartitionStorage.java:
##########
@@ -71,20 +71,39 @@ public interface MvPartitionStorage extends ManuallyCloseable {
CompletableFuture<Void> flush();
/**
- * Index of the highest write command applied to the storage. {@code 0} if index is unknown.
+ * Index of the highest write command applied to the storage. {@code 0} if the index is unknown.
*/
long lastAppliedIndex();
/**
- * Sets the last applied index value.
+ * Term of the highest write command applied to the storage. {@code 0} if the term is unknown.
Review Comment:
What do you mean by `highest write command`?
##########
modules/table/src/test/java/org/apache/ignite/internal/table/distributed/raft/snapshot/outgoing/OutgoingSnapshotsManagerTest.java:
##########
@@ -59,15 +67,22 @@ void emptyOngoingSnapshotsIfNoSnapshotWasRegistered() {
}
@Test
- void registersSnapshot() {
- OutgoingSnapshot snapshot = mock(OutgoingSnapshot.class);
- doReturn(partitionKey).when(snapshot).partitionKey();
+ void startsSnapshot() {
+ MvPartitionStorage mvPartitionStorage = mock(MvPartitionStorage.class);
+
+ when(partitionAccess.partitionKey()).thenReturn(partitionKey);
+ when(partitionAccess.mvPartitionStorage()).thenReturn(mvPartitionStorage);
+ when(partitionAccess.txStatePartitionStorage()).thenReturn(mock(TxStateStorage.class));
+
+ when(mvPartitionStorage.committedGroupConfiguration()).thenReturn(mock(GroupConfiguration.class));
+
+ OutgoingSnapshot snapshot = new OutgoingSnapshot(UUID.randomUUID(), partitionAccess);
manager.startOutgoingSnapshot(UUID.randomUUID(), snapshot);
Review Comment:
Do you test that `startOutgoingSnapshot` does not throw any exceptions?
##########
modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/AbstractMvPartitionStorageTest.java:
##########
@@ -1323,6 +1325,35 @@ void committedMethodCallDoesNotInterfereWithIteratingOverScanCursor(ScanTimestam
}
}
+ @Test
+ void groupConfigurationOnEmptyStorageIsNull() {
+ assertThat(storage.committedGroupConfiguration(), is(nullValue()));
+ }
+
+ @Test
+ void groupConfigurationIsUpdated() {
+ GroupConfiguration configToSave = new GroupConfiguration(
+ List.of("peer1", "peer2"),
+ List.of("learner1", "learner2"),
+ List.of("old-peer1", "old-peer2"),
+ List.of("old-learner1", "old-learner2")
+ );
+
+ storage.runConsistently(() -> {
+ storage.committedGroupConfiguration(configToSave);
+
+ return null;
+ });
+
+ GroupConfiguration returnedConfig = storage.committedGroupConfiguration();
+
+ assertThat(returnedConfig, is(notNullValue()));
+ assertThat(returnedConfig.peers(), is(List.of("peer1", "peer2")));
Review Comment:
`GroupConfiguration` has `equals` implemented, so you can compare the objects directly
##########
modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTablePersistenceTest.java:
##########
@@ -213,6 +230,28 @@ public BooleanSupplier snapshotCheckClosure(JraftServerImpl restarted, boolean i
};
}
+ private static Map<ByteBuffer, RowId> rowIdsToRows(MvPartitionStorage storage) {
Review Comment:
I think this method should be called the opposite
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org