IGNITE-24206 Add index and term to RaftGroupConfiguration

modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/impl/raft/MetaStorageSnapshotStorageFactory.java

@@ -68,6 +68,8 @@ public MetaStorageSnapshotStorageFactory(KeyValueStorage storage) {
         assert configuration != null;
 
         return new RaftMessagesFactory().snapshotMeta()
+                .cfgIndex(configuration.index())
+                .cfgTerm(configuration.term())
                 .lastIncludedIndex(indexWithTerm.index())
                 .lastIncludedTerm(indexWithTerm.term())
                 .peersList(configuration.peers())

modules/raft-api/src/main/java/org/apache/ignite/internal/raft/RaftGroupConfiguration.java

@@ -32,11 +32,16 @@
 public class RaftGroupConfiguration implements Serializable {
     private static final long serialVersionUID = 0;
 
+    private final long index;
+    private final long term;
+
     @IgniteToStringInclude
     private final List<String> peers;
     @IgniteToStringInclude
     private final List<String> learners;
 
+
+
     @IgniteToStringInclude
     private final @Nullable List<String> oldPeers;
     @IgniteToStringInclude
@@ -46,11 +51,16 @@ public class RaftGroupConfiguration implements Serializable {
      * Creates a new instance.
      */
     public RaftGroupConfiguration(
+            long index,
+            long term,
             Collection<String> peers,
             Collection<String> learners,
             @Nullable Collection<String> oldPeers,
             @Nullable Collection<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);
@@ -62,13 +72,23 @@ public RaftGroupConfiguration(
      */
     public static RaftGroupConfiguration fromCommittedConfiguration(CommittedConfiguration config) {
         return new RaftGroupConfiguration(
+                config.index(),
+                config.term(),
                 config.peers(),
                 config.learners(),
                 config.oldPeers(),
                 config.oldLearners()
         );
     }
 
+    public long index() {
+        return index;
+    }
+
+    public long term() {
+        return term;
+    }
+
     /**
      * Returns peers of the current configuration.
      *
@@ -121,7 +141,7 @@ public boolean equals(Object o) {
             return false;
         }
         RaftGroupConfiguration that = (RaftGroupConfiguration) o;
-        return Objects.equals(peers, that.peers) && Objects.equals(learners, that.learners)
+        return index == that.index && term == that.term && Objects.equals(peers, that.peers) && Objects.equals(learners, that.learners)
                 && Objects.equals(oldPeers, that.oldPeers) && Objects.equals(oldLearners, that.oldLearners);
     }
 

modules/raft-api/src/main/java/org/apache/ignite/internal/raft/RaftGroupConfigurationSerializer.java

@@ -34,6 +34,8 @@ public class RaftGroupConfigurationSerializer extends VersionedSerializer<RaftGr
 
     @Override
     protected void writeExternalData(RaftGroupConfiguration config, IgniteDataOutput out) throws IOException {
+        out.writeLong(config.index());
+        out.writeLong(config.term());
         writeStringList(config.peers(), out);
         writeStringList(config.learners(), out);
         writeNullableStringList(config.oldPeers(), out);
@@ -57,12 +59,14 @@ private static void writeNullableStringList(@Nullable List<String> strings, Igni
 
     @Override
     protected RaftGroupConfiguration readExternalData(byte protoVer, IgniteDataInput in) throws IOException {
+        long index = in.readLong();
+        long term = in.readLong();
         List<String> peers = readStringList(in);
         List<String> learners = readStringList(in);
         List<String> oldPeers = readNullableStringList(in);
         List<String> oldLearners = readNullableStringList(in);
 
-        return new RaftGroupConfiguration(peers, learners, oldPeers, oldLearners);
+        return new RaftGroupConfiguration(index, term, peers, learners, oldPeers, oldLearners);
     }
 
     private static List<String> readStringList(IgniteDataInput in) throws IOException {

modules/raft-api/src/test/java/org/apache/ignite/internal/raft/RaftGroupConfigurationSerializerTest.java

@@ -32,6 +32,8 @@ class RaftGroupConfigurationSerializerTest {
     @Test
     void serializationAndDeserializationWithoutNulls() {
         RaftGroupConfiguration originalConfig = new RaftGroupConfiguration(
+                13L,
+                37L,
                 List.of("peer1", "peer2"),
                 List.of("learner1", "learner2"),
                 List.of("old-peer1", "old-peer2"),
@@ -47,6 +49,8 @@ void serializationAndDeserializationWithoutNulls() {
     @Test
     void serializationAndDeserializationWithNulls() {
         RaftGroupConfiguration originalConfig = new RaftGroupConfiguration(
+                13L,
+                37L,
                 List.of("peer1", "peer2"),
                 List.of("learner1", "learner2"),
                 null,
@@ -61,10 +65,12 @@ void serializationAndDeserializationWithNulls() {
 
     @Test
     void v1CanBeDeserialized() {
-        byte[] bytes = Base64.getDecoder().decode("Ae++QwMGcGVlcjEGcGVlcjIDCWxlYXJuZXIxCWxlYXJuZXIyAwpvbGQtcGVlcjEKb2xkLXBlZXIyAw1vbGQ"
-                + "tbGVhcm5lcjENb2xkLWxlYXJuZXIy");
+        byte[] bytes = Base64.getDecoder().decode("Ae++Qw0AAAAAAAAAJQAAAAAAAAADBnBlZXIxBnBlZXIyAwlsZWFybmVyMQlsZWFybmVyMgMKb2xkLXBl"
+                + "ZXIxCm9sZC1wZWVyMgMNb2xkLWxlYXJuZXIxDW9sZC1sZWFybmVyMg==");
         RaftGroupConfiguration restoredConfig = VersionedSerialization.fromBytes(bytes, serializer);
 
+        assertThat(restoredConfig.index(), is(13L));
+        assertThat(restoredConfig.term(), is(37L));
         assertThat(restoredConfig.peers(), is(List.of("peer1", "peer2")));
         assertThat(restoredConfig.learners(), is(List.of("learner1", "learner2")));
         assertThat(restoredConfig.oldPeers(), is(List.of("old-peer1", "old-peer2")));

modules/raft/src/integrationTest/java/org/apache/ignite/raft/jraft/core/ItNodeTest.java

@@ -126,6 +126,7 @@
 import org.apache.ignite.raft.jraft.entity.EnumOutter;
 import org.apache.ignite.raft.jraft.entity.NodeId;
 import org.apache.ignite.raft.jraft.entity.PeerId;
+import org.apache.ignite.raft.jraft.entity.RaftOutter.SnapshotMeta;
 import org.apache.ignite.raft.jraft.entity.Task;
 import org.apache.ignite.raft.jraft.entity.UserLog;
 import org.apache.ignite.raft.jraft.error.LogIndexOutOfBoundsException;
@@ -3506,6 +3507,102 @@ public void onRawConfigurationCommitted(ConfigurationEntry conf) {
         );
     }
 
+    @Test
+    public void testIndexAndTermOfCfgArePropagatedToSnapshotMeta() throws Exception {
+        TestPeer peer0 = new TestPeer(testInfo, TestUtils.INIT_PORT);
+
+        AtomicLong term = new AtomicLong(-1);
+        AtomicLong index = new AtomicLong(-1);
+
+        AtomicLong metaTerm = new AtomicLong(-1);
+        AtomicLong metaIndex = new AtomicLong(-1);
+
+        cluster = new TestCluster(
+                "testIndexAndTermOfCfgArePropagatedToSnapshotMeta",
+                dataPath,
+                Collections.singletonList(peer0),
+                new LinkedHashSet<>(),
+                ELECTION_TIMEOUT_MILLIS,
+                (peerId, opts) -> {
+                    opts.setFsm(new MockStateMachine(peerId) {
+                        @Override
+                        public boolean onSnapshotLoad(SnapshotReader reader) {
+                            SnapshotMeta meta = reader.load();
+
+                            metaTerm.set(meta.cfgTerm());
+                            metaIndex.set(meta.cfgIndex());
+
+                            return super.onSnapshotLoad(reader);
+                        }
+
+                        @Override
+                        public void onRawConfigurationCommitted(ConfigurationEntry conf) {
+                            term.set(conf.getId().getTerm());
+                            index.set(conf.getId().getIndex());
+                            System.out.println("Index + term = " + index + " " + term);
+                            super.onRawConfigurationCommitted(conf);
+                        }
+                    });
+                },
+                testInfo
+        );
+
+        assertTrue(cluster.start(peer0));
+
+        Node leader = cluster.waitAndGetLeader();
+
+        assertEquals(1, term.get());
+        assertEquals(1, index.get());
+
+        TestPeer newPeer = new TestPeer(testInfo, TestUtils.INIT_PORT + 1);
+        TestPeer otherPeer = new TestPeer(testInfo, TestUtils.INIT_PORT + 2);
+
+        assertTrue(cluster.start(newPeer, false, 300));
+        assertTrue(cluster.start(otherPeer, false, 300));
+
+        // Wait until new node node sees every other node, otherwise
+        // changePeersAndLearnersAsync can fail.
+        waitForTopologyOnEveryNode(1, cluster);
+
+        SynchronizedClosure done = new SynchronizedClosure();
+        leader.changePeersAndLearnersAsync(
+                new Configuration(List.of(peer0.getPeerId(), newPeer.getPeerId(), otherPeer.getPeerId()), List.of()), leader.getCurrentTerm(),
+                done
+        );
+
+        assertEquals(done.await(), Status.OK());
+
+        assertTrue(waitForCondition(() -> cluster.getLeader().listAlivePeers().contains(newPeer.getPeerId()), 10_000));
+
+        assertTrue(cluster.stop(newPeer.getPeerId()));
+
+        // apply something more
+        sendTestTaskAndWait(leader);
+        triggerLeaderSnapshot(cluster, leader);
+
+        sendTestTaskAndWait(leader, 10);
+
+        triggerLeaderSnapshot(cluster, leader, 2);
+
+        //restart follower.
+        cluster.clean(newPeer.getPeerId());
+        assertTrue(cluster.start(newPeer, false, 300));
+
+        cluster.ensureSame();
+
+        assertEquals(3, cluster.getFsms().size());
+        for (MockStateMachine fsm : cluster.getFsms())
+            assertEquals(20, fsm.getLogs().size(), fsm.getPeerId().toString());
+
+        // Leader hasn't been changed, term must stay the same
+        assertEquals(1, term.get());
+        // idx_2 == joint consensus, idx_3 is expected final cfg
+        assertEquals(3, index.get());
+
+        assertEquals(1, metaTerm.get());
+        assertEquals(3, metaIndex.get());
+    }
+
     @Test
     public void testOnNewPeersConfigurationAppliedIsNotCalledAfterResetPeers() throws Exception {
         TestPeer peer0 = new TestPeer(testInfo, TestUtils.INIT_PORT);

modules/raft/src/main/java/org/apache/ignite/raft/jraft/core/FSMCallerImpl.java

@@ -603,6 +603,8 @@ private void doSnapshotSave(final SaveSnapshotClosure done) {
         }
 
         SnapshotMetaBuilder metaBuilder = msgFactory.snapshotMeta()
+            .cfgIndex(confEntry.getId().getIndex())
+            .cfgTerm(confEntry.getId().getTerm())
             .lastIncludedIndex(lastAppliedIndex)
             .lastIncludedTerm(this.lastAppliedTerm)
             .peersList(confEntry.getConf().getPeers().stream().map(Object::toString).collect(toList()))
@@ -700,7 +702,7 @@ private void doSnapshotLoad(final LoadSnapshotClosure done) {
         // so we have to protect from this. In production, these methods never return null.
         if (meta.peersList() != null && meta.learnersList() != null) {
             ConfigurationEntry configurationEntry = new ConfigurationEntry(
-                    snapshotId.copy(),
+                    new LogId(meta.cfgIndex(), meta.cfgTerm()),
                     new Configuration(
                             meta.peersList().stream().map(PeerId::parsePeer).collect(toList()),
                             meta.learnersList().stream().map(PeerId::parsePeer).collect(toList())

modules/raft/src/main/java/org/apache/ignite/raft/jraft/entity/RaftOutter.java

@@ -57,6 +57,10 @@ public interface SnapshotMeta extends Message {
 
         long lastIncludedTerm();
 
+        long cfgIndex();
+
+        long cfgTerm();
+
         @Nullable Collection<String> peersList();
 
         @Nullable Collection<String> oldPeersList();

modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/snapshot/incoming/IncomingSnapshotCopier.java

@@ -473,6 +473,8 @@ private CompletableFuture<Void> completeRebalance(@Nullable Throwable throwable)
             PartitionSnapshotMeta meta = snapshotMeta;
 
             RaftGroupConfiguration raftGroupConfig = new RaftGroupConfiguration(
+                    meta.cfgIndex(),
+                    meta.cfgTerm(),
                     meta.peersList(),
                     meta.learnersList(),
                     meta.oldPeersList(),

modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/snapshot/outgoing/SnapshotMetaUtils.java

@@ -63,6 +63,8 @@ public static PartitionSnapshotMeta snapshotMetaAt(
             @Nullable String primaryReplicaNodeName
     ) {
         PartitionSnapshotMetaBuilder metaBuilder = new PartitionReplicationMessagesFactory().partitionSnapshotMeta()
+                .cfgIndex(config.index())
+                .cfgTerm(config.term())
                 .lastIncludedIndex(logIndex)
                 .lastIncludedTerm(term)
                 .peersList(config.peers())

modules/table/src/test/java/org/apache/ignite/internal/table/distributed/raft/PartitionCommandListenerTest.java

@@ -620,11 +620,15 @@ void locksOnCommandApplication() {
 
     @Test
     void updatesGroupConfigurationOnConfigCommit() {
+        long index = raftIndex.incrementAndGet();
+
         commandListener.onConfigurationCommitted(new CommittedConfiguration(
-                raftIndex.incrementAndGet(), 2, List.of("peer"), List.of("learner"), List.of("old-peer"), List.of("old-learner")
+                index, 2, List.of("peer"), List.of("learner"), List.of("old-peer"), List.of("old-learner")
         ));
 
         RaftGroupConfiguration expectedConfig = new RaftGroupConfiguration(
+                index,
+                2,
                 List.of("peer"),
                 List.of("learner"),
                 List.of("old-peer"),

modules/table/src/test/java/org/apache/ignite/internal/table/distributed/raft/RaftGroupConfigurationConverterTest.java

@@ -32,6 +32,8 @@ class RaftGroupConfigurationConverterTest {
     @Test
     void convertsAndParses() {
         RaftGroupConfiguration configuration = new RaftGroupConfiguration(
+                13L,
+                37L,
                 List.of("peer"),
                 List.of("learner"),
                 List.of("old-peer"),

modules/table/src/test/java/org/apache/ignite/internal/table/distributed/raft/snapshot/incoming/IncomingSnapshotCopierTest.java

@@ -732,6 +732,8 @@ private static List<UUID> generateTxIds() {
 
     private static RaftGroupConfiguration generateRaftGroupConfig() {
         return new RaftGroupConfiguration(
+                13L,
+                37L,
                 List.of("peer"),
                 List.of("learner"),
                 List.of("old-peer"),

modules/table/src/test/java/org/apache/ignite/internal/table/distributed/raft/snapshot/outgoing/OutgoingSnapshotCommonTest.java

@@ -73,6 +73,8 @@ void sendsSnapshotMeta() {
         when(partitionAccess.maxLastAppliedIndex()).thenReturn(100L);
         when(partitionAccess.maxLastAppliedTerm()).thenReturn(3L);
         when(partitionAccess.committedGroupConfiguration()).thenReturn(new RaftGroupConfiguration(
+                13L,
+                37L,
                 List.of("peer1:3000", "peer2:3000"),
                 List.of("learner1:3000", "learner2:3000"),
                 List.of("peer1:3000"),
@@ -89,6 +91,8 @@ void sendsSnapshotMeta() {
 
         SnapshotMetaResponse response = getSnapshotMetaResponse();
 
+        assertThat(response.meta().cfgIndex(), is(13L));
+        assertThat(response.meta().cfgTerm(), is(37L));
         assertThat(response.meta().lastIncludedIndex(), is(100L));
         assertThat(response.meta().lastIncludedTerm(), is(3L));
         assertThat(response.meta().peersList(), is(List.of("peer1:3000", "peer2:3000")));
@@ -121,7 +125,7 @@ private SnapshotMetaResponse getNullableSnapshotMetaResponse() {
     @Test
     void doesNotSendOldConfigWhenItIsNotThere() {
         when(partitionAccess.committedGroupConfiguration()).thenReturn(new RaftGroupConfiguration(
-                List.of(), List.of(), null, null
+                13L, 37L, List.of(), List.of(), null, null
         ));
 
         snapshot.freezeScopeUnderMvLock();

modules/table/src/test/java/org/apache/ignite/internal/table/distributed/raft/snapshot/outgoing/SnapshotAwarePartitionDataStorageTest.java

@@ -136,6 +136,8 @@ void delegatesLastAppliedSetter() {
     @Test
     void convertsCommittedGroupConfigurationOnSave() {
         RaftGroupConfiguration config = new RaftGroupConfiguration(
+                13L,
+                37L,
                 List.of("peer"),
                 List.of("learner"),
                 List.of("old-peer"),

modules/table/src/test/java/org/apache/ignite/internal/table/distributed/raft/snapshot/outgoing/SnapshotMetaUtilsTest.java

@@ -60,6 +60,8 @@ class SnapshotMetaUtilsTest extends BaseIgniteAbstractTest {
     @Test
     void buildsSnapshotMeta() {
         RaftGroupConfiguration config = new RaftGroupConfiguration(
+                13L,
+                37L,
                 List.of("peer1:3000", "peer2:3000"), List.of("learner1:3000", "learner2:3000"),
                 List.of("peer1:3000"), List.of("learner1:3000")
         );
@@ -78,6 +80,8 @@ void buildsSnapshotMeta() {
                 "primary"
         );
 
+        assertThat(meta.cfgIndex(), is(13L));
+        assertThat(meta.cfgTerm(), is(37L));
         assertThat(meta.lastIncludedIndex(), is(100L));
         assertThat(meta.lastIncludedTerm(), is(3L));
         assertThat(meta.peersList(), is(List.of("peer1:3000", "peer2:3000")));
@@ -96,7 +100,7 @@ void doesNotIncludeOldConfigWhenItIsNotThere() {
         PartitionSnapshotMeta meta = snapshotMetaAt(
                 100,
                 3,
-                new RaftGroupConfiguration(List.of(), List.of(), null, null),
+                new RaftGroupConfiguration(13, 37, List.of(), List.of(), null, null),
                 42,
                 Map.of(),
                 777L,