IGNITE-22218 Remove TableRaftService in favor of using RaftGroupService from Replica instances

modules/client/src/test/java/org/apache/ignite/client/fakes/FakeInternalTable.java

@@ -52,7 +52,6 @@
 import org.apache.ignite.internal.storage.engine.MvTableStorage;
 import org.apache.ignite.internal.table.InternalTable;
 import org.apache.ignite.internal.table.StreamerReceiverRunner;
-import org.apache.ignite.internal.table.TableRaftService;
 import org.apache.ignite.internal.tx.InternalTransaction;
 import org.apache.ignite.internal.tx.storage.state.TxStateTableStorage;
 import org.apache.ignite.internal.util.PendingComparableValuesTracker;
@@ -461,12 +460,6 @@ public Publisher<BinaryRow> lookup(
         throw new IgniteInternalException(new OperationNotSupportedException());
     }
 
-    @Override
-    public TableRaftService tableRaftService() {
-        throw new IgniteInternalException(new OperationNotSupportedException());
-    }
-
-
     @Override public TxStateTableStorage txStateStorage() {
         return null;
     }

modules/index/build.gradle

@@ -81,6 +81,7 @@ dependencies {
     integrationTestImplementation testFixtures(project(':ignite-table'))
     integrationTestImplementation testFixtures(project(':ignite-storage-api'))
     integrationTestImplementation testFixtures(project(':ignite-catalog'))
+    integrationTestImplementation testFixtures(project(':ignite-replicator'))
     integrationTestImplementation libs.jetbrains.annotations
 }
 

modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/ItReplicaLifecycleTest.java

@@ -118,6 +118,7 @@
 import org.apache.ignite.internal.metastorage.server.KeyValueStorage;
 import org.apache.ignite.internal.metastorage.server.SimpleInMemoryKeyValueStorage;
 import org.apache.ignite.internal.metrics.NoOpMetricManager;
+import org.apache.ignite.internal.network.ClusterNodeImpl;
 import org.apache.ignite.internal.network.ClusterService;
 import org.apache.ignite.internal.network.StaticNodeFinder;
 import org.apache.ignite.internal.network.configuration.NetworkConfiguration;
@@ -176,6 +177,7 @@
 import org.apache.ignite.internal.tx.message.WriteIntentSwitchReplicaRequest;
 import org.apache.ignite.internal.tx.test.TestLocalRwTxCounter;
 import org.apache.ignite.internal.vault.VaultManager;
+import org.apache.ignite.network.ClusterNode;
 import org.apache.ignite.network.NetworkAddress;
 import org.apache.ignite.raft.jraft.rpc.impl.RaftGroupEventsClientListener;
 import org.apache.ignite.sql.IgniteSql;
@@ -304,6 +306,8 @@ private void startNodes(TestInfo testInfo, int amount) throws NodeStoppingExcept
 
         node0.cmgManager.initCluster(List.of(node0.name), List.of(node0.name), "cluster");
 
+        placementDriver.setPrimary(node0.toClusterNode());
+
         nodes.values().forEach(Node::waitWatches);
 
         assertThat(
@@ -346,6 +350,7 @@ private void stopNode(int idx) {
     }
 
     @Test
+    @Disabled("https://issues.apache.org/jira/browse/IGNITE-22928")
     public void testZoneReplicaListener(TestInfo testInfo) throws Exception {
         startNodes(testInfo, 3);
 
@@ -1125,8 +1130,9 @@ public CompletableFuture<Boolean> invoke(
                     lowWatermark,
                     threadPoolsManager.tableIoExecutor(),
                     rebalanceScheduler,
-                    threadPoolsManager.partitionOperationsExecutor()
-            );
+                    threadPoolsManager.partitionOperationsExecutor(),
+                    clockService,
+                    placementDriver);
 
             StorageUpdateConfiguration storageUpdateConfiguration = clusterConfigRegistry.getConfiguration(StorageUpdateConfiguration.KEY);
 
@@ -1280,6 +1286,10 @@ void stop() {
             nodeCfgGenerator.close();
             clusterCfgGenerator.close();
         }
+
+        ClusterNode toClusterNode() {
+            return new ClusterNodeImpl(name, name, networkAddress);
+        }
     }
 
     @FunctionalInterface

modules/placement-driver/src/integrationTest/java/org/apache/ignite/internal/placementdriver/ItPrimaryReplicaChoiceTest.java

@@ -48,6 +48,7 @@
 import org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEvent;
 import org.apache.ignite.internal.raft.Peer;
 import org.apache.ignite.internal.raft.service.RaftGroupService;
+import org.apache.ignite.internal.replicator.ReplicaTestUtils;
 import org.apache.ignite.internal.replicator.TablePartitionId;
 import org.apache.ignite.internal.schema.BinaryRow;
 import org.apache.ignite.internal.schema.BinaryTuple;
@@ -154,13 +155,14 @@ public void testPrimaryChangeSubscription() throws Exception {
 
     @Test
     public void testPrimaryChangeLongHandling() throws Exception {
-        TableViewInternal tbl = unwrapTableImpl(node(0).tables().table(TABLE_NAME));
+        IgniteImpl node = node(0);
+        TableViewInternal tbl = unwrapTableImpl(node.tables().table(TABLE_NAME));
 
         var tblReplicationGrp = new TablePartitionId(tbl.tableId(), PART_ID);
 
-        CompletableFuture<ReplicaMeta> primaryReplicaFut = node(0).placementDriver().awaitPrimaryReplica(
+        CompletableFuture<ReplicaMeta> primaryReplicaFut = node.placementDriver().awaitPrimaryReplica(
                 tblReplicationGrp,
-                node(0).clock().now(),
+                node.clock().now(),
                 AWAIT_PRIMARY_REPLICA_TIMEOUT,
                 SECONDS
         );
@@ -184,7 +186,7 @@ public void testPrimaryChangeLongHandling() throws Exception {
         CompletableFuture<String> primaryChangeTask =
                 IgniteTestUtils.runAsync(() -> NodeUtils.transferPrimary(nodes, tblReplicationGrp, primary));
 
-        waitingForLeaderCache(tbl, primary);
+        waitingForLeaderCache(node, tbl);
 
         assertFalse(primaryChangeTask.isDone());
 
@@ -399,12 +401,13 @@ private int getIndexId(String idxName) {
     /**
      * Waits when the leader would be a different with the current primary replica.
      *
+     * @param node Ignite node.
      * @param tbl Table.
-     * @param primary Current primary replica name.
      * @throws InterruptedException If fail.
      */
-    private static void waitingForLeaderCache(TableViewInternal tbl, String primary) throws InterruptedException {
-        RaftGroupService raftSrvc = tbl.internalTable().tableRaftService().partitionRaftGroupService(0);
+    private static void waitingForLeaderCache(IgniteImpl node, TableViewInternal tbl) throws InterruptedException {
+        RaftGroupService raftSrvc = ReplicaTestUtils.getRaftClient(node, tbl.tableId(), 0)
+                        .orElseThrow(AssertionError::new);
 
         assertTrue(waitForCondition(() -> {
             raftSrvc.refreshLeader();

modules/replicator/src/integrationTest/java/org/apache/ignite/internal/replicator/ItPlacementDriverReplicaSideTest.java

@@ -528,7 +528,6 @@ public RaftCommandRunner raftClient() {
                     return replicaManager.startReplica(
                             groupId,
                             newConfiguration,
-                            (unused) -> { },
                             (unused) -> listener,
                             new PendingComparableValuesTracker<>(Long.MAX_VALUE),
                             completedFuture(raftClient));

modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java

@@ -53,7 +53,6 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.LongSupplier;
 import java.util.function.Supplier;
@@ -578,7 +577,6 @@ private CompletableFuture<Replica> startReplicaInternal(
             RaftGroupListener raftGroupListener,
             boolean isVolatileStorage,
             SnapshotStorageFactory snapshotStorageFactory,
-            Consumer<RaftGroupService> updateTableRaftService,
             Function<RaftGroupService, ReplicaListener> createListener,
             PendingComparableValuesTracker<Long, Void> storageIndexTracker,
             TablePartitionId replicaGrpId,
@@ -604,7 +602,6 @@ private CompletableFuture<Replica> startReplicaInternal(
         return startReplica(
                 replicaGrpId,
                 newConfiguration,
-                updateTableRaftService,
                 createListener,
                 storageIndexTracker,
                 newRaftClientFut
@@ -618,8 +615,6 @@ private CompletableFuture<Replica> startReplicaInternal(
      * @param raftGroupListener Raft group listener for raft group starting.
      * @param isVolatileStorage is table storage volatile?
      * @param snapshotStorageFactory Snapshot storage factory for raft group option's parameterization.
-     * @param updateTableRaftService Temporal consumer while TableRaftService wouldn't be removed in
-     *      TODO: https://issues.apache.org/jira/browse/IGNITE-22218.
      * @param createListener Due to creation of ReplicaListener in TableManager, the function returns desired listener by created
      *      raft-client inside {@link #startReplica} method.
      * @param replicaGrpId Replication group id.
@@ -633,7 +628,6 @@ public CompletableFuture<Replica> startReplica(
             RaftGroupListener raftGroupListener,
             boolean isVolatileStorage,
             SnapshotStorageFactory snapshotStorageFactory,
-            Consumer<RaftGroupService> updateTableRaftService,
             Function<RaftGroupService, ReplicaListener> createListener,
             PendingComparableValuesTracker<Long, Void> storageIndexTracker,
             TablePartitionId replicaGrpId,
@@ -649,7 +643,6 @@ public CompletableFuture<Replica> startReplica(
                     raftGroupListener,
                     isVolatileStorage,
                     snapshotStorageFactory,
-                    updateTableRaftService,
                     createListener,
                     storageIndexTracker,
                     replicaGrpId,
@@ -745,8 +738,6 @@ public CompletableFuture<Replica> startReplica(
      *
      * @param replicaGrpId Replication group id.
      * @param newConfiguration Peers and Learners of the Raft group.
-     * @param updateTableRaftService A temporal clojure that updates table raft service with new raft-client, but
-     *      TODO: will be removed https://issues.apache.org/jira/browse/IGNITE-22218
      * @param createListener A clojure that returns done {@link ReplicaListener} by given raft-client {@link RaftGroupService}.
      * @param storageIndexTracker Storage index tracker.
      * @param newRaftClientFut A future that returns created raft-client.
@@ -758,19 +749,14 @@ public CompletableFuture<Replica> startReplica(
     public CompletableFuture<Replica> startReplica(
             ReplicationGroupId replicaGrpId,
             PeersAndLearners newConfiguration,
-            Consumer<RaftGroupService> updateTableRaftService,
             Function<RaftGroupService, ReplicaListener> createListener,
             PendingComparableValuesTracker<Long, Void> storageIndexTracker,
             CompletableFuture<TopologyAwareRaftGroupService> newRaftClientFut
     ) throws NodeStoppingException {
         LOG.info("Replica is about to start [replicationGroupId={}].", replicaGrpId);
 
         return newRaftClientFut
-                .thenApplyAsync(raftClient -> {
-                    // TODO: will be removed in https://issues.apache.org/jira/browse/IGNITE-22218
-                    updateTableRaftService.accept(raftClient);
-                    return createListener.apply(raftClient);
-                }, replicasCreationExecutor)
+                .thenApplyAsync(createListener, replicasCreationExecutor)
                 .thenCompose(replicaListener -> startReplica(replicaGrpId, storageIndexTracker, completedFuture(replicaListener)));
     }
 
@@ -1147,6 +1133,9 @@ private void sendSafeTimeSyncIfReplicaReady(CompletableFuture<Replica> replicaFu
      * @param replicaGrpId Replication group id.
      * @return True if the replica is started.
      */
+    @TestOnly
+    @VisibleForTesting
+    @Deprecated
     public boolean isReplicaStarted(ReplicationGroupId replicaGrpId) {
         CompletableFuture<Replica> replicaFuture = replicas.get(replicaGrpId);
         return replicaFuture != null && isCompletedSuccessfully(replicaFuture);

modules/replicator/src/test/java/org/apache/ignite/internal/replicator/ReplicaManagerTest.java

@@ -183,7 +183,6 @@ void testReplicaEvents(
         CompletableFuture<Replica> startReplicaFuture = replicaManager.startReplica(
                 groupId,
                 newConfiguration,
-                (unused) -> { },
                 (unused) -> replicaListener,
                 new PendingComparableValuesTracker<>(0L),
                 completedFuture(raftGroupService)

modules/replicator/src/testFixtures/java/org/apache/ignite/internal/replicator/ReplicaTestUtils.java

@@ -0,0 +1,139 @@
+/*
+ * 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.replicator;
+
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.internal.lang.IgniteInternalException;
+import org.apache.ignite.internal.network.ClusterService;
+import org.apache.ignite.internal.network.TopologyService;
+import org.apache.ignite.internal.raft.service.RaftGroupService;
+import org.apache.ignite.internal.testframework.IgniteTestUtils;
+import org.apache.ignite.network.ClusterNode;
+import org.jetbrains.annotations.TestOnly;
+
+/** Utilities for working with replicas and replicas manager in tests. */
+public final class ReplicaTestUtils {
+    /**
+     * Returns raft-client if exists.
+     *
+     * @param node Ignite node that hosts the raft-client.
+     * @param tableId Desired table's ID.
+     * @param partId Desired partition's ID.
+     *
+     * @return Optional with raft-client if exists on the node by given identifiers.
+     */
+    @TestOnly
+    public static Optional<RaftGroupService> getRaftClient(Ignite node, int tableId, int partId) {
+        return getRaftClient(getReplicaManager(node), tableId, partId);
+    }
+
+    /**
+     * Returns raft-client if exists.
+     *
+     * @param replicaManager Ignite node's replica manager with replica that should contains a raft client.
+     * @param tableId Desired table's ID.
+     * @param partId Desired partition's ID.
+     *
+     * @return Optional with raft-client if exists on the node by given identifiers.
+     */
+    @TestOnly
+    public static Optional<RaftGroupService> getRaftClient(ReplicaManager replicaManager, int tableId, int partId) {
+        CompletableFuture<Replica> replicaFut = replicaManager
+                .replica(new TablePartitionId(tableId, partId));
+
+        if  (replicaFut == null) {
+            return Optional.empty();
+        }
+
+        try {
+            return Optional.of(replicaFut.get(15, TimeUnit.SECONDS).raftClient());
+        } catch (ExecutionException | InterruptedException | TimeoutException e) {
+            return Optional.empty();
+        }
+    }
+
+    /**
+     * Extracts {@link ReplicaManager} from the given {@link Ignite} node.
+     *
+     * @param node The given node with desired replica manager.
+     *
+     * @return Replica manager component from given node.
+     */
+    @TestOnly
+    public static ReplicaManager getReplicaManager(Ignite node) {
+        return IgniteTestUtils.getFieldValue(node, "replicaMgr");
+    }
+
+    /**
+     * Extracts {@link TopologyService} from the given {@link Ignite} node.
+     *
+     * @param node The given node with desired topology service.
+     *
+     * @return Topology service component from given node.
+     */
+    @TestOnly
+    private static TopologyService getTopologyService(Ignite node) {
+        ClusterService clusterService = IgniteTestUtils.getFieldValue(node, "clusterSvc");
+        return clusterService.topologyService();
+    }
+
+    /**
+     * Returns cluster node that is the leader of the corresponding partition group or throws an exception if it cannot be found.
+     *
+     * @param node Ignite node with raft client.
+     * @param tableId Table identifier.
+     * @param partId Partition number.
+     *
+     * @return Leader node of the partition group corresponding to the partition
+     */
+    @TestOnly
+    public static ClusterNode leaderAssignment(Ignite node, int tableId, int partId) {
+        return leaderAssignment(getReplicaManager(node), getTopologyService(node), tableId, partId);
+    }
+
+    /**
+     * Returns cluster node that is the leader of the corresponding partition group or throws an exception if it cannot be found.
+     *
+     * @param replicaManager Ignite node's replica manager with replica that should contains a raft client.
+     * @param topologyService Ignite node's topology service that should find and return leader cluster node.
+     * @param tableId Table identifier.
+     * @param partId Partition number.
+     *
+     * @return Leader node of the partition group corresponding to the partition
+     */
+    @TestOnly
+    public static ClusterNode leaderAssignment(ReplicaManager replicaManager, TopologyService topologyService, int tableId, int partId) {
+        RaftGroupService raftClient = getRaftClient(replicaManager, tableId, partId)
+                .orElseThrow(() -> new IgniteInternalException("No such partition " + partId + " in table " + tableId));
+
+        if (raftClient.leader() == null) {
+            try {
+                raftClient.refreshLeader().get(15, TimeUnit.SECONDS);
+            } catch (InterruptedException | ExecutionException | TimeoutException e) {
+                throw new IgniteInternalException("Couldn't get a leader for partition " + partId + " in table " + tableId, e);
+            }
+        }
+
+        return topologyService.getByConsistentId(raftClient.leader().consistentId());
+    }
+}

modules/runner/build.gradle

@@ -183,6 +183,7 @@ dependencies {
     integrationTestImplementation testFixtures(project(':ignite-failure-handler'))
     integrationTestImplementation testFixtures(project(':ignite-metrics:'))
     integrationTestImplementation testFixtures(project(':ignite-raft'))
+    integrationTestImplementation testFixtures(project(':ignite-replicator'))
     integrationTestImplementation libs.jetbrains.annotations
     integrationTestImplementation libs.awaitility
     integrationTestImplementation libs.rocksdb.jni