IGNITE-23252 ItReplicaLifecycleTest is unstable

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

@@ -225,7 +225,7 @@
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.RepeatedTest;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.TestInfo;
 import org.junit.jupiter.api.Timeout;
@@ -237,7 +237,6 @@
 @ExtendWith({WorkDirectoryExtension.class, ConfigurationExtension.class, ExecutorServiceExtension.class})
 @Timeout(60)
 // TODO: https://issues.apache.org/jira/browse/IGNITE-22522 remove this test after the switching to zone-based replication
-@Disabled("https://issues.apache.org/jira/browse/IGNITE-23252")
 public class ItReplicaLifecycleTest extends BaseIgniteAbstractTest {
     private static final IgniteLogger LOG = Loggers.forClass(ItReplicaLifecycleTest.class);
 
@@ -455,7 +454,6 @@ public void testZoneReplicaListener(TestInfo testInfo) throws Exception {
     }
 
     @Test
-    @Disabled("https://issues.apache.org/jira/browse/IGNITE-22944")
     void testAlterReplicaTrigger(TestInfo testInfo) throws Exception {
         startNodes(testInfo, 3);
 
@@ -641,7 +639,7 @@ void testTableReplicaListenersCreationAfterRebalance(TestInfo testInfo) throws E
         ));
     }
 
-    @Test
+    @RepeatedTest(20)
     void testTableReplicaListenersRemoveAfterRebalance(TestInfo testInfo) throws Exception {
         String zoneName = "TEST_ZONE";
         String tableName = "TEST_TABLE";

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

@@ -24,6 +24,7 @@
 import static java.util.stream.Collectors.toSet;
 import static org.apache.ignite.internal.failure.FailureType.CRITICAL_ERROR;
 import static org.apache.ignite.internal.lang.IgniteStringFormatter.format;
+import static org.apache.ignite.internal.lang.IgniteSystemProperties.getBoolean;
 import static org.apache.ignite.internal.raft.PeersAndLearners.fromAssignments;
 import static org.apache.ignite.internal.replicator.LocalReplicaEvent.AFTER_REPLICA_STARTED;
 import static org.apache.ignite.internal.replicator.LocalReplicaEvent.BEFORE_REPLICA_STOPPED;
@@ -152,6 +153,10 @@ public class ReplicaManager extends AbstractEventProducer<LocalReplicaEvent, Loc
 
     private static final PlacementDriverMessagesFactory PLACEMENT_DRIVER_MESSAGES_FACTORY = new PlacementDriverMessagesFactory();
 
+    /* Feature flag for zone based collocation track */
+    // TODO IGNITE-22115 remove it; direct property name usage because the flag isn't a reason to add replicator as a dependency.
+    public static final boolean ZONE_COLOCATION_IS_ENABLED = getBoolean("IGNITE_ZONE_BASED_REPLICATION", false);
+
     private final IgniteThrottledLogger throttledLog;
 
     /** Busy lock to stop synchronously. */
@@ -656,7 +661,7 @@ private CompletableFuture<HybridTimestamp> stopLeaseProlongation(
     }
 
     /**
-     * Creates and starts a new replica.
+     * Creates and starts a new partition replica.
      *
      * @param raftGroupEventsListener Raft group events listener for raft group starting.
      * @param raftGroupListener Raft group listener for raft group starting.
@@ -680,6 +685,8 @@ public CompletableFuture<Replica> startReplica(
             ReplicationGroupId replicaGrpId,
             PeersAndLearners newConfiguration
     ) throws NodeStoppingException {
+        assert !ZONE_COLOCATION_IS_ENABLED : "Partition replicas are prohibited while zone based colocation is enabled.";
+
         if (!enterBusy()) {
             throw new NodeStoppingException();
         }
@@ -711,7 +718,7 @@ public CompletableFuture<Replica> startReplica(
     }
 
     /**
-     * Starts a replica. If a replica with the same partition id already exists, the method throws an exception.
+     * Starts a zone based replica. If a replica with the same partition id already exists, the method throws an exception.
      *
      * @param replicaGrpId Replication group id.
      * @param snapshotStorageFactory Snapshot storage factory for raft group option's parameterization.

modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java

@@ -48,6 +48,7 @@
 import static org.apache.ignite.internal.event.EventListener.fromConsumer;
 import static org.apache.ignite.internal.hlc.HybridTimestamp.LOGICAL_TIME_BITS_SIZE;
 import static org.apache.ignite.internal.hlc.HybridTimestamp.hybridTimestampToLong;
+import static org.apache.ignite.internal.lang.IgniteSystemProperties.getBoolean;
 import static org.apache.ignite.internal.metastorage.dsl.Conditions.notExists;
 import static org.apache.ignite.internal.metastorage.dsl.Operations.put;
 import static org.apache.ignite.internal.partitiondistribution.PartitionDistributionUtils.calculateAssignmentForPartition;
@@ -265,6 +266,10 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
     /** Table messages factory. */
     private static final PartitionReplicationMessagesFactory TABLE_MESSAGES_FACTORY = new PartitionReplicationMessagesFactory();
 
+    /* Feature flag for zone based collocation track */
+    // TODO IGNITE-22115 remove it
+    public static final boolean ZONE_COLOCATION_IS_ENABLED = getBoolean(PartitionReplicaLifecycleManager.FEATURE_FLAG_NAME, false);
+
     private final TopologyService topologyService;
 
     /** Replica manager. */
@@ -654,11 +659,14 @@ public CompletableFuture<Void> startAsync(ComponentContext componentContext) {
 
             startTables(recoveryRevision, lowWatermark.getLowWatermark());
 
-            processAssignmentsOnRecovery(recoveryRevision);
+            if (!ZONE_COLOCATION_IS_ENABLED) {
+                processAssignmentsOnRecovery(recoveryRevision);
 
-            metaStorageMgr.registerPrefixWatch(new ByteArray(PENDING_ASSIGNMENTS_PREFIX_BYTES), pendingAssignmentsRebalanceListener);
-            metaStorageMgr.registerPrefixWatch(new ByteArray(STABLE_ASSIGNMENTS_PREFIX_BYTES), stableAssignmentsRebalanceListener);
-            metaStorageMgr.registerPrefixWatch(new ByteArray(ASSIGNMENTS_SWITCH_REDUCE_PREFIX_BYTES), assignmentsSwitchRebalanceListener);
+                metaStorageMgr.registerPrefixWatch(new ByteArray(PENDING_ASSIGNMENTS_PREFIX_BYTES), pendingAssignmentsRebalanceListener);
+                metaStorageMgr.registerPrefixWatch(new ByteArray(STABLE_ASSIGNMENTS_PREFIX_BYTES), stableAssignmentsRebalanceListener);
+                metaStorageMgr.registerPrefixWatch(new ByteArray(ASSIGNMENTS_SWITCH_REDUCE_PREFIX_BYTES),
+                        assignmentsSwitchRebalanceListener);
+            }
 
             catalogService.listen(CatalogEvent.TABLE_CREATE, parameters -> onTableCreate((CreateTableEventParameters) parameters));
             catalogService.listen(CatalogEvent.TABLE_CREATE, parameters ->
@@ -903,6 +911,10 @@ private void preparePartitionResourcesAndLoadToZoneReplica(
 
 
     private CompletableFuture<Boolean> onPrimaryReplicaExpired(PrimaryReplicaEventParameters parameters) {
+        // if (ZONE_COLOCATION_IS_ENABLED) {
+        //     return falseCompletedFuture();
+        // }
+
         if (topologyService.localMember().id().equals(parameters.leaseholderId())) {
             TablePartitionId groupId = (TablePartitionId) parameters.groupId();
 
@@ -1267,6 +1279,14 @@ private CompletableFuture<Void> startPartitionAndStartClient(
 
                     mvGc.addStorage(replicaGrpId, partitionUpdateHandlers.gcUpdateHandler);
 
+                    minTimeCollectorService.addPartition(new TablePartitionId(tableId, partId));
+
+                    if (ZONE_COLOCATION_IS_ENABLED) {
+                        return trueCompletedFuture();
+                    }
+
+                    assert !ZONE_COLOCATION_IS_ENABLED : "Partition replicas are prohibited while zone based colocation is enabled.";
+
                     RaftGroupListener raftGroupListener = new PartitionListener(
                             txManager,
                             partitionDataStorage,
@@ -1281,8 +1301,6 @@ private CompletableFuture<Void> startPartitionAndStartClient(
                             minTimeCollectorService
                     );
 
-                    minTimeCollectorService.addPartition(new TablePartitionId(tableId, partId));
-
                     SnapshotStorageFactory snapshotStorageFactory = createSnapshotStorageFactory(replicaGrpId,
                             partitionUpdateHandlers, internalTbl);
 
@@ -1892,6 +1910,9 @@ private CompletableFuture<Void> destroyTableLocally(int tableId) {
                 .collect(toSet());
         metaStorageMgr.removeAll(assignmentKeys);
 
+        CompletableFuture<?> stopReplicaAndDestroyFuture;
+
+        // if (ZONE_COLOCATION_IS_ENABLED) {
         CompletableFuture<?>[] stopReplicaAndDestroyFutures = new CompletableFuture<?>[partitions];
 
         // TODO https://issues.apache.org/jira/browse/IGNITE-19170 Partitions should be stopped on the assignments change
@@ -1902,7 +1923,12 @@ private CompletableFuture<Void> destroyTableLocally(int tableId) {
             stopReplicaAndDestroyFutures[partitionId] = stopAndDestroyPartition(replicationGroupId, table);
         }
 
-        return allOf(stopReplicaAndDestroyFutures)
+        stopReplicaAndDestroyFuture = allOf(stopReplicaAndDestroyFutures);
+        // } else {
+        //     stopReplicaAndDestroyFuture = nullCompletedFuture();
+        // }
+
+        return stopReplicaAndDestroyFuture
                 .thenComposeAsync(
                         unused -> inBusyLockAsync(busyLock, () -> allOf(
                                 internalTable.storage().destroy(),
@@ -2369,6 +2395,7 @@ private CompletableFuture<Void> handleChangePendingAssignmentEvent(
                     }
 
                     assert replicaMgr.isReplicaStarted(replicaGrpId) : "The local node is outside of the replication group ["
+                            + "groupId=[type=" + replicaGrpId.getClass() + ", id=" + replicaGrpId + "]"
                             + ", stable=" + stableAssignments
                             + ", pending=" + pendingAssignments
                             + ", localName=" + localNode().name() + "].";
@@ -2766,12 +2793,16 @@ private CompletableFuture<Void> stopPartition(TablePartitionId tablePartitionId,
 
         CompletableFuture<Boolean> stopReplicaFuture;
 
+        // if (ZONE_COLOCATION_IS_ENABLED) {
+        //     stopReplicaFuture = trueCompletedFuture();
+        // } else {
         try {
             stopReplicaFuture = replicaMgr.stopReplica(tablePartitionId);
         } catch (NodeStoppingException e) {
             // No-op.
             stopReplicaFuture = falseCompletedFuture();
         }
+        // }
 
         return stopReplicaFuture
                 .thenCompose(v -> {