IGNITE-22233 Zone replica listener

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

@@ -282,7 +282,7 @@ void after() {
     }
 
     @Test
-    public void testEmptyReplicaListener() throws NodeStoppingException {
+    public void testZoneReplicaListener() throws NodeStoppingException {
         Assignment replicaAssignment = (Assignment) AffinityUtils.calculateAssignmentForPartition(
                 nodes.stream().map(n -> n.name).collect(Collectors.toList()), 0, 1).toArray()[0];
 
@@ -293,17 +293,33 @@ public void testEmptyReplicaListener() throws NodeStoppingException {
         createZone(node, "test_zone", 1, 1);
         int zoneId = DistributionZonesTestUtil.getZoneId(node.catalogManager, "test_zone", node.hybridClock.nowLong());
 
-        createTable(node, "test_zone", "test_table");
-        int tableId = TableTestUtils.getTableId(node.catalogManager, "test_table", node.hybridClock.nowLong());
+        {
+            createTable(node, "test_zone", "test_table");
+            int tableId = TableTestUtils.getTableId(node.catalogManager, "test_table", node.hybridClock.nowLong());
 
-        node.converter.put(new TablePartitionId(tableId, 0), new ZonePartitionId(zoneId, 0));
+            node.converter.put(new TablePartitionId(tableId, 0), new ZonePartitionId(zoneId, 0));
 
-        KeyValueView<Long, Integer> keyValueView = node.tableManager.table(tableId).keyValueView(Long.class, Integer.class);
+            KeyValueView<Long, Integer> keyValueView = node.tableManager.table(tableId).keyValueView(Long.class, Integer.class);
 
-        assertDoesNotThrow(() -> keyValueView.put(null, 1L, 1));
+            assertDoesNotThrow(() -> keyValueView.put(null, 1L, 100));
 
-        // Actually we are testing not the fair put value, but the hardcoded one from temporary noop replica listener
-        assertEquals(-1, keyValueView.get(null, 1L));
+            // Actually we are testing not the fair put value, but the hardcoded one from temporary noop replica listener
+            assertEquals(100, keyValueView.get(null, 1L));
+        }
+
+        {
+            createTable(node, "test_zone", "test_table1");
+            int tableId = TableTestUtils.getTableId(node.catalogManager, "test_table1", node.hybridClock.nowLong());
+
+            node.converter.put(new TablePartitionId(tableId, 0), new ZonePartitionId(zoneId, 0));
+
+            KeyValueView<Long, Integer> keyValueView = node.tableManager.table(tableId).keyValueView(Long.class, Integer.class);
+
+            assertDoesNotThrow(() -> keyValueView.put(null, 1L, 200));
+
+            // Actually we are testing not the fair put value, but the hardcoded one from temporary noop replica listener
+            assertEquals(200, keyValueView.get(null, 1L));
+        }
     }
 
     private Node getNode(int nodeIndex) {
@@ -682,7 +698,7 @@ class Node {
                     clusterService.topologyService(),
                     clusterService.serializationRegistry(),
                     replicaManager,
-                    mock(LockManager.class),
+                    lockManager,
                     replicaSvc,
                     txManager,
                     dataStorageMgr,
@@ -704,7 +720,8 @@ class Node {
                     resourcesRegistry,
                     lowWatermark,
                     transactionInflights,
-                    indexMetaStorage
+                    indexMetaStorage,
+                    partitionReplicaLifecycleManager
             );
 
             indexManager = new IndexManager(

modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/PartitionReplicaLifecycleManager.java

@@ -71,9 +71,12 @@
 import org.apache.ignite.internal.raft.PeersAndLearners;
 import org.apache.ignite.internal.raft.RaftGroupEventsListener;
 import org.apache.ignite.internal.raft.service.RaftGroupListener;
+import org.apache.ignite.internal.replicator.Replica;
 import org.apache.ignite.internal.replicator.ReplicaManager;
 import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.TablePartitionId;
 import org.apache.ignite.internal.replicator.ZonePartitionId;
+import org.apache.ignite.internal.replicator.ZonePartitionReplicaImpl;
 import org.apache.ignite.internal.util.ExceptionUtils;
 import org.apache.ignite.internal.util.IgniteSpinBusyLock;
 import org.apache.ignite.internal.util.IgniteUtils;
@@ -200,7 +203,6 @@ private CompletableFuture<Void> createZonePartitionReplicationNodes(int zoneId,
         try {
             return replicaMgr.startReplica(
                     replicaGrpId,
-                    new ZonePartitionReplicaListener(),
                     new FailFastSnapshotStorageFactory(),
                     realConfiguration,
                     raftGroupListener,
@@ -229,6 +231,24 @@ public void beforeNodeStop() {
         cleanUpPartitionsResources(replicationGroupIds);
     }
 
+    /**
+     * Add table replica to the aggregated zone replica.
+     *
+     * @param zonePartitionId Zone partition id.
+     * @param replicationGroupId Table partition id.
+     * @param replica Table replica
+     * @return Future, which will be completed when operation done
+     */
+    public CompletableFuture<Void> addTableReplica(ZonePartitionId zonePartitionId, TablePartitionId replicationGroupId, Replica replica) {
+        if (!ENABLED) {
+            return nullCompletedFuture();
+        }
+
+        return replicaMgr
+                .replica(zonePartitionId)
+                .thenAccept(r -> ((ZonePartitionReplicaImpl) r).addReplica(replicationGroupId, replica));
+    }
+
     /**
      * Writes the set of assignments to meta storage. If there are some assignments already, gets them from meta storage. Returns
      * the list of assignments that really are in meta storage.

modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/replication/ReadOnlyDirectMultiRowReplicaRequest.java

@@ -20,10 +20,11 @@
 import org.apache.ignite.internal.network.annotations.Transferable;
 import org.apache.ignite.internal.partition.replicator.network.PartitionReplicationMessageGroup;
 import org.apache.ignite.internal.replicator.message.ReadOnlyDirectReplicaRequest;
+import org.apache.ignite.internal.replicator.message.TableAware;
 
 /**
  * Read only direct multi row replica request.
  */
 @Transferable(PartitionReplicationMessageGroup.RO_DIRECT_MULTI_ROW_REPLICA_REQUEST)
-public interface ReadOnlyDirectMultiRowReplicaRequest extends MultipleRowPkReplicaRequest, ReadOnlyDirectReplicaRequest {
+public interface ReadOnlyDirectMultiRowReplicaRequest extends MultipleRowPkReplicaRequest, ReadOnlyDirectReplicaRequest, TableAware {
 }

modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/replication/ReadOnlyDirectSingleRowReplicaRequest.java

@@ -20,11 +20,12 @@
 import org.apache.ignite.internal.network.annotations.Transferable;
 import org.apache.ignite.internal.partition.replicator.network.PartitionReplicationMessageGroup;
 import org.apache.ignite.internal.replicator.message.ReadOnlyDirectReplicaRequest;
+import org.apache.ignite.internal.replicator.message.TableAware;
 
 /**
  * Read only direct node single row replica request.
  * The type of RO request never waits and is executed at the current node timestamp.
  */
 @Transferable(PartitionReplicationMessageGroup.RO_DIRECT_SINGLE_ROW_REPLICA_REQUEST)
-public interface ReadOnlyDirectSingleRowReplicaRequest extends SingleRowPkReplicaRequest, ReadOnlyDirectReplicaRequest {
+public interface ReadOnlyDirectSingleRowReplicaRequest extends SingleRowPkReplicaRequest, ReadOnlyDirectReplicaRequest, TableAware {
 }

modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/replication/ReadOnlyMultiRowPkReplicaRequest.java

@@ -19,10 +19,11 @@
 
 import org.apache.ignite.internal.network.annotations.Transferable;
 import org.apache.ignite.internal.partition.replicator.network.PartitionReplicationMessageGroup;
+import org.apache.ignite.internal.replicator.message.TableAware;
 
 /**
  * Read only multi row replica request.
  */
 @Transferable(PartitionReplicationMessageGroup.RO_MULTI_ROW_REPLICA_REQUEST)
-public interface ReadOnlyMultiRowPkReplicaRequest extends MultipleRowPkReplicaRequest, ReadOnlyReplicaRequest {
+public interface ReadOnlyMultiRowPkReplicaRequest extends MultipleRowPkReplicaRequest, ReadOnlyReplicaRequest, TableAware {
 }

modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/replication/ReadOnlyScanRetrieveBatchReplicaRequest.java

@@ -20,12 +20,13 @@
 import java.util.UUID;
 import org.apache.ignite.internal.network.annotations.Transferable;
 import org.apache.ignite.internal.partition.replicator.network.PartitionReplicationMessageGroup;
+import org.apache.ignite.internal.replicator.message.TableAware;
 
 /**
  * Scan retrieve batch replica request.
  */
 @Transferable(PartitionReplicationMessageGroup.RO_SCAN_RETRIEVE_BATCH_REPLICA_REQUEST)
-public interface ReadOnlyScanRetrieveBatchReplicaRequest extends ScanRetrieveBatchReplicaRequest, ReadOnlyReplicaRequest {
+public interface ReadOnlyScanRetrieveBatchReplicaRequest extends ScanRetrieveBatchReplicaRequest, ReadOnlyReplicaRequest, TableAware {
     UUID transactionId();
 
     /**

modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/replication/ReadOnlySingleRowPkReplicaRequest.java

@@ -19,10 +19,11 @@
 
 import org.apache.ignite.internal.network.annotations.Transferable;
 import org.apache.ignite.internal.partition.replicator.network.PartitionReplicationMessageGroup;
+import org.apache.ignite.internal.replicator.message.TableAware;
 
 /**
  * Read only single row replica request.
  */
 @Transferable(PartitionReplicationMessageGroup.RO_SINGLE_ROW_REPLICA_REQUEST)
-public interface ReadOnlySingleRowPkReplicaRequest extends SingleRowPkReplicaRequest, ReadOnlyReplicaRequest {
+public interface ReadOnlySingleRowPkReplicaRequest extends SingleRowPkReplicaRequest, ReadOnlyReplicaRequest, TableAware {
 }

modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/replication/ReadWriteMultiRowPkReplicaRequest.java

@@ -19,12 +19,13 @@
 
 import org.apache.ignite.internal.network.annotations.Transferable;
 import org.apache.ignite.internal.partition.replicator.network.PartitionReplicationMessageGroup;
+import org.apache.ignite.internal.replicator.message.TableAware;
 
 /**
  * Read-write multi-row replica request involving table's Primary Keys.
  */
 @Transferable(PartitionReplicationMessageGroup.RW_MULTI_ROW_PK_REPLICA_REQUEST)
-public interface ReadWriteMultiRowPkReplicaRequest extends MultipleRowPkReplicaRequest, ReadWriteReplicaRequest {
+public interface ReadWriteMultiRowPkReplicaRequest extends MultipleRowPkReplicaRequest, ReadWriteReplicaRequest, TableAware {
     /**
      * Disable delayed ack optimization.
      *

modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/replication/ReadWriteMultiRowReplicaRequest.java

@@ -20,13 +20,14 @@
 import java.util.BitSet;
 import org.apache.ignite.internal.network.annotations.Transferable;
 import org.apache.ignite.internal.partition.replicator.network.PartitionReplicationMessageGroup;
+import org.apache.ignite.internal.replicator.message.TableAware;
 import org.jetbrains.annotations.Nullable;
 
 /**
  * Read-write multi-row replica request.
  */
 @Transferable(PartitionReplicationMessageGroup.RW_MULTI_ROW_REPLICA_REQUEST)
-public interface ReadWriteMultiRowReplicaRequest extends MultipleRowReplicaRequest, ReadWriteReplicaRequest {
+public interface ReadWriteMultiRowReplicaRequest extends MultipleRowReplicaRequest, ReadWriteReplicaRequest, TableAware {
     /**
      * Disable delayed ack optimization.
      *

modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/replication/ReadWriteScanRetrieveBatchReplicaRequest.java

@@ -19,10 +19,11 @@
 
 import org.apache.ignite.internal.network.annotations.Transferable;
 import org.apache.ignite.internal.partition.replicator.network.PartitionReplicationMessageGroup;
+import org.apache.ignite.internal.replicator.message.TableAware;
 
 /**
  * Scan retrieve batch replica request.
  */
 @Transferable(PartitionReplicationMessageGroup.RW_SCAN_RETRIEVE_BATCH_REPLICA_REQUEST)
-public interface ReadWriteScanRetrieveBatchReplicaRequest extends ScanRetrieveBatchReplicaRequest, ReadWriteReplicaRequest {
+public interface ReadWriteScanRetrieveBatchReplicaRequest extends ScanRetrieveBatchReplicaRequest, ReadWriteReplicaRequest, TableAware {
 }

modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/replication/ReadWriteSingleRowPkReplicaRequest.java

@@ -19,10 +19,11 @@
 
 import org.apache.ignite.internal.network.annotations.Transferable;
 import org.apache.ignite.internal.partition.replicator.network.PartitionReplicationMessageGroup;
+import org.apache.ignite.internal.replicator.message.TableAware;
 
 /**
  * Read-write single-row replica request involving a table's Primary Key..
  */
 @Transferable(PartitionReplicationMessageGroup.RW_SINGLE_ROW_PK_REPLICA_REQUEST)
-public interface ReadWriteSingleRowPkReplicaRequest extends SingleRowPkReplicaRequest, ReadWriteReplicaRequest {
+public interface ReadWriteSingleRowPkReplicaRequest extends SingleRowPkReplicaRequest, ReadWriteReplicaRequest, TableAware {
 }

modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/replication/ReadWriteSingleRowReplicaRequest.java

@@ -19,10 +19,11 @@
 
 import org.apache.ignite.internal.network.annotations.Transferable;
 import org.apache.ignite.internal.partition.replicator.network.PartitionReplicationMessageGroup;
+import org.apache.ignite.internal.replicator.message.TableAware;
 
 /**
  * Read-write single-row replica request.
  */
 @Transferable(PartitionReplicationMessageGroup.RW_SINGLE_ROW_REPLICA_REQUEST)
-public interface ReadWriteSingleRowReplicaRequest extends SingleRowReplicaRequest, ReadWriteReplicaRequest {
+public interface ReadWriteSingleRowReplicaRequest extends SingleRowReplicaRequest, ReadWriteReplicaRequest, TableAware {
 }

modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/replication/ReadWriteSwapRowReplicaRequest.java

@@ -19,11 +19,12 @@
 
 import org.apache.ignite.internal.network.annotations.Transferable;
 import org.apache.ignite.internal.partition.replicator.network.PartitionReplicationMessageGroup;
+import org.apache.ignite.internal.replicator.message.TableAware;
 
 /**
  * Read-write dual row replica request.
  */
 @Transferable(PartitionReplicationMessageGroup.RW_DUAL_ROW_REPLICA_REQUEST)
-public interface ReadWriteSwapRowReplicaRequest extends SwapRowReplicaRequest, ReadWriteReplicaRequest {
+public interface ReadWriteSwapRowReplicaRequest extends SwapRowReplicaRequest, ReadWriteReplicaRequest, TableAware {
 
 }

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

@@ -504,7 +504,7 @@ private CompletableFuture<TopologyAwareRaftGroupService> createReplicationGroup(
             );
             serviceFutures.add(raftClientFut);
 
-            CompletableFuture<Boolean> replicaFuture = raftClientFut.thenCompose(raftClient -> {
+            CompletableFuture<Replica> replicaFuture = raftClientFut.thenCompose(raftClient -> {
                 try {
                     ReplicaListener listener = new ReplicaListener() {
                         @Override

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

@@ -558,7 +558,7 @@ private void stopLeaseProlongation(ReplicationGroupId groupId, @Nullable String
         });
     }
 
-    private CompletableFuture<Boolean> startReplicaInternal(
+    private CompletableFuture<Replica> startReplicaInternal(
             RaftGroupEventsListener raftGroupEventsListener,
             RaftGroupListener raftGroupListener,
             boolean isVolatileStorage,
@@ -611,7 +611,7 @@ private CompletableFuture<Boolean> startReplicaInternal(
      * @param newConfiguration A configuration for new raft group.
      * @return Future that promises ready new replica when done.
      */
-    public CompletableFuture<Boolean> startReplica(
+    public CompletableFuture<Replica> startReplica(
             RaftGroupEventsListener raftGroupEventsListener,
             RaftGroupListener raftGroupListener,
             boolean isVolatileStorage,
@@ -646,7 +646,6 @@ public CompletableFuture<Boolean> startReplica(
      * Starts a replica. If a replica with the same partition id already exists, the method throws an exception.
      *
      * @param replicaGrpId Replication group id.
-     * @param listener Replica listener.
      * @param snapshotStorageFactory Snapshot storage factory for raft group option's parameterization.
      * @param newConfiguration A configuration for new raft group.
      * @param raftGroupListener Raft group listener for raft group starting.
@@ -657,7 +656,6 @@ public CompletableFuture<Boolean> startReplica(
      */
     public CompletableFuture<Replica> startReplica(
             ReplicationGroupId replicaGrpId,
-            ReplicaListener listener,
             SnapshotStorageFactory snapshotStorageFactory,
             PeersAndLearners newConfiguration,
             RaftGroupListener raftGroupListener,
@@ -680,8 +678,7 @@ public CompletableFuture<Replica> startReplica(
         LOG.info("Replica is about to start [replicationGroupId={}].", replicaGrpId);
 
         Replica newReplica = new ZonePartitionReplicaImpl(
-                replicaGrpId,
-                listener
+                replicaGrpId
         );
 
         CompletableFuture<Replica> replicaFuture = replicas.compute(replicaGrpId, (k, existingReplicaFuture) -> {
@@ -726,7 +723,7 @@ public CompletableFuture<Replica> startReplica(
      */
     @VisibleForTesting
     @Deprecated
-    public CompletableFuture<Boolean> startReplica(
+    public CompletableFuture<Replica> startReplica(
             ReplicationGroupId replicaGrpId,
             PeersAndLearners newConfiguration,
             Consumer<RaftGroupService> updateTableRaftService,
@@ -736,14 +733,13 @@ public CompletableFuture<Boolean> startReplica(
     ) throws NodeStoppingException {
         LOG.info("Replica is about to start [replicationGroupId={}].", replicaGrpId);
 
-        CompletableFuture<Boolean> resultFuture = newRaftClientFut.thenAccept(updateTableRaftService)
-                .thenApply((v) -> true);
-
-        CompletableFuture<ReplicaListener> newReplicaListenerFut = newRaftClientFut.thenApply(createListener);
+        CompletableFuture<TopologyAwareRaftGroupService> resultFuture = newRaftClientFut
+                .thenAccept(updateTableRaftService)
+                .thenCompose((v) -> newRaftClientFut);
 
-        startReplica(replicaGrpId, storageIndexTracker, newReplicaListenerFut);
+        CompletableFuture<ReplicaListener> newReplicaListenerFut = resultFuture.thenApply(createListener);
 
-        return resultFuture;
+        return startReplica(replicaGrpId, storageIndexTracker, newReplicaListenerFut);
     }
 
     /**