diff --git a/modules/ducktests/README.md b/modules/ducktests/README.md
index b52c22c8a0296..0712f86c88acc 100644
--- a/modules/ducktests/README.md
+++ b/modules/ducktests/README.md
@@ -200,6 +200,7 @@ You can modify test environments at execution time using global flags injected t
 | **AppSpec** | Specifies the class to use for application specifications in Ignite applications. Controls how Ignite applications are configured and started. | ```{"AppSpec": "myapp.services.MyAppSpec"}``` |
 | **IgniteTestContext** | Class name for the test context implementation. Allows customization of test context behavior. | ```{"IgniteTestContext": "myapp.context.CustomTestContext"}``` |
 | **project** | Project/fork name for version handling (e.g., "ignite", "fork"). Used to distinguish between different Ignite variants. Default is "ignite". | ```{"project": "fork"}``` |
+| **mdc_cache_topology_validator** | Whether the MDC tests create their caches with the cache level `MdcTopologyValidator`. Default is True. | ```{"mdc_cache_topology_validator": false}``` |
 
 #### Paths & Directories
 
diff --git a/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/mdc/MdcCacheAwareApplication.java b/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/mdc/MdcCacheAwareApplication.java
index 447486faa9b75..3e14d187c80de 100644
--- a/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/mdc/MdcCacheAwareApplication.java
+++ b/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/mdc/MdcCacheAwareApplication.java
@@ -19,6 +19,7 @@
 
 import java.util.Collections;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Set;
 import com.fasterxml.jackson.databind.JsonNode;
 import org.apache.ignite.IgniteCache;
@@ -29,9 +30,11 @@
 import org.apache.ignite.cache.QueryEntity;
 import org.apache.ignite.cache.affinity.rendezvous.MdcAffinityBackupFilter;
 import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
+import org.apache.ignite.cluster.ClusterNode;
 import org.apache.ignite.configuration.CacheConfiguration;
 import org.apache.ignite.internal.ducktest.tests.dto.IndexedDataRecord;
 import org.apache.ignite.internal.ducktest.utils.IgniteAwareApplication;
+import org.apache.ignite.lang.IgniteBiPredicate;
 import org.apache.ignite.topology.MdcTopologyValidator;
 
 import static org.apache.ignite.IgniteSystemProperties.IGNITE_DATA_CENTER_ID;
@@ -50,8 +53,10 @@
  * <ul>
  *     <li>{@code cacheName} - cache name;</li>
  *     <li>{@code backups} - number of backups; {@code (backups + 1)} must be divisible by {@code dcsNum};</li>
+ *     <li>{@code topologyValidator} - whether to set the cache level {@link MdcTopologyValidator}, default
+ *         {@code true};</li>
  *     <li>{@code mainDc} - main data center for the topology validator (2 DC mode); required, and must be
- *         non-empty, unless {@code datacenters} is given;</li>
+ *         non-empty, unless {@code datacenters} is given or the cache level validator is disabled;</li>
  *     <li>{@code datacenters} - full DC set for majority-based validation (odd DC count mode),
  *         takes precedence over {@code mainDc};</li>
  *     <li>{@code dcsNum} - number of data centers, default 2;</li>
@@ -82,6 +87,9 @@ public abstract class MdcCacheAwareApplication extends IgniteAwareApplication {
     /** */
     protected static final int DFLT_PARTITIONS = 512;
 
+    /** The cache level topology validator is set unless the parameters say otherwise. */
+    protected static final boolean DFLT_CACHE_TOP_VALIDATOR = true;
+
     /** */
     protected static final CacheAtomicityMode DFLT_ATOMICITY_MODE = ATOMIC;
 
@@ -129,6 +137,49 @@ protected <V> CacheConfiguration<Integer, V> mdcCacheConfiguration(JsonNode jNod
 
         int dcsNum = jNode.path("dcsNum").asInt(DFLT_DCS_NUM);
 
+        RendezvousAffinityFunction affinity = new RendezvousAffinityFunction().setPartitions(partitions);
+
+        IgniteBiPredicate<ClusterNode, List<ClusterNode>> backupFilter = backupFilter(jNode, dcsNum, backups);
+
+        if (backupFilter != null)
+            affinity.setAffinityBackupFilter(backupFilter);
+
+        CacheConfiguration<Integer, V> cacheCfg = new CacheConfiguration<Integer, V>()
+            .setName(cacheName)
+            .setCacheMode(cacheMode)
+            .setAtomicityMode(atomicity)
+            .setWriteSynchronizationMode(writeSync)
+            .setBackups(backups)
+            .setReadFromBackup(readFromBackup)
+            .setAffinity(affinity);
+
+        if (jNode.path("topologyValidator").asBoolean(DFLT_CACHE_TOP_VALIDATOR))
+            cacheCfg.setTopologyValidator(mdcTopologyValidator(jNode));
+        else
+            log.info("Cache level topology validator is disabled [cache=" + cacheName + "]");
+
+        return cacheCfg;
+    }
+
+    /**
+     * The affinity backup filter the cache is configured with. A {@link RendezvousAffinityFunction}
+     * holds exactly one, so an override replaces the MDC filter rather than complementing it.
+     *
+     * @param jNode Parameters.
+     * @param dcsNum Number of data centers.
+     * @param backups Number of backups.
+     * @return Affinity backup filter to set, or {@code null} for plain rendezvous affinity.
+     */
+    protected IgniteBiPredicate<ClusterNode, List<ClusterNode>> backupFilter(JsonNode jNode, int dcsNum,
+        int backups) {
+        return new MdcAffinityBackupFilter(dcsNum, backups);
+    }
+
+    /**
+     * @param jNode Parameters.
+     * @return Cache level topology validator compiled from the application parameters.
+     */
+    private MdcTopologyValidator mdcTopologyValidator(JsonNode jNode) {
         MdcTopologyValidator topValidator = new MdcTopologyValidator();
 
         if (jNode.hasNonNull("datacenters")) {
@@ -147,17 +198,7 @@ protected <V> CacheConfiguration<Integer, V> mdcCacheConfiguration(JsonNode jNod
             topValidator.setMainDatacenter(mainDc);
         }
 
-        return new CacheConfiguration<Integer, V>()
-            .setName(cacheName)
-            .setTopologyValidator(topValidator)
-            .setCacheMode(cacheMode)
-            .setAtomicityMode(atomicity)
-            .setWriteSynchronizationMode(writeSync)
-            .setBackups(backups)
-            .setReadFromBackup(readFromBackup)
-            .setAffinity(new RendezvousAffinityFunction()
-                .setPartitions(partitions)
-                .setAffinityBackupFilter(new MdcAffinityBackupFilter(dcsNum, backups)));
+        return topValidator;
     }
 
     /**
diff --git a/modules/ducktests/tests/checks/services/mdc/__init__.py b/modules/ducktests/tests/checks/services/mdc/__init__.py
new file mode 100644
index 0000000000000..ec2014340d78f
--- /dev/null
+++ b/modules/ducktests/tests/checks/services/mdc/__init__.py
@@ -0,0 +1,14 @@
+# 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.
diff --git a/modules/ducktests/tests/checks/services/mdc/check_mdc_cluster.py b/modules/ducktests/tests/checks/services/mdc/check_mdc_cluster.py
new file mode 100644
index 0000000000000..57b408df9d2d0
--- /dev/null
+++ b/modules/ducktests/tests/checks/services/mdc/check_mdc_cluster.py
@@ -0,0 +1,192 @@
+# 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.
+
+"""
+Checks the DC-count dependent parts of the MDC fixture.
+
+The number of data centers decides which MdcTopologyValidator mode the caches are
+configured for, how many backups spread one copy per DC, and which links a partition has
+to cut. All of that is compiled without a cluster, so it is checked without one.
+"""
+
+import pytest
+
+from ignitetest.services.mdc.mdc_cluster import MdcCluster, mdc_topology_params, min_backups, all_pairs, \
+    isolation_pairs, cross_dc_network, per_dc, CACHE_TOP_VALIDATOR_GLOBAL, DCS_2, DCS_3, DC_1, DC_2, DC_3
+
+DELAY_MS = 100
+
+DFLT_DELAY = f"{DELAY_MS}ms"
+
+
+class FakeMdcCluster:
+    """
+    The two members cross_dc_network() reads off an MdcCluster.
+    """
+    def __init__(self, dcs):
+        self.dcs = dcs
+
+    def network_registry(self):
+        """Every DC group is non-empty; the services themselves are irrelevant here."""
+        return {dc: [] for dc in self.dcs}
+
+
+class CheckMdcTopologyParams:
+    """
+    Checks the cache parameters that select the topology validator mode.
+    """
+    def check_even_dc_count_uses_a_main_dc(self):
+        """An even DC set is validated against a main DC, the first one by default."""
+        assert mdc_topology_params(DCS_2) == {"dcsNum": 2, "mainDc": DC_1}
+
+        assert mdc_topology_params(DCS_2, main_dc=DC_2) == {"dcsNum": 2, "mainDc": DC_2}
+
+    def check_odd_dc_count_uses_the_dc_set(self):
+        """An odd DC set is validated by majority, so it carries the DC set instead."""
+        assert mdc_topology_params(DCS_3) == {"dcsNum": 3, "datacenters": [DC_1, DC_2, DC_3]}
+
+    def check_the_two_modes_are_never_mixed(self):
+        """
+        MdcTopologyValidator.checkConfiguration() rejects a main DC alongside an odd DC
+        set, so a main DC must not leak into the majority mode even when one is asked for.
+        """
+        params = mdc_topology_params(DCS_3, main_dc=DC_1)
+
+        assert "mainDc" not in params, "A main DC alongside an odd DC set fails cache startup"
+
+    @pytest.mark.parametrize(["dcs", "expected"], [(DCS_2, 1), (DCS_3, 2)])
+    def check_min_backups_gives_one_copy_per_dc(self, dcs, expected):
+        """(backups + 1) must divide by the DC count - the MdcAffinityBackupFilter contract."""
+        assert min_backups(dcs) == expected
+
+        assert (min_backups(dcs) + 1) % len(dcs) == 0
+
+
+class CheckMdcNetworkLayout:
+    """
+    Checks the DC pairings a partition is expressed in, and the impairment mesh.
+    """
+    def check_all_pairs_covers_the_mesh(self):
+        """Every cross-DC link appears exactly once, in a stable order."""
+        assert all_pairs(DCS_2) == [(DC_1, DC_2)]
+
+        assert all_pairs(DCS_3) == [(DC_1, DC_2), (DC_1, DC_3), (DC_2, DC_3)]
+
+    def check_isolation_pairs_cut_one_dc_only(self):
+        """Isolating a DC cuts its own links and leaves the rest of the mesh intact."""
+        cut = isolation_pairs(DC_3, DCS_3)
+
+        assert cut == [(DC_3, DC_1), (DC_3, DC_2)]
+
+        assert (DC_1, DC_2) not in [tuple(sorted(pair)) for pair in cut], \
+            "The DCs left behind must keep seeing each other"
+
+    def check_symmetric_impairment_reaches_every_pair(self):
+        """One delay argument impairs the whole mesh, not just the first pair."""
+        net = cross_dc_network(None, FakeMdcCluster(DCS_3), delay_ms=DELAY_MS)
+
+        for dc_a, dc_b in all_pairs(DCS_3):
+            cfg = net.network_group_store.get_config(dc_a, dc_b)
+
+            assert cfg is not None and cfg.delay == DFLT_DELAY, f"{dc_a} -> {dc_b} is unimpaired"
+
+            assert net.network_group_store.get_config(dc_b, dc_a) == cfg, "Impairments are bidirectional"
+
+    def check_no_impairment_leaves_the_store_empty(self):
+        """Without delay or loss the manager still owns partitions, but deploys no netem."""
+        net = cross_dc_network(None, FakeMdcCluster(DCS_3))
+
+        assert net.network_group_store.matrix == {}
+
+
+class CheckMdcPerDcCounts:
+    """
+    Checks how the per-DC service counts are spread over the DC set.
+    """
+    def check_a_scalar_count_covers_every_dc(self):
+        """One number means that number of nodes in every DC the cluster spans."""
+        assert per_dc(2, DCS_3) == {DC_1: 2, DC_2: 2, DC_3: 2}
+
+    def check_a_dict_count_is_taken_as_is(self):
+        """An asymmetric layout names only the DCs it populates."""
+        assert per_dc({DC_1: 3}, DCS_3) == {DC_1: 3}
+
+    def check_a_dict_naming_a_foreign_dc_is_rejected(self):
+        """
+        A DC outside the cluster's own set is skipped by network_registry(), so its nodes
+        would run with no impairments and no partition rules - and nothing else would say
+        so. It has to fail where it is declared.
+        """
+        with pytest.raises(AssertionError, match=DC_3):
+            per_dc({DC_1: 1, DC_3: 1}, DCS_2)
+
+
+def _fixture(dcs, top_validator=True):
+    """
+    An MdcCluster carrying only what _with_cache_params() reads - no services are built, so
+    no ducktape cluster is needed. Bypassing the constructor is the point: it pins down how
+    little of the fixture the cache parameter compilation actually depends on.
+    """
+    mdc = MdcCluster.__new__(MdcCluster)
+
+    mdc.dcs = tuple(dcs)
+    mdc.main_dc = dcs[0]
+    mdc.cache_defaults = {"topologyValidator": top_validator}
+
+    return mdc
+
+
+class CheckMdcCacheParams:
+    """
+    Checks the single point every cache of an MDC test is configured from.
+    """
+    def check_an_app_that_creates_the_cache_is_handed_the_dc_set(self):
+        """A cache created by a scenario must agree with the DC set the cluster spans."""
+        params = _fixture(DCS_3)._with_cache_params({"cacheName": "c", "createCache": True})
+
+        assert params["dcsNum"] == 3
+
+        assert params["datacenters"] == [DC_1, DC_2, DC_3]
+
+        assert params["topologyValidator"] is True
+
+    def check_an_app_that_only_uses_the_cache_is_handed_nothing(self):
+        """Cache parameters an application would only ignore must not reach it at all."""
+        params = {"cacheName": "c", "mode": "GET"}
+
+        assert _fixture(DCS_3)._with_cache_params(params) == params
+
+    def check_an_app_that_always_creates_the_cache_needs_no_flag(self):
+        """The generator carries no createCache parameter, so its call site says so instead."""
+        params = _fixture(DCS_2)._with_cache_params({"cacheName": "c"}, creates_cache=True)
+
+        assert params["mainDc"] == DC_1
+
+    def check_an_explicit_parameter_wins(self):
+        """A scenario stays able to override what the fixture injects."""
+        params = _fixture(DCS_2)._with_cache_params({"createCache": True, "mainDc": DC_2})
+
+        assert params["mainDc"] == DC_2
+
+    def check_the_global_reaches_the_cache(self):
+        """
+        The global is only ever read into cache_defaults, so this covers the whole path from
+        --global-json to the application parameters. Its name is part of the README.
+        """
+        assert CACHE_TOP_VALIDATOR_GLOBAL == "mdc_cache_topology_validator"
+
+        params = _fixture(DCS_3, top_validator=False)._with_cache_params({"createCache": True})
+
+        assert params["topologyValidator"] is False
diff --git a/modules/ducktests/tests/checks/services/network_group/check_partition.py b/modules/ducktests/tests/checks/services/network_group/check_partition.py
new file mode 100644
index 0000000000000..434c832533922
--- /dev/null
+++ b/modules/ducktests/tests/checks/services/network_group/check_partition.py
@@ -0,0 +1,157 @@
+# 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.
+
+"""
+Checks how NetworkGroupManager compiles a partition into per-node commands.
+
+A cluster of three or more groups can be cut apart along several links at once, and the
+intermediate topologies of a link-by-link rollout are themselves valid segmentations the
+cluster would react to. So what matters is that every node is configured by exactly one
+SSH round-trip carrying every chain that node takes part in - and no chain it does not.
+"""
+
+import logging
+
+import pytest
+
+from ignitetest.services.network_group.manager import NetworkGroupManager
+from ignitetest.services.network_group.tc_rule_args import partition_chain_name
+
+DC_1, DC_2, DC_3 = "DC1", "DC2", "DC3"
+
+DCS = (DC_1, DC_2, DC_3)
+
+NODES_PER_DC = 2
+
+
+class FakeNode:
+    """A node is only ever an identity and an account here."""
+    def __init__(self, name):
+        self.name = name
+
+    def __repr__(self):
+        return self.name
+
+
+class FakeService:
+    """The single member NetworkGroupManager reads off a registered service."""
+    def __init__(self, nodes):
+        self.nodes = nodes
+
+
+@pytest.fixture(name="manager")
+def _manager(monkeypatch):
+    """
+    A manager over three groups whose SSH layer is replaced by a recorder: yields the
+    manager and the list of (node, command) tasks it submits.
+    """
+    registry = {dc: [FakeService([FakeNode(f"{dc}-{i}") for i in range(NODES_PER_DC)])] for dc in DCS}
+
+    tasks = []
+
+    monkeypatch.setattr(NetworkGroupManager, "_resolve_group_ips", lambda self, group: [f"{group}-ip"])
+    monkeypatch.setattr(NetworkGroupManager, "_ssh_parallel", lambda self, submitted, tag: tasks.extend(submitted))
+    monkeypatch.setattr(NetworkGroupManager, "_log_network", lambda self, log_tag: None)
+
+    yield NetworkGroupManager(logging.getLogger(__name__), None, registry), tasks
+
+
+def _commands_by_node(tasks):
+    return {node.name: cmd for node, cmd in tasks}
+
+
+class CheckNetworkPartition:
+    """
+    Checks the per-node command compilation of single and multi-link partitions.
+    """
+    def check_one_round_trip_per_node(self, manager):
+        """A three way split configures each of the six nodes exactly once."""
+        mgr, tasks = manager
+
+        mgr.enable_network_partitions((DC_1, DC_2), (DC_1, DC_3), (DC_2, DC_3))
+
+        assert len(tasks) == len(DCS) * NODES_PER_DC, "A node must be configured by a single SSH round-trip"
+
+        assert len(_commands_by_node(tasks)) == len(tasks), "Node commands must not be split across tasks"
+
+    def check_a_node_carries_every_chain_it_is_in(self, manager):
+        """Each node gets the chains of its own links, and none of the link it sits out."""
+        mgr, tasks = manager
+
+        mgr.enable_network_partitions((DC_1, DC_2), (DC_1, DC_3), (DC_2, DC_3))
+
+        cmds = _commands_by_node(tasks)
+
+        for dc in DCS:
+            own_chains = {partition_chain_name(dc, other) for other in DCS if other != dc}
+
+            foreign_chain = partition_chain_name(*[other for other in DCS if other != dc])
+
+            for i in range(NODES_PER_DC):
+                cmd = cmds[f"{dc}-{i}"]
+
+                for chain in own_chains:
+                    assert chain in cmd, f"{dc}-{i} is missing chain {chain}"
+
+                assert foreign_chain not in cmd, f"{dc}-{i} took part in the foreign chain {foreign_chain}"
+
+    def check_isolating_one_group_leaves_the_others_connected(self, manager):
+        """Cutting DC3 off touches DC1 and DC2 only through their links to DC3."""
+        mgr, tasks = manager
+
+        mgr.enable_network_partitions((DC_3, DC_1), (DC_3, DC_2))
+
+        cmds = _commands_by_node(tasks)
+
+        assert len(cmds) == len(DCS) * NODES_PER_DC, "Every node of every group takes part in the cut"
+
+        assert partition_chain_name(DC_1, DC_2) not in cmds[f"{DC_1}-0"], \
+            "The groups left behind must keep seeing each other"
+
+        # The isolated group holds both chains; the ones left behind hold only their own.
+        assert partition_chain_name(DC_1, DC_3) in cmds[f"{DC_3}-0"]
+        assert partition_chain_name(DC_2, DC_3) in cmds[f"{DC_3}-0"]
+
+        assert partition_chain_name(DC_2, DC_3) not in cmds[f"{DC_1}-0"]
+
+    def check_single_pair_partition_is_a_multi_partition_of_one(self, manager):
+        """The pairwise entry points stay the one-link case of the batched ones."""
+        mgr, tasks = manager
+
+        mgr.enable_network_partition(DC_1, DC_2)
+
+        cmds = _commands_by_node(tasks)
+
+        assert set(cmds) == {f"{DC_1}-0", f"{DC_1}-1", f"{DC_2}-0", f"{DC_2}-1"}, \
+            "Only the two groups of the cut link are configured"
+
+        chain = partition_chain_name(DC_1, DC_2)
+
+        assert all(chain in cmd for cmd in cmds.values())
+
+    def check_heal_flushes_every_chain_on_every_node(self, manager):
+        """Healing a multi-link split flushes each node's chains in one round-trip."""
+        mgr, tasks = manager
+
+        mgr.disable_network_partitions((DC_1, DC_2), (DC_1, DC_3), (DC_2, DC_3))
+
+        cmds = _commands_by_node(tasks)
+
+        assert len(cmds) == len(DCS) * NODES_PER_DC
+
+        for dc in DCS:
+            for other in DCS:
+                if other != dc:
+                    assert partition_chain_name(dc, other) in cmds[f"{dc}-0"]
diff --git a/modules/ducktests/tests/checks/utils/check_jmx_utils.py b/modules/ducktests/tests/checks/utils/check_jmx_utils.py
new file mode 100644
index 0000000000000..42c24ee3798e1
--- /dev/null
+++ b/modules/ducktests/tests/checks/utils/check_jmx_utils.py
@@ -0,0 +1,124 @@
+# 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.
+
+"""
+Checks the MBean name pattern of a metric registry.
+
+The pattern is grepped against the MBean list of a node, so it is checked here the same way:
+the match has to cover the WHOLE object name, since that name is what the following
+'get -b <name>' is issued against. Every case runs through Python's re and, where a grep is
+installed, through 'grep -E' itself - the pattern is an ERE, not a Python regular expression.
+"""
+import re
+import shutil
+import subprocess
+
+import pytest
+
+from ignitetest.services.utils.jmx_utils import metric_registry_pattern
+
+# MBean names as a node prints them: the properties are ordered alphabetically, which puts
+# the instance name between the group and the name.
+PREFIX = "org.apache:clsLdr=3e2a56ab,group=%s,igniteInstanceName=ducker01,name=%s"
+NO_GROUP_PREFIX = "org.apache:clsLdr=3e2a56ab,igniteInstanceName=ducker01,name=%s"
+
+GREP = shutil.which("grep")
+
+
+def python_grep(pattern, negative_pattern, mbean_name):
+    """
+    :return: The part of the name the patterns leave, as 'grep -E -o | grep -E -v' would return it.
+    """
+    match = re.search(pattern, mbean_name)
+
+    if not match or (negative_pattern and re.search(negative_pattern, match.group())):
+        return None
+
+    return match.group()
+
+
+def real_grep(tmp_path, pattern, negative_pattern, mbean_name):
+    """
+    :return: The same as :func:`python_grep`, but produced by 'grep -E' itself. The patterns go
+             through files, since a quote in a command line argument does not survive Windows.
+    """
+    def grep(flag, grep_pattern, text):
+        pattern_file = tmp_path / f"pattern{flag}"
+        pattern_file.write_bytes((grep_pattern + "\n").encode())
+
+        return subprocess.run([GREP, "-E", flag, "-f", str(pattern_file)], input=text.encode(),
+                              capture_output=True).stdout.decode()
+
+    out = grep("-o", pattern, mbean_name + "\n")
+
+    if out and negative_pattern:
+        out = grep("-v", negative_pattern, out)
+
+    return out.strip() or None
+
+
+# (registry, MBean name, whether the registry's pattern must match that name)
+CASES = [
+    # A purely alphanumeric name is registered without quotes.
+    ("cache.myCache", PREFIX % ("cache", "myCache"), True),
+    ("cache.cache_1", PREFIX % ("cache", "cache_1"), True),
+    # Anything else - a dash is enough - is registered quoted, closing quote included.
+    ("cache.my-cache", PREFIX % ("cache", '"my-cache"'), True),
+    ("cache.mdc-demo-backup-filter", PREFIX % ("cache", '"mdc-demo-backup-filter"'), True),
+    # Only the FIRST dot splits the registry, so the rest of it is the MBean name.
+    ("io.dataregion.default", PREFIX % ("io", '"dataregion.default"'), True),
+    # 'cache' must not pick up 'cacheGroups': the two hold different metrics of the same cache.
+    ("cache.my-cache", PREFIX % ("cacheGroups", '"my-cache"'), False),
+    ("cacheGroups.my-cache", PREFIX % ("cacheGroups", '"my-cache"'), True),
+    # 'name' sorts last, so the pattern ends at the end of the line and a longer name does not answer.
+    ("cache.myCache", PREFIX % ("cache", "myCacheV2"), False),
+    ("cache.my-cache", PREFIX % ("cache", '"my-cache-2"'), False),
+    # The name is matched literally, ERE metacharacters included.
+    ("cache.my.cache", PREFIX % ("cache", '"myXcache"'), False),
+    ("cache.my.cache", PREFIX % ("cache", '"my.cache"'), True),
+    ("cache.a(b)+c", PREFIX % ("cache", '"a(b)+c"'), True),
+    ("cache.a(b)+c", PREFIX % ("cache", '"abbc"'), False),
+    # Characters that are special only right after an opening '[' or '{' stay unescaped.
+    ("cache.a]b}c", PREFIX % ("cache", '"a]b}c"'), True),
+    ("cache.a[b]{2}", PREFIX % ("cache", '"a[b]{2}"'), True),
+    ("cache.a[b]{2}", PREFIX % ("cache", '"abb"'), False),
+    # The exporter escapes '*' inside the quotes.
+    ("cache.a*b", PREFIX % ("cache", r'"a\*b"'), True),
+    # A registry without a dot has no group, and must not pick up the system view of the same name.
+    ("snapshot", NO_GROUP_PREFIX % "snapshot", True),
+    ("snapshot", PREFIX % ("views", "snapshot"), False),
+    ("cdc", PREFIX % ("cdc", "consumer"), False),
+]
+
+
+class CheckMetricRegistryPattern:
+    """
+    Checks that the pattern covers how the JMX exporter really names a registry.
+    """
+    @pytest.mark.parametrize(["registry", "mbean_name", "expected"], CASES)
+    def check_the_pattern_matches_exactly_the_registry(self, registry, mbean_name, expected):
+        assert python_grep(*metric_registry_pattern(registry), mbean_name) == (mbean_name if expected else None)
+
+    @pytest.mark.skipif(GREP is None, reason="grep is not installed")
+    @pytest.mark.parametrize(["registry", "mbean_name", "expected"], CASES)
+    def check_grep_agrees(self, tmp_path, registry, mbean_name, expected):
+        assert real_grep(tmp_path, *metric_registry_pattern(registry), mbean_name) == \
+               (mbean_name if expected else None)
+
+    @pytest.mark.parametrize("registry", ["cache.it's", 'cache.a"b', "cache.a\\b", "cache.a?b"])
+    def check_a_name_the_jmxterm_commands_cannot_carry_is_rejected(self, registry):
+        """The MBean name goes through the shell quoting of the grep and jmxterm commands."""
+        with pytest.raises(ValueError):
+            metric_registry_pattern(registry)
diff --git a/modules/ducktests/tests/ignitetest/services/mdc/mdc_cluster.py b/modules/ducktests/tests/ignitetest/services/mdc/mdc_cluster.py
index a9ccbe4496d3c..a262f956a2012 100644
--- a/modules/ducktests/tests/ignitetest/services/mdc/mdc_cluster.py
+++ b/modules/ducktests/tests/ignitetest/services/mdc/mdc_cluster.py
@@ -14,19 +14,22 @@
 # limitations under the License.
 
 """
-MDC test fixture.
+The fixture spans an arbitrary number of data centers, which selects the
+:class:`MdcTopologyValidator` mode (see :func:`mdc_topology_params`):
 
-    mdc = MdcCluster(self, ignite_version, srv_per_dc=3, runners_per_dc=1)
+    mdc = MdcCluster(self, ignite_version, dcs=DCS_3, srv_per_dc=2, runners_per_dc=1)
 
     with cross_dc_network(self.logger, mdc, delay_ms=20) as net:
-        mdc.start_servers()
-        mdc.generate_data(DC_1, CACHE, 0, 1000, backups=1)
-        mdc.verify_cache_distribution(CACHE, copies_per_dc=1)
+        net.enable_network_partitions(*isolation_pairs(DC_3, mdc.dcs))
+        mdc.verify_segments((DC_1, DC_2), DC_3)
 
-        net.enable_network_partition(DC_1, DC_2)
-        ...
+Globals:
+
+    mdc_cache_topology_validator - whether the MDC caches are created with the cache level
+    MdcTopologyValidator, default true.
 """
-from typing import Dict, List, Optional, Union
+from itertools import combinations
+from typing import Dict, List, Optional, Sequence, Tuple, Union
 
 from ignitetest.services.ignite import IgniteService
 from ignitetest.services.ignite_app import IgniteApplicationService
@@ -41,10 +44,18 @@
 
 DC_1 = "DC1"
 DC_2 = "DC2"
-DCS = (DC_1, DC_2)
+DC_3 = "DC3"
+
+# The two DC layouts the MDC topology validator distinguishes: an even DC count is
+# validated against a main DC, an odd one by a majority of visible DCs.
+DCS_2 = (DC_1, DC_2)
+DCS_3 = (DC_1, DC_2, DC_3)
 
 IGNITE_STARTUP_TIMEOUT_SEC = 90
 
+# Global: set to false to create the MDC caches without the cache level topology validator.
+CACHE_TOP_VALIDATOR_GLOBAL = "mdc_cache_topology_validator"
+
 DATA_CENTER_ATTR = "IGNITE_DATA_CENTER_ID"
 IGNITE_SQL_RETRY_TIMEOUT_ATTR = "IGNITE_SQL_RETRY_TIMEOUT"
 
@@ -67,6 +78,24 @@
 # Every log of a server node, including the ones rotated by a restart.
 ALL_LOGS_GLOB = "ignite*.log*"
 
+# Needed by everything that reads a node metric over JMX - await_rebalance(), the snapshot
+# commands and the MDC safety metrics below among them.
+JMX_METRIC_EXPORTER = "org.apache.ignite.spi.metric.jmx.JmxMetricExporterSpi"
+
+# Per-cache metrics holding the cluster's own verdict on the MDC guarantees. Registered on
+# every server node that carries a DC id, for every cache.
+#
+# The two are not the same statement. The affinity one is about CONFIGURATION - whether the
+# cache is set up to keep a copy of every partition in every DC at all, which is what the
+# MdcAffinityBackupFilter provides. The distribution one is about the CURRENT assignment
+# actually doing so, which a correctly configured cache still fails while some DC has no
+# nodes to place a copy on.
+MDC_SAFE_AFFINITY_METRIC = "IsCacheAffinityConfigurationMdcSafe"
+MDC_SAFE_DISTRIBUTION_METRIC = "IsCachePartitionDistributionSafe"
+
+# A segment of a partitioned cluster: one DC or a group of DCs that still see each other.
+Segment = Union[str, Sequence[str]]
+
 
 def dc_jvm_opts(dc: str) -> List[str]:
     """
@@ -75,11 +104,86 @@ def dc_jvm_opts(dc: str) -> List[str]:
     return [f"-D{DATA_CENTER_ATTR}={dc}", f"-D{IGNITE_SQL_RETRY_TIMEOUT_ATTR}={IGNITE_SQL_RETRY_TIMEOUT_MS}"]
 
 
-def _per_dc(value: Union[int, Dict[str, int]]) -> Dict[str, int]:
+def mdc_topology_params(dcs: Sequence[str], main_dc: Optional[str] = None) -> dict:
+    """
+    Compiles the cache parameters that pin ``MdcTopologyValidator`` and
+    ``MdcAffinityBackupFilter`` to the given DC set.
+
+    The validator has two modes and the DC count picks one: with an EVEN number of DCs a
+    segment stays writable while it sees the main DC (``mainDc``), with an ODD number
+    while it sees a majority of the DC set (``datacenters``). Passing both is rejected by
+    ``MdcTopologyValidator.checkConfiguration()``, so exactly one is emitted here.
+
+    :param dcs: All data centers the cluster spans.
+    :param main_dc: Main DC for the even-count mode, defaults to the first DC. Ignored for
+           an odd DC count, where the validator is majority based.
+    """
+    params = {"dcsNum": len(dcs)}
+
+    if len(dcs) % 2 == 1:
+        params["datacenters"] = list(dcs)
+    else:
+        params["mainDc"] = main_dc if main_dc is not None else dcs[0]
+
+    return params
+
+
+def min_backups(dcs: Sequence[str]) -> int:
+    """
+    :return: Smallest backup count that gives every DC exactly one copy of every partition.
+             ``MdcAffinityBackupFilter`` requires ``(backups + 1)`` to be divisible by the
+             number of DCs, so this is the smallest admissible value at all.
+    """
+    return len(dcs) - 1
+
+
+def all_pairs(dcs: Sequence[str]) -> List[Tuple[str, str]]:
+    """
+    :return: Every unordered DC pair - the full cross-DC mesh.
+    """
+    return list(combinations(dcs, 2))
+
+
+def isolation_pairs(dc: str, dcs: Sequence[str]) -> List[Tuple[str, str]]:
+    """
+    :return: The DC pairs that cut ``dc`` off from every other DC, leaving the rest
+             connected. Feed to :meth:`NetworkGroupManager.enable_network_partitions`.
+    """
+    return [(dc, other) for other in dcs if other != dc]
+
+
+def per_dc(value: Union[int, Dict[str, int]], dcs: Sequence[str]) -> Dict[str, int]:
     """
     Normalizes an int-or-dict per-DC count into a dict, e.g. 3 -> {DC1: 3, DC2: 3}.
+
+    A dict naming a DC the cluster does not span is rejected here rather than left to fail
+    later: such a DC is skipped by :meth:`MdcCluster.network_registry`, so its nodes would
+    run with no impairments and no partition rules while every other call site kept working.
+    """
+    if not isinstance(value, dict):
+        return {dc: value for dc in dcs}
+
+    unknown = sorted(dc for dc in value if dc not in dcs)
+
+    assert not unknown, \
+        f"Per-DC counts name data centers the cluster does not span [unknown={unknown}, dcs={list(dcs)}]"
+
+    return dict(value)
+
+
+def _as_segment(segment: Segment) -> Tuple[str, ...]:
     """
-    return dict(value) if isinstance(value, dict) else {dc: value for dc in DCS}
+    Normalizes a single DC name or a collection of DC names into a tuple of DC names.
+    """
+    return (segment,) if isinstance(segment, str) else tuple(segment)
+
+
+def _fmt_segment(segment: Tuple[str, ...]) -> str:
+    """
+    :return: Segment rendered for an assertion message, e.g. "DC1+DC2" - a Python tuple
+             reads poorly in the middle of one, and a single DC renders as itself.
+    """
+    return "+".join(segment)
 
 
 class MdcCluster:
@@ -89,28 +193,42 @@ class MdcCluster:
 
     :param test: The ducktape test instance.
     :param ignite_version: Ignite version string.
+    :param dcs: Data centers the cluster spans, two by default. The count selects the
+           topology validator mode - see :func:`mdc_topology_params`.
+    :param main_dc: Main DC for an even-sized DC set, defaults to the first DC.
     :param srv_per_dc: Servers per DC, an int or a per-DC dict (asymmetric DCs).
     :param runners_per_dc: Reusable run-to-completion app services per DC (generator,
            checkers, load bursts). An int or a per-DC dict.
     :param loaders_per_dc: Dedicated background load app services per DC. They run
            concurrently with runner apps, hence separate containers.
     :param client_connector: Whether to expose the thin client connector on servers.
+    :param jmx_metrics: Whether to export the node metrics over JMX. Required by everything
+           that reads one - see :meth:`cache_mdc_metrics`.
     """
-    def __init__(self, test, ignite_version: str, srv_per_dc: Union[int, Dict[str, int]] = 3,
+    def __init__(self, test, ignite_version: str, dcs: Sequence[str] = DCS_2,
+                 main_dc: Optional[str] = None,
+                 srv_per_dc: Union[int, Dict[str, int]] = 3,
                  runners_per_dc: Union[int, Dict[str, int]] = 1,
                  loaders_per_dc: Union[int, Dict[str, int]] = 0,
                  client_connector: bool = False,
+                 jmx_metrics: bool = False,
                  network_timeout: int = 5_000,
                  tcp_connect_timeout: int = 5_000):
         self.test_context = test.test_context
         self.logger = test.logger
 
-        # A single discovery SPI (hence a single ip finder) shared by both DCs' server
-        # services is what makes the two DCs form ONE cluster: prepare_on_start()
-        # memoizes the addresses of the first started DC into the shared ip finder, so
-        # the second DC discovers through the first DC's nodes, and restart() re-joins
-        # the same way. Restarting the first started DC itself is the one case this
-        # breaks - see sync_service_discovery().
+        self.dcs = tuple(dcs)
+
+        assert len(self.dcs) >= 2, f"An MDC cluster spans at least two data centers [dcs={self.dcs}]"
+
+        self.main_dc = main_dc if main_dc is not None else self.dcs[0]
+
+        # A single discovery SPI (hence a single ip finder) shared by all DCs' server
+        # services is what makes the DCs form ONE cluster: prepare_on_start() memoizes the
+        # addresses of the first started DC into the shared ip finder, so every later DC
+        # discovers through the first DC's nodes, and restart() re-joins the same way.
+        # Restarting the first started DC itself is the one case this breaks - see
+        # sync_service_discovery().
         cfg_kwargs = {
             "version": IgniteVersion(ignite_version),
             "discovery_spi": TcpDiscoverySpi(),
@@ -121,25 +239,27 @@ def __init__(self, test, ignite_version: str, srv_per_dc: Union[int, Dict[str, i
         if client_connector:
             cfg_kwargs["client_connector_configuration"] = ClientConnectorConfiguration()
 
+        if jmx_metrics:
+            # A fresh set, never the shared mutable default of IgniteConfiguration.
+            cfg_kwargs["metric_exporters"] = {JMX_METRIC_EXPORTER}
+
         self.ignite_config = IgniteConfiguration(**cfg_kwargs)
 
-        self.srv_per_dc = _per_dc(srv_per_dc)
+        self.srv_per_dc = per_dc(srv_per_dc, self.dcs)
 
         self.servers: Dict[str, IgniteService] = {
-            dc: IgniteService(self.test_context, self.ignite_config, num_nodes=num, jvm_opts=dc_jvm_opts(dc),
-                              startup_timeout_sec=IGNITE_STARTUP_TIMEOUT_SEC)
-            for dc, num in self.srv_per_dc.items() if num > 0}
+            dc: self._server_service(dc, num) for dc, num in self.srv_per_dc.items() if num > 0}
 
         self.runners: Dict[str, List[IgniteApplicationService]] = {
             dc: [self._app_service(dc) for _ in range(num)]
-            for dc, num in _per_dc(runners_per_dc).items()}
+            for dc, num in per_dc(runners_per_dc, self.dcs).items()}
 
         self.loaders: Dict[str, List[IgniteApplicationService]] = {
             dc: [self._app_service(dc) for _ in range(num)]
-            for dc, num in _per_dc(loaders_per_dc).items()}
+            for dc, num in per_dc(loaders_per_dc, self.dcs).items()}
 
         # Extra services (e.g. thin clients) registered into a DC's network group.
-        self.extras: Dict[str, List] = {dc: [] for dc in DCS}
+        self.extras: Dict[str, List] = {dc: [] for dc in self.dcs}
 
         # App services that have been started at least once: the first start is clean,
         # subsequent ones preserve work dirs (and logs - hence unique result prefixes).
@@ -148,21 +268,65 @@ def __init__(self, test, ignite_version: str, srv_per_dc: Union[int, Dict[str, i
         # Admissibility checks run on reusable services, so each check needs a unique result prefix.
         self._adm_checks = 0
 
+        # Cache parameters applied to every cache this fixture creates, unless a call overrides them.
+        self.cache_defaults = {
+            "topologyValidator": self.test_context.globals.get(CACHE_TOP_VALIDATOR_GLOBAL, True)
+        }
+
+        self.logger.info(f"MDC cache defaults [{self.cache_defaults}]")
+
+    def _server_service(self, dc: str, num_nodes: int) -> IgniteService:
+        """
+        Builds the server service of one DC. The single place a server command line is put
+        together.
+        """
+        return IgniteService(self.test_context, self.ignite_config, num_nodes=num_nodes,
+                             jvm_opts=dc_jvm_opts(dc), startup_timeout_sec=IGNITE_STARTUP_TIMEOUT_SEC)
+
+    def dc_servers(self, dc: str) -> List[IgniteService]:
+        """
+        :return: All server services of the given DC, one per DC in this fixture.
+        """
+        return [self.servers[dc]] if dc in self.servers else []
+
+    def all_servers(self) -> List[IgniteService]:
+        """
+        :return: Every server service of the cluster, DCs in order.
+        """
+        return [svc for dc in sorted(self.servers) for svc in self.dc_servers(dc)]
+
+    @property
+    def min_backups(self) -> int:
+        """
+        :return: Smallest backup count giving every DC one copy of every partition
+                 (2 for a three DC cluster, 1 for a two DC one).
+        """
+        return min_backups(self.dcs)
+
+    def topology_params(self) -> dict:
+        """
+        :return: Cache parameters pinning the topology validator and the affinity backup
+                 filter to this cluster's DC set.
+        """
+        return mdc_topology_params(self.dcs, self.main_dc)
+
     def sync_service_discovery(self):
         """
         Points every server service at a discovery SPI covering all DCs.
 
         Required before restarting the FIRST started DC: the shared ip finder holds only
         that DC's addresses, so after a full stop its nodes would seed off themselves and
-        form a separate cluster instead of rejoining the surviving DC.
+        form a separate cluster instead of rejoining the surviving DCs.
         """
-        discovery_spi = from_ignite_services(list(self.servers.values()))
+        discovery_spi = from_ignite_services(self.all_servers())
 
-        for service in self.servers.values():
+        for service in self.all_servers():
             service.config = service.config._replace(discovery_spi=discovery_spi)
 
     def _app_service(self, dc: str) -> IgniteApplicationService:
-        client_cfg = self.ignite_config._replace(client_mode=True, discovery_spi=from_ignite_cluster(self.servers[dc]))
+        # Seeding off the DC's first server service is enough: all of them are one cluster.
+        client_cfg = self.ignite_config._replace(client_mode=True,
+                                                 discovery_spi=from_ignite_cluster(self.dc_servers(dc)[0]))
 
         return IgniteApplicationService(self.test_context, client_cfg, jvm_opts=dc_jvm_opts(dc))
 
@@ -180,11 +344,8 @@ def network_registry(self) -> Dict[str, List]:
         """
         registry = {}
 
-        for dc in DCS:
-            services = []
-
-            if dc in self.servers:
-                services.append(self.servers[dc])
+        for dc in self.dcs:
+            services = list(self.dc_servers(dc))
 
             services += self.runners.get(dc, [])
             services += self.loaders.get(dc, [])
@@ -208,9 +369,9 @@ def describe(self) -> List[str]:
         """
         lines = ["DATA CENTERS"]
 
-        for dc in DCS:
+        for dc in self.dcs:
             roles = [(label, [node.account.hostname for svc in services for node in svc.nodes])
-                     for label, services in (("server", [self.servers[dc]] if dc in self.servers else []),
+                     for label, services in (("server", self.dc_servers(dc)),
                                              ("runner", self.runners.get(dc, [])),
                                              ("loader", self.loaders.get(dc, [])),
                                              ("extra", self.extras.get(dc, [])))]
@@ -231,33 +392,58 @@ def thin_client_addresses(self) -> List[str]:
         """
         port = self.ignite_config.client_connector_configuration.port
 
-        return [f"{node.account.hostname}:{port}"
-                for dc in sorted(self.servers) for node in self.servers[dc].nodes]
+        return [f"{node.account.hostname}:{port}" for svc in self.all_servers() for node in svc.nodes]
 
     def start_servers(self):
         """
         Starts all server services.
         """
-        for dc in sorted(self.servers):
-            self.servers[dc].start()
+        for svc in self.all_servers():
+            svc.start()
 
     def stop_servers(self):
         """
         Stops all server services.
         """
-        for dc in sorted(self.servers):
-            self.servers[dc].stop()
+        for svc in self.all_servers():
+            svc.stop()
+
+    def stop_dcs(self, *dcs: str):
+        """
+        Stops the server services of the given DCs, in the order given - a data center
+        outage, as opposed to the network partition :func:`cross_dc_network` produces.
+        """
+        for dc in dcs:
+            for svc in self.dc_servers(dc):
+                svc.stop()
+
+    def start_dcs(self, *dcs: str, clean: bool = False, await_rebalance: bool = True):
+        """
+        Starts the given DCs back, by default preserving their persistence, and waits until
+        the cluster has rebalanced onto them.
+
+        Every DC is started before the first wait, so that their joins are not serialized
+        behind each other's rebalance.
+
+        Restarting the FIRST started DC needs :meth:`sync_service_discovery` beforehand.
+        """
+        for dc in dcs:
+            for svc in self.dc_servers(dc):
+                svc.start(clean=clean)
+
+        if await_rebalance:
+            for dc in dcs:
+                for svc in self.dc_servers(dc):
+                    svc.await_rebalance()
 
     def restart(self, dc: str, clean: bool = False, await_rebalance: bool = True):
         """
         Restarts a whole DC preserving its persistence (the pattern used to rejoin a
-        read-only half-ring back into the main cluster after a partition heals).
+        read-only segment back into the main cluster after a partition heals).
         """
-        self.servers[dc].stop()
-        self.servers[dc].start(clean=clean)
+        self.stop_dcs(dc)
 
-        if await_rebalance:
-            self.servers[dc].await_rebalance()
+        self.start_dcs(dc, clean=clean, await_rebalance=await_rebalance)
 
     def run_app(self, dc: str, java_class: str, params: dict, runner: int = 0) -> IgniteApplicationService:
         """
@@ -276,7 +462,7 @@ def run_service(self, svc: IgniteApplicationService, params: dict,
         if java_class is not None:
             svc.java_class_name = java_class
 
-        svc.params = params
+        svc.params = self._with_cache_params(params)
 
         svc.start(clean=self._first_start(svc))
         svc.wait()
@@ -288,12 +474,14 @@ def start_loader(self, dc: str, params: dict, loader: int = 0,
                      java_class: str = LOAD_APP) -> IgniteApplicationService:
         """
         Starts a background load application (runs until stopped). Any exception raised
-        by the application surfaces in :meth:`stop_loader`.
+        by the application surfaces in :meth:`stop_loader`. A load that creates the cache
+        (``createCache``) has the MDC cache parameters injected - see
+        :meth:`_with_cache_params`.
         """
         svc = self.loaders[dc][loader]
 
         svc.java_class_name = java_class
-        svc.params = params
+        svc.params = self._with_cache_params(params)
 
         svc.start(clean=self._first_start(svc))
 
@@ -310,6 +498,24 @@ def stop_loader(self, dc: str, loader: int = 0) -> IgniteApplicationService:
 
         return svc
 
+    def _with_cache_params(self, params: dict, creates_cache: bool = False) -> dict:
+        """
+        Injects everything the MDC cache is configured from - the topology validator mode,
+        the DC count the affinity backup filter needs, and :attr:`cache_defaults` - into
+        the parameters of an application that creates it, so no call site can configure a
+        cache that disagrees with the DC set. Explicit parameters still win.
+
+        The single injection point for all of it: an application that does not create the
+        cache is handed none of it, since it would only ever be ignored.
+
+        :param creates_cache: Whether the application always creates the cache. The ones
+               that decide at run time say so with a ``createCache`` parameter instead.
+        """
+        if not (creates_cache or params.get("createCache")):
+            return params
+
+        return {**self.topology_params(), **self.cache_defaults, **params}
+
     def _first_start(self, svc) -> bool:
         first = id(svc) not in self._started_apps
 
@@ -317,17 +523,25 @@ def _first_start(self, svc) -> bool:
 
         return first
 
-    def generate_data(self, dc: str, cache_name: str, from_idx: int, to_idx: int, backups: int,
-                      main_dc: str = DC_1, sql_mode: bool = False, **cache_params) -> IgniteApplicationService:
+    def generate_data(self, dc: str, cache_name: str, from_idx: int, to_idx: int, backups: Optional[int] = None,
+                      sql_mode: bool = False, **cache_params) -> IgniteApplicationService:
         """
         Creates the MDC cache (if absent) and populates keys ``[from_idx, to_idx)``.
         Extra cache parameters (``atomicity``, ``writeSync``, ``readFromBackup``,
-        ``partitions``, ...) are passed through to the cache configuration builder.
+        ``partitions``, ...) are passed through to the cache configuration builder, on top
+        of the MDC cache parameters - see :meth:`_with_cache_params`.
+
+        :param backups: Backup count, by default the smallest one that gives every DC a
+               single copy of every partition (see :attr:`min_backups`).
         """
-        params = {"cacheName": cache_name, "backups": backups, "mainDc": main_dc,
-                  "from": from_idx, "to": to_idx, "sqlMode": sql_mode, **cache_params}
+        params = {"cacheName": cache_name,
+                  "backups": self.min_backups if backups is None else backups,
+                  "from": from_idx, "to": to_idx, "sqlMode": sql_mode,
+                  **cache_params}
 
-        return self.run_app(dc, GENERATOR_APP, params)
+        # The generator always creates the cache, so it carries no createCache parameter
+        # for _with_cache_params() to key off.
+        return self.run_app(dc, GENERATOR_APP, self._with_cache_params(params, creates_cache=True))
 
     def check_data(self, dc: str, cache_name: str, from_idx: int, to_idx: int) -> Optional[IgniteApplicationService]:
         """
@@ -347,8 +561,8 @@ def check_data(self, dc: str, cache_name: str, from_idx: int, to_idx: int) -> Op
     def check_put_admissibility(self, dc: str, cache_name: str, admissible: bool,
                                 key_offset: int = 1_000_000, probes: int = 100) -> IgniteApplicationService:
         """
-        Verifies that put load from the given DC is admissible (primary DC visible) or
-        rejected by the topology validator (read-only DC). A PUT burst of the load
+        Verifies that put load from the given DC is admissible (the segment passes the
+        topology validator) or rejected by it (read-only segment). A PUT burst of the load
         application: an admissible check fails fast on the first rejected put, an
         inadmissible check fails if any of the probe puts succeeds.
 
@@ -366,18 +580,66 @@ def run_load(self, dc: str, mode: str, cache_name: str, result_prefix: str,
         """
         Runs a load burst (see ``MdcContinuousLoadApplication``) and returns the service.
         ``result_prefix`` must be unique per burst because runner services are reused.
+        A burst that creates the cache (``createCache``) has the MDC cache parameters
+        injected - see :meth:`_with_cache_params`.
         """
         load_params = {"mode": mode, "cacheName": cache_name, "resultPrefix": result_prefix, **params}
 
         return self.run_app(dc, LOAD_APP, load_params, runner=runner)
 
-    def control(self, dc: str = DC_1) -> ControlUtility:
+    def control(self, dc: Optional[str] = None) -> ControlUtility:
+        """
+        :return: Control utility bound to the given DC's servers, the first DC by default.
         """
-        :return: Control utility bound to the given DC's servers.
+        return ControlUtility(self.dc_servers(dc if dc is not None else self.dcs[0])[0])
+
+    def cache_mdc_metrics(self, cache_name: str, dc: Optional[str] = None) -> Dict[str, bool]:
         """
-        return ControlUtility(self.servers[dc])
+        Reads the cache's MDC safety metrics off a server node over JMX - see
+        :data:`MDC_SAFE_AFFINITY_METRIC` and :data:`MDC_SAFE_DISTRIBUTION_METRIC` for what
+        each of them claims.
+
+        Requires the cluster to have been built with ``jmx_metrics=True``, since the metrics
+        are only exposed by the JMX metric exporter.
 
-    def verify_cache_distribution(self, cache_name: str, copies_per_dc: Optional[int] = None, dc: str = DC_1):
+        :param cache_name: Cache to read the metrics of.
+        :param dc: DC whose node answers, the first one by default. Every server node reports
+               the same verdict, so this only matters for a partitioned cluster - where each
+               segment answers about the topology IT can see.
+        :return: Metric name -> value.
+        """
+        node = next(node for svc in self.dc_servers(dc if dc is not None else self.dcs[0])
+                    for node in svc.alive_nodes)
+
+        mbean = node.metric_registry_mbean(f'cache.{cache_name}')
+
+        return {name: mbean.bool_value(name)
+                for name in (MDC_SAFE_AFFINITY_METRIC, MDC_SAFE_DISTRIBUTION_METRIC)}
+
+    def verify_cache_mdc_metrics(self, cache_name: str, affinity_safe: Optional[bool] = None,
+                                 distribution_safe: Optional[bool] = None, dc: Optional[str] = None):
+        """
+        Verifies the MDC safety metrics of a cache against what the scenario expects. Both
+        expectations are optional: a metric left as None is only reported, which is how a
+        scenario reads out a value whose verdict depends on the state of the topology rather
+        than on the point being made.
+
+        :return: The metrics that were read.
+        """
+        metrics = self.cache_mdc_metrics(cache_name, dc)
+
+        self.logger.info(f"MDC safety metrics [cache={cache_name}, dc={dc}, {metrics}]")
+
+        for name, expected in ((MDC_SAFE_AFFINITY_METRIC, affinity_safe),
+                               (MDC_SAFE_DISTRIBUTION_METRIC, distribution_safe)):
+            if expected is not None:
+                assert metrics[name] == expected, \
+                    f"{name} should be {expected} [cache={cache_name}, actual={metrics[name]}]"
+
+        return metrics
+
+    def verify_cache_distribution(self, cache_name: str, copies_per_dc: Optional[int] = None,
+                                  dc: Optional[str] = None):
         """
         Verifies that every partition of the cache has an OWNING copy in every DC, and
         optionally that each DC holds exactly ``copies_per_dc`` copies.
@@ -387,77 +649,98 @@ def verify_cache_distribution(self, cache_name: str, copies_per_dc: Optional[int
         distribution = self.control(dc).cache_distribution(cache_names=cache_name, user_attributes=DATA_CENTER_ATTR)
 
         assert_cross_dc_distribution_by_attribute(distribution, dc_attr=DATA_CENTER_ATTR,
-                                                  expected_dcs=DCS, copies_per_dc=copies_per_dc)
+                                                  expected_dcs=self.dcs, copies_per_dc=copies_per_dc)
 
         return distribution
 
     def verify_split_brain(self):
         """
-        Verifies that after the network partition the cluster has split into two independent
-        half-rings: their baselines don't intersect and each half elected its own coordinator.
+        Verifies that the network partition split the cluster into as many independent
+        segments as there are DCs, i.e. every DC ended up on its own.
+        """
+        self.verify_segments(*self.dcs)
+
+    def verify_segments(self, *segments: Segment):
+        """
+        Verifies that the cluster has split into exactly the given independent segments:
+        every segment is healthy on its own, no two segments share a baseline node, and
+        each segment elected its own coordinator.
+
+        A segment is a DC name or a collection of DC names that still see each other, e.g.
+        ``verify_segments((DC_1, DC_2), DC_3)`` for a cluster with DC3 cut off.
         """
-        for dc in DCS:
-            self.verify_half_ring_healthy(dc)
+        normalized = [_as_segment(segment) for segment in segments]
 
-        state = {dc: self.control(dc).cluster_state() for dc in DCS}
+        # The state each segment is checked healthy against is the same one its baseline and
+        # coordinator are read from: a partitioned segment answers control.sh over the very
+        # links the test just cut, so it is fetched once per segment and passed around.
+        states = {segment: self.verify_segment_healthy(segment) for segment in normalized}
 
-        baselines = {dc: {node.consistent_id for node in state[dc].baseline} for dc in DCS}
+        baselines = {segment: {node.consistent_id for node in states[segment].baseline} for segment in normalized}
 
-        common_nodes = baselines[DC_1] & baselines[DC_2]
+        for seg_a, seg_b in combinations(normalized, 2):
+            common_nodes = baselines[seg_a] & baselines[seg_b]
 
-        assert not common_nodes, \
-            f"Half-ring baselines should not intersect " \
-            f"[common={sorted(common_nodes)}, dc1={sorted(baselines[DC_1])}, dc2={sorted(baselines[DC_2])}]"
+            assert not common_nodes, \
+                f"Segment baselines should not intersect [common={sorted(common_nodes)}, " \
+                f"{_fmt_segment(seg_a)}={sorted(baselines[seg_a])}, " \
+                f"{_fmt_segment(seg_b)}={sorted(baselines[seg_b])}]"
 
-        for dc in DCS:
-            coordinator = state[dc].coordinator
+        coordinators = {}
 
-            assert coordinator, f"Coordinator is not found in {dc} half-ring baseline output!"
+        for segment in normalized:
+            coordinator = states[segment].coordinator
 
-            assert coordinator.consistent_id in baselines[dc], \
-                f"{dc} coordinator should belong to its own half-ring baseline " \
-                f"[coordinator={coordinator.consistent_id}, baseline={sorted(baselines[dc])}]"
+            assert coordinator, \
+                f"Coordinator is not found in the {_fmt_segment(segment)} segment baseline output!"
 
-        assert state[DC_1].coordinator.consistent_id != state[DC_2].coordinator.consistent_id, \
-            f"Half-rings should have different coordinators " \
-            f"[coordinator={state[DC_1].coordinator.consistent_id}]"
+            assert coordinator.consistent_id in baselines[segment], \
+                f"{_fmt_segment(segment)} coordinator should belong to its own segment baseline " \
+                f"[coordinator={coordinator.consistent_id}, baseline={sorted(baselines[segment])}]"
 
-    def verify_half_ring_healthy(self, dc: str):
+            coordinators[_fmt_segment(segment)] = coordinator.consistent_id
+
+        assert len(set(coordinators.values())) == len(normalized), \
+            f"Every segment should have elected its own coordinator [coordinators={coordinators}]"
+
+    def verify_segment_healthy(self, segment: Segment):
         """
-        Verifies that a half-ring is fully alive, ACTIVE, and its baseline matches its size.
+        Verifies that a segment is fully alive, ACTIVE, and its baseline covers exactly
+        the servers of the DCs it consists of - and nothing else.
+
+        :return: The ClusterState the segment was verified against, so that a caller
+                 asserting further on it (see :meth:`verify_segments`) needs no second
+                 control.sh round-trip into a segment that may be cut off.
         """
-        exp_alive_nodes = self.srv_per_dc[dc]
-        act_alive_nodes = len(self.servers[dc].alive_nodes)
+        dcs = _as_segment(segment)
+
+        name = _fmt_segment(dcs)
+
+        # get(): a per-DC dict is allowed to name only the DCs it populates, and
+        # verify_whole_cluster_healthy() asks about every DC the cluster spans.
+        exp_alive_nodes = sum(self.srv_per_dc.get(dc, 0) for dc in dcs)
+        act_alive_nodes = sum(len(svc.alive_nodes) for dc in dcs for svc in self.dc_servers(dc))
 
         assert act_alive_nodes == exp_alive_nodes, \
-            f"{exp_alive_nodes} nodes should be alive in {dc}! [actual={act_alive_nodes}]"
+            f"{exp_alive_nodes} nodes should be alive in {name}! [actual={act_alive_nodes}]"
 
-        cluster_state = self.control(dc).cluster_state()
+        cluster_state = self.control(dcs[0]).cluster_state()
 
         assert "ACTIVE" == cluster_state.state, \
-            f"{dc} half-ring state should remain ACTIVE [actual={cluster_state.state}]"
+            f"{name} segment state should remain ACTIVE [actual={cluster_state.state}]"
 
         assert len(cluster_state.baseline) == exp_alive_nodes, \
-            f"{dc} half-ring baseline is not expected " \
+            f"{name} segment baseline is not expected " \
             f"[exp={exp_alive_nodes}, actual_baseline={cluster_state.baseline}]"
 
+        return cluster_state
+
     def verify_whole_cluster_healthy(self):
         """
-        Verifies that both DCs form a single ACTIVE cluster: every server node is alive
-        and the baseline seen from DC1 covers all servers of both DCs.
+        Verifies that all DCs form a single ACTIVE cluster: every server node is alive
+        and the baseline seen from the first DC covers all servers of every DC.
         """
-        exp_total = sum(self.srv_per_dc.values())
-
-        act_alive = sum(len(self.servers[dc].alive_nodes) for dc in self.servers)
-
-        assert act_alive == exp_total, f"All {exp_total} server nodes should be alive [actual={act_alive}]"
-
-        cluster_state = self.control(DC_1).cluster_state()
-
-        assert "ACTIVE" == cluster_state.state, f"Cluster should be ACTIVE [actual={cluster_state.state}]"
-
-        assert len(cluster_state.baseline) == exp_total, \
-            f"Cluster baseline should cover both DCs [exp={exp_total}, actual={cluster_state.baseline}]"
+        self.verify_segment_healthy(self.dcs)
 
     def verify_servers_log_clean(self):
         """
@@ -465,10 +748,10 @@ def verify_servers_log_clean(self):
         were detected, no PME hang and no lost partitions were reported.
         """
         for pattern in (LRT_PATTERN, PME_FREEZE_PATTERN, LOST_PARTITIONS_PATTERN, ASSERTION_ERROR_PATTERN):
-            for svc in self.servers.values():
+            for svc in self.all_servers():
                 svc.check_event_absent(pattern, log_file=ALL_LOGS_GLOB)
 
-    def verify_no_hanging_txs(self, dc: str = DC_1, try_kill_hanging_tx: bool = False):
+    def verify_no_hanging_txs(self, dc: Optional[str] = None, try_kill_hanging_tx: bool = False):
         """
         Verifies that no active transactions are left on the cluster.
         """
@@ -509,20 +792,25 @@ def result_bool(svc: IgniteApplicationService, name: str) -> bool:
 def cross_dc_network(logger, mdc: MdcCluster, delay_ms: Optional[int] = None,
                      loss: Optional[float] = None) -> NetworkGroupManager:
     """
-    Builds a :class:`NetworkGroupManager` (context manager) for the cluster with symmetric
-    DC1 <-> DC2 impairments. With no impairments the manager still owns partition
+    Builds a :class:`NetworkGroupManager` (context manager) for the cluster, applying the
+    same impairment to every DC pair. With no impairments the manager still owns partition
     enable/disable and the final network cleanup.
 
+    A cluster whose links are not all alike needs no fixture support: build the
+    :class:`NetworkGroupStore` and construct the manager directly, the registry is all it
+    takes from here - ``NetworkGroupManager(logger, store, mdc.network_registry())``.
+
     :param delay_ms: One-way cross-DC latency in milliseconds (the effective RTT is twice
            that, since netem delay is applied on egress in both directions).
     :param loss: Cross-DC packet loss fraction in [0.0, 1.0].
     """
-    store = NetworkGroupStore()
-
     cfg = CrossNetworkGroupConfiguration(delay=f"{delay_ms}ms" if delay_ms is not None else None, loss=loss)
 
+    store = NetworkGroupStore()
+
     if not cfg.is_empty:
-        store.set_config(DC_1, DC_2, cfg)
+        for dc_a, dc_b in all_pairs(mdc.dcs):
+            store.set_config(dc_a, dc_b, cfg)
 
     return NetworkGroupManager(logger, store, mdc.network_registry())
 
@@ -545,45 +833,65 @@ def assert_cross_dc_distribution_by_attribute(distribution, dc_attr, expected_dc
     def dc_of(copy):
         return copy.user_attributes.get(dc_attr)
 
-    _assert_cross_dc(distribution, set(expected_dcs), dc_of, owning_only, copies_per_dc,
-                     layout_hint=f"DC attribute: {dc_attr}, expected DCs: {sorted(expected_dcs)}")
+    assert_spread(distribution, set(expected_dcs), dc_of, owning_only, copies_per_dc, label="DC",
+                  layout_hint=f"DC attribute: {dc_attr}, expected DCs: {sorted(expected_dcs)}")
 
 
-def _assert_cross_dc(distribution, expected_dcs, dc_of, owning_only, copies_per_dc, layout_hint):
+def assert_spread(distribution, expected_groups, group_of, owning_only, copies_per_group, label,
+                  layout_hint):
+    """
+    Asserts that every partition of every cache group has a copy in every expected group,
+    where a group is whatever ``group_of(copy)`` returns.
+
+    Public because the grouping is the only thing that varies: the DC spread above is one
+    ``group_of``, and an affinity backup filter that spreads copies by something finer - a
+    cell, an availability zone, a tuple of node attributes - is another. Such a check reuses
+    this rather than walking the distribution again.
+
+    :param distribution: CacheDistribution requested with the attributes ``group_of`` reads.
+    :param expected_groups: Group keys that must each own a copy of every partition.
+    :param group_of: Copy -> its group key.
+    :param owning_only: Count only copies in OWNING state as present.
+    :param copies_per_group: If set, each group must hold exactly this many copies.
+    :param label: What a group is called in the assertion message, e.g. "DC".
+    :param layout_hint: Line appended to the message, naming the layout that was expected.
+    """
     violations = []
 
     for group in distribution.groups.values():
         for part, copies in sorted(group.partitions.items()):
             counted = [c for c in copies if not owning_only or c.state == "OWNING"]
 
-            per_dc = {dc: 0 for dc in expected_dcs}
+            per_group = {key: 0 for key in expected_groups}
 
             for copy in counted:
-                dc = dc_of(copy)
+                key = group_of(copy)
 
-                if dc in per_dc:
-                    per_dc[dc] += 1
+                if key in per_group:
+                    per_group[key] += 1
 
-            missing = {dc for dc, cnt in per_dc.items() if cnt == 0}
+            missing = {key for key, cnt in per_group.items() if cnt == 0}
 
-            unbalanced = {} if copies_per_dc is None else \
-                {dc: cnt for dc, cnt in per_dc.items() if cnt != copies_per_dc}
+            unbalanced = {} if copies_per_group is None else \
+                {key: cnt for key, cnt in per_group.items() if cnt != copies_per_group}
 
             if missing or unbalanced:
                 copies_dump = ", ".join(
-                    f"{c.node_id}({'P' if c.primary else 'B'},{c.state},dc={dc_of(c)},{c.node_addresses})"
+                    f"{c.node_id}({'P' if c.primary else 'B'},{c.state},{label}={group_of(c)},"
+                    f"{c.node_addresses})"
                     for c in copies)
 
                 problems = []
 
                 if missing:
-                    problems.append(f"missing DCs={sorted(missing)}")
+                    problems.append(f"missing {label}s={sorted(missing)}")
 
                 if unbalanced:
-                    problems.append(f"copies per DC != {copies_per_dc}: {unbalanced}")
+                    problems.append(f"copies per {label} != {copies_per_group}: {unbalanced}")
 
                 violations.append(f"group={group.name}(id={group.group_id}), partition={part}, "
                                   f"{', '.join(problems)}, copies=[{copies_dump}]")
 
     assert not violations, \
-        "Partition distribution is not cross-DC:\n  " + "\n  ".join(violations) + "\n" + layout_hint
+        f"Partition distribution does not cover every {label}:\n  " + "\n  ".join(violations) + \
+        "\n" + layout_hint
diff --git a/modules/ducktests/tests/ignitetest/services/network_group/manager.py b/modules/ducktests/tests/ignitetest/services/network_group/manager.py
index 819212456bde7..493185d2d5a9d 100644
--- a/modules/ducktests/tests/ignitetest/services/network_group/manager.py
+++ b/modules/ducktests/tests/ignitetest/services/network_group/manager.py
@@ -44,7 +44,7 @@
 MAX_PARALLEL_SSH_SESSIONS = 16
 
 # Separates the qdisc, filter and iptables sections in the output of the batched
-# network probe issued by _log_network.
+# network probe issued by _probe_network.
 PROBE_SECTION_SEPARATOR = "=== ignitetest network probe section ==="
 
 # A rule spec: (src_group, dst_group, action, config).
@@ -128,23 +128,36 @@ def enable_network_partition(self, group_a: str, group_b: str):
         simulating a split-brain. The netem impairments deployed via tcset are
         left untouched underneath.
         """
-        self.logger.info(f"Enabling network partition between [{group_a}] <---> [{group_b}]")
+        self.enable_network_partitions((group_a, group_b))
 
-        chain = partition_chain_name(group_a, group_b)
+    def enable_network_partitions(self, *group_pairs: Tuple[str, str]):
+        """
+        Cuts several group pairs apart at once, e.g. to isolate one group from
+        every other one or to split a three group cluster three ways.
 
-        tasks = []
+        Every chain a node takes part in is installed by the same single SSH
+        round-trip, so a multi-way split takes effect near-atomically instead of
+        rolling out pair by pair - the intermediate topologies of a pair-by-pair
+        rollout are themselves valid segmentations the cluster would react to.
+        """
+        pairs_str = ", ".join(f"[{a}] <---> [{b}]" for a, b in group_pairs)
+
+        self.logger.info(f"Enabling network partition between {pairs_str}")
+
+        per_node = {}
 
-        for src_group, dst_group in self._bidirectional(group_a, group_b):
-            remote_ips = self._resolve_group_ips(dst_group)
+        for group_a, group_b in group_pairs:
+            chain = partition_chain_name(group_a, group_b)
 
-            cmd = to_partition_enable_cmd(chain, remote_ips)
+            for src_group, dst_group in self._bidirectional(group_a, group_b):
+                cmd = to_partition_enable_cmd(chain, self._resolve_group_ips(dst_group))
 
-            for node in self._iter_group_nodes(src_group):
-                tasks.append((node, cmd))
+                for node in self._iter_group_nodes(src_group):
+                    per_node.setdefault(id(node), (node, []))[1].append(cmd)
 
-        self._ssh_parallel(tasks, tag="PARTITION_ON")
+        self._ssh_parallel([(node, " && ".join(cmds)) for node, cmds in per_node.values()], tag="PARTITION_ON")
 
-        self._log_network(f"PARTITION {group_a} <-> {group_b}")
+        self._log_network(f"PARTITION {pairs_str}")
 
     def disable_network_partition(self, group_a: str, group_b: str):
         """
@@ -153,17 +166,31 @@ def disable_network_partition(self, group_a: str, group_b: str):
         node. The originally deployed tcset impairments were never modified,
         so they are back in effect immediately without any re-application.
         """
-        self.logger.info(f"Disabling network partition between [{group_a}] <---> [{group_b}]")
+        self.disable_network_partitions((group_a, group_b))
+
+    def disable_network_partitions(self, *group_pairs: Tuple[str, str]):
+        """
+        Heals several partitions at once, one SSH round-trip per node - the
+        counterpart of :meth:`enable_network_partitions`.
+        """
+        pairs_str = ", ".join(f"[{a}] <---> [{b}]" for a, b in group_pairs)
+
+        self.logger.info(f"Disabling network partition between {pairs_str}")
+
+        per_node = {}
 
-        cmd = to_partition_disable_cmd(partition_chain_name(group_a, group_b))
+        for group_a, group_b in group_pairs:
+            cmd = to_partition_disable_cmd(partition_chain_name(group_a, group_b))
 
-        tasks = [(node, cmd)
-                 for group in (group_a, group_b)
-                 for node in self._iter_group_nodes(group)]
+            for group in (group_a, group_b):
+                for node in self._iter_group_nodes(group):
+                    per_node.setdefault(id(node), (node, []))[1].append(cmd)
 
-        self._ssh_parallel(tasks, tag="PARTITION_OFF")
+        # ';' rather than '&&': healing is best effort, and a chain that was never installed
+        # on this node must not stop the rest of its chains from being flushed.
+        self._ssh_parallel([(node, " ; ".join(cmds)) for node, cmds in per_node.values()], tag="PARTITION_OFF")
 
-        self._log_network(f"NET RESTORED {group_a} <-> {group_b}")
+        self._log_network(f"NET RESTORED {pairs_str}")
 
     def _resolve_group_ips(self, group: str) -> List[str]:
         return [socket.gethostbyname(node.account.externally_routable_ip)
diff --git a/modules/ducktests/tests/ignitetest/services/utils/cdc/cdc_helper.py b/modules/ducktests/tests/ignitetest/services/utils/cdc/cdc_helper.py
index 9965dbde21f30..0b8f83fb93a08 100644
--- a/modules/ducktests/tests/ignitetest/services/utils/cdc/cdc_helper.py
+++ b/modules/ducktests/tests/ignitetest/services/utils/cdc/cdc_helper.py
@@ -207,8 +207,6 @@ def last_ignite_cdc_event_time(ignite_cdc):
     :return: Timestamp of the last CDC event (unix time in seconds).
     """
     def last_event_time_on(node):
-        jmx_client = JmxClient(node)
-
         if isinstance(ignite_cdc, IgniteCdcUtility):
             main_java_class = ignite_cdc.APP_SERVICE_CLASS
         else:
@@ -219,12 +217,12 @@ def last_event_time_on(node):
         if len(pids) == 0:
             raise AssertionError("ignite_cdc java process is not found on node: " + node.account.hostname)
 
-        jmx_client.pid = pids[0]
+        jmx_client = JmxClient(node, pids[0])
 
         try:
-            mbean = jmx_client.find_mbean('.*name=cdc.*')
+            mbean = jmx_client.find_metric_registry('cdc')
 
-            return int(next(mbean.LastEventTime).strip())
+            return int(mbean.value("LastEventTime"))
         except (StopIteration, RemoteCommandError):
             ignite_cdc.logger.warn("Filed to read LastEventTime metric from ignite_cdc, node: " + node.account.hostname)
 
diff --git a/modules/ducktests/tests/ignitetest/services/utils/control_utility.py b/modules/ducktests/tests/ignitetest/services/utils/control_utility.py
index 8ace03db5ae8d..329e5321974e5 100644
--- a/modules/ducktests/tests/ignitetest/services/utils/control_utility.py
+++ b/modules/ducktests/tests/ignitetest/services/utils/control_utility.py
@@ -28,7 +28,6 @@
 from ignitetest.services.utils.auth import get_credentials, is_auth_enabled
 from ignitetest.services.utils.ignite_spec import envs_to_exports
 from ignitetest.services.utils.ssl.ssl_params import get_ssl_params, is_ssl_enabled, IGNITE_ADMIN_ALIAS
-from ignitetest.services.utils.jmx_utils import JmxClient
 from ignitetest.utils.version import V_2_11_0
 
 
@@ -303,17 +302,17 @@ def snapshot_create(self, snapshot_name: str, timeout_sec: int = 60):
 
         while datetime.now() < delta_time:
             for node in self._cluster.nodes:
-                mbean = JmxClient(node).find_mbean('.*name=snapshot.*', negative_pattern='group=views')
+                mbean = node.metric_registry_mbean('snapshot')
 
-                if snapshot_name != next(mbean.LastSnapshotName, ""):
+                if snapshot_name != mbean.value("LastSnapshotName", ""):
                     continue
 
-                start_time = int(next(mbean.LastSnapshotStartTime))
-                end_time = int(next(mbean.LastSnapshotEndTime))
-                err_msg = next(mbean.LastSnapshotErrorMessage)
+                start_time = int(mbean.value("LastSnapshotStartTime"))
+                end_time = int(mbean.value("LastSnapshotEndTime"))
+                err_msg = mbean.value("LastSnapshotErrorMessage")
 
                 if (start_time < end_time) and (err_msg == ''):
-                    assert snapshot_name == next(mbean.LastSnapshotName)
+                    assert snapshot_name == mbean.value("LastSnapshotName")
                     return
 
         raise TimeoutError(f'Failed to wait for the snapshot operation to complete: '
diff --git a/modules/ducktests/tests/ignitetest/services/utils/ignite_aware.py b/modules/ducktests/tests/ignitetest/services/utils/ignite_aware.py
index 8bd758f339d04..0f7079d75b157 100644
--- a/modules/ducktests/tests/ignitetest/services/utils/ignite_aware.py
+++ b/modules/ducktests/tests/ignitetest/services/utils/ignite_aware.py
@@ -37,7 +37,7 @@
 from ignitetest.services.utils.background_thread import BackgroundThreadService
 from ignitetest.services.utils.concurrent import CountDownLatch, AtomicValue
 from ignitetest.services.utils.ignite_spec import resolve_spec, SHARED_PREPARED_FILE
-from ignitetest.services.utils.jmx_utils import ignite_jmx_mixin, JmxClient
+from ignitetest.services.utils.jmx_utils import ignite_jmx_mixin
 from ignitetest.services.utils.jvm_utils import JvmProcessMixin, JvmVersionMixin
 from ignitetest.services.utils.log_utils import monitor_log
 from ignitetest.services.utils.path import IgnitePathAware
@@ -600,10 +600,10 @@ def await_rebalance(self, timeout_sec=600):
         node = random.choice(self.alive_nodes)
 
         rebalanced = False
-        mbean = JmxClient(node).find_mbean('.*name=cluster')
+        mbean = node.metric_registry_mbean('cluster')
 
         while datetime.now() < delta_time and not rebalanced:
-            rebalanced = next(mbean.Rebalanced) == 'true'
+            rebalanced = mbean.bool_value("Rebalanced")
 
         if rebalanced:
             return
diff --git a/modules/ducktests/tests/ignitetest/services/utils/jmx_utils.py b/modules/ducktests/tests/ignitetest/services/utils/jmx_utils.py
index 48cfe00003f53..73ca0abdc46ec 100644
--- a/modules/ducktests/tests/ignitetest/services/utils/jmx_utils.py
+++ b/modules/ducktests/tests/ignitetest/services/utils/jmx_utils.py
@@ -20,21 +20,87 @@
 import os
 import re
 
-from ignitetest.services.utils.decorators import memoize
 from ignitetest.services.utils.jvm_utils import java_version, java_major_version
 
+_NO_DEFAULT = object()
+
+# Not allowed in a metric registry name: the single quote would break the quoted grep argument, and the
+# $'...' quoting of the jmxterm commands turns the exporter's \" and \? escapes, and a lone backslash, into
+# something else.
+_UNSUPPORTED_REGISTRY_CHARS = frozenset("'\"\\?")
+
+
+def _object_name_value(value):
+    """
+    Spells a property value the way IgniteUtils.escapeObjectNameValue() does: as is when it is
+    purely alphanumeric or underscore, otherwise quoted, with backslash, quote, '?' and '*' escaped.
+    """
+    if re.fullmatch(r'[A-Za-z0-9_]+', value):
+        return value
+
+    return '"' + re.sub(r'([\\"?*])', r'\\\1', value) + '"'
+
+
+def _ere_escape(value):
+    """
+    Escapes a literal for a POSIX extended regular expression, which is what 'grep -E' reads -
+    re.escape() escapes for the Python dialect instead.
+    """
+    return re.sub(r'([\[\\.^$*+?(){|])', r'\\\1', value)
+
+
+def metric_registry_pattern(registry):
+    """
+    Builds the MBean name pattern of a single metric registry.
+
+    The JMX metric exporter splits a registry name at its FIRST dot and makes the head the
+    MBean group and the tail its name: ``cache.myCache`` becomes ``group=cache`` plus
+    ``name=myCache``, ``io.dataregion.default`` becomes ``group=io`` plus
+    ``name="dataregion.default"``. A registry without a dot, ``snapshot`` say, gets no group at all.
+
+    Both values are spelled as the exporter spells them - quoted and escaped unless purely
+    alphanumeric - and matched literally. The properties of an MBean name are ordered
+    alphabetically, so the group always comes before the name but never right next to it. And
+    ``name`` sorts last of them all, so the pattern ends at the end of the line: without that
+    anchor ``myCache`` would also match the registry of ``myCacheV2``.
+
+    A registry without a group needs the negative pattern as well - an ERE has no lookahead, and
+    the system view of the same name (``group=views,...,name=snapshot``) matches otherwise.
+
+    The MBean name of a registry carries no pid, but it does carry the class loader hash, so it
+    may differ between two runs of a node.
+
+    :param registry: Metric registry name. It must not contain a single or double quote, a backslash
+           or a question mark - see :data:`_UNSUPPORTED_REGISTRY_CHARS`.
+    :return: Tuple of the pattern and the negative pattern to pass to :meth:`JmxClient.find_mbean`.
+    """
+    if _UNSUPPORTED_REGISTRY_CHARS.intersection(registry):
+        raise ValueError(f"A metric registry name with a quote, a backslash or '?' is not supported: {registry}")
+
+    group, dot, name = registry.partition('.')
+
+    if not dot:
+        return rf'.*name={_ere_escape(_object_name_value(registry))}\s*$', 'group='
+
+    return rf'.*group={_ere_escape(_object_name_value(group))},' \
+           rf'.*name={_ere_escape(_object_name_value(name))}\s*$', None
+
 
 def ignite_jmx_mixin(node, service):
     """
-    Dynamically mixin JMX attributes to Ignite service node.
+    Dynamically mixin JMX attributes to Ignite service node. Called on every start of the node,
+    which is what hands a restarted node a JMX client of its new JVM.
     :param node: Ignite service node.
     :param service: Ignite service.
     """
     setattr(node, 'pids', service.pids(node, service.main_java_class))
     setattr(node, 'install_root', service.install_root)
-    base_cls = node.__class__
-    base_cls_name = node.__class__.__name__
-    node.__class__ = type(base_cls_name, (base_cls, IgniteJmxMixin), {})
+    setattr(node, '_jmx_client', None)
+
+    if not isinstance(node, IgniteJmxMixin):
+        base_cls = node.__class__
+        base_cls_name = node.__class__.__name__
+        node.__class__ = type(base_cls_name, (base_cls, IgniteJmxMixin), {})
 
 
 class JmxMBean:
@@ -53,6 +119,30 @@ def __getattr__(self, attr):
         """
         return self.client.mbean_attribute(self.name, attr)
 
+    def value(self, attr, default=_NO_DEFAULT):
+        """
+        Reads a single valued attribute - which is what a metric read almost always wants,
+        as opposed to the raw line iterator the attribute access itself returns.
+
+        :param attr: Attribute name.
+        :param default: Value to return if the attribute reads nothing; StopIteration is raised if not passed.
+        :return: Attribute value, whitespace stripped.
+        """
+        try:
+            return next(self.client.mbean_attribute(self.name, attr)).strip()
+        except StopIteration:
+            if default is _NO_DEFAULT:
+                raise
+
+            return default
+
+    def bool_value(self, attr):
+        """
+        :param attr: Attribute name.
+        :return: Attribute value as a boolean; anything but "true" is False.
+        """
+        return self.value(attr).lower() == "true"
+
     def run(self, operation, params):
         """"
         Runs through JMX client MBean operation.
@@ -66,11 +156,16 @@ def run(self, operation, params):
 class JmxClient:
     """JMX client, invokes jmxterm on node locally.
     """
-    def __init__(self, node):
+    def __init__(self, node, pid=None):
+        """
+        :param node: Node to run jmxterm on.
+        :param pid: JVM to connect to; the first of the node's pids if not passed.
+        """
         self.node = node
         self.install_root = node.install_root
-        self.pid = node.pids[0]
+        self.pid = pid if pid is not None else node.pids[0]
         self.java_major = java_major_version(java_version(self.node))
+        self._mbeans = {}
 
     @property
     def jmx_util_cmd(self):
@@ -81,24 +176,34 @@ def jmx_util_cmd(self):
 
         return os.path.join(f"java {extra_flag} -jar {self.install_root}/jmxterm.jar -v silent -n")
 
-    @memoize
     def find_mbean(self, pattern, negative_pattern=None, domain='org.apache'):
         """
-        Find mbean by specified pattern and domain on node.
+        Find mbean by specified pattern and domain on node. The lookup is cached by this client, which
+        is safe as long as the client does not outlive the JVM it was built for.
         :param pattern: MBean name pattern.
         :param negative_pattern: if passed used to filter out some MBeans
         :param domain: Domain of MBean
         :return: JmxMBean instance
         """
-        cmd = "echo $'open %s\\n beans -d %s \\n close' | %s | grep -E -o '%s'" \
-              % (self.pid, domain, self.jmx_util_cmd, pattern)
+        key = (pattern, negative_pattern, domain)
+
+        if key not in self._mbeans:
+            cmd = "echo $'open %s\\n beans -d %s \\n close' | %s | grep -E -o '%s'" \
+                  % (self.pid, domain, self.jmx_util_cmd, pattern)
 
-        if negative_pattern:
-            cmd += " | grep -E -v '%s'" % negative_pattern
+            if negative_pattern:
+                cmd += " | grep -E -v '%s'" % negative_pattern
 
-        name = next(self.__run_cmd(cmd)).strip()
+            self._mbeans[key] = JmxMBean(self, next(self.__run_cmd(cmd)).strip())
 
-        return JmxMBean(self, name)
+        return self._mbeans[key]
+
+    def find_metric_registry(self, registry):
+        """
+        :param registry: Metric registry name, e.g. ``cache.myCache`` - see :func:`metric_registry_pattern`.
+        :return: JmxMBean of the metric registry.
+        """
+        return self.find_mbean(*metric_registry_pattern(registry))
 
     def mbean_attribute(self, mbean, attr):
         """
@@ -188,31 +293,34 @@ def __find__(self, pattern):
 class IgniteJmxMixin:
     """
     Mixin to IgniteService node, exposing useful properties, obtained from JMX.
+
+    Everything here is read through the JMX client of the node's CURRENT JVM: a client holds the
+    pid and caches the MBean names of the JVM it was built for, so ignite_jmx_mixin() drops it on
+    every start of the node and the next access builds a new one.
     """
-    @memoize
     def jmx_client(self):
         """
-        :return: JmxClient instance.
+        :return: JmxClient instance of the running node.
         """
-        # noinspection PyTypeChecker
-        return JmxClient(self)
+        if self._jmx_client is None:
+            # noinspection PyTypeChecker
+            self._jmx_client = JmxClient(self)
+
+        return self._jmx_client
 
-    @memoize
     def node_id(self):
         """
         :return: Local node id.
         """
-        return next(self.kernal_mbean().LocalNodeId).strip()
+        return self.kernal_mbean().value("LocalNodeId")
 
     def discovery_info(self):
         """
         :return: DiscoveryInfo instance.
         """
         disco_mbean = self.disco_mbean()
-        crd = next(disco_mbean.Coordinator).strip()
-        local = next(disco_mbean.LocalNodeFormatted).strip()
 
-        return DiscoveryInfo(crd, local)
+        return DiscoveryInfo(disco_mbean.value("Coordinator"), disco_mbean.value("LocalNodeFormatted"))
 
     def kernal_mbean(self):
         """
@@ -220,12 +328,18 @@ def kernal_mbean(self):
         """
         return self.jmx_client().find_mbean('.*group=Kernal.*name=IgniteKernal')
 
-    @memoize
+    def metric_registry_mbean(self, registry):
+        """
+        :param registry: Metric registry name, e.g. ``cache.myCache`` or ``cacheGroups.myGroup``.
+        :return: MBean of the metric registry.
+        """
+        return self.jmx_client().find_metric_registry(registry)
+
     def disco_mbean(self):
         """
         :return: DiscoverySpi MBean.
         """
-        disco_spi = next(self.kernal_mbean().DiscoverySpiFormatted).strip()
+        disco_spi = self.kernal_mbean().value("DiscoverySpiFormatted")
 
         if 'ZookeeperDiscoverySpi' in disco_spi:
             return self.jmx_client().find_mbean('.*group=SPIs.*name=ZookeeperDiscoverySpi')
diff --git a/modules/ducktests/tests/ignitetest/tests/dump_test.py b/modules/ducktests/tests/ignitetest/tests/dump_test.py
index db58d5064441a..0a8bfcd4d6964 100644
--- a/modules/ducktests/tests/ignitetest/tests/dump_test.py
+++ b/modules/ducktests/tests/ignitetest/tests/dump_test.py
@@ -115,8 +115,8 @@ def get_data_region_size(ignite):
         data_region_size = {}
 
         for node in ignite.nodes:
-            mbean = node.jmx_client().find_mbean('.*group=io.*name="dataregion.default"')
-            data_region_size[node.consistent_id] = int(next(mbean.TotalUsedSize))
+            mbean = node.metric_registry_mbean('io.dataregion.default')
+            data_region_size[node.consistent_id] = int(mbean.value("TotalUsedSize"))
 
         return {
             "data_region_size": data_region_size
diff --git a/modules/ducktests/tests/ignitetest/tests/mdc/majority_partition_test.py b/modules/ducktests/tests/ignitetest/tests/mdc/majority_partition_test.py
new file mode 100644
index 0000000000000..7c3beb8edbde9
--- /dev/null
+++ b/modules/ducktests/tests/ignitetest/tests/mdc/majority_partition_test.py
@@ -0,0 +1,255 @@
+# 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.
+
+"""
+Three data center MDC resilience: majority based topology validation.
+
+With an ODD number of data centers MdcTopologyValidator switches from "the main DC must
+be visible" to "a majority of the DC set must be visible". That is a different guarantee
+and not a bigger version of the two DC one:
+
+  - no data center is privileged - the cluster keeps accepting writes through the loss of
+    ANY single DC, including the first one, which two DC mode cannot do;
+  - a segment holding a minority of DCs goes read-only even though it is perfectly
+    healthy internally - so a three way split leaves NO writable segment at all.
+
+Every partition owns exactly one copy per DC (backups = 2), so every segment - down to a
+single isolated DC - can still serve every read. That is what separates the read
+assertions from the write ones throughout these tests.
+"""
+from time import sleep
+
+from ducktape.mark import parametrize
+
+from ignitetest.services.mdc.mdc_cluster import MdcCluster, cross_dc_network, all_pairs, isolation_pairs, \
+    DCS_3, DC_1, DC_2, DC_3
+from ignitetest.utils import cluster, ignite_versions
+from ignitetest.utils.ignite_test import IgniteTest
+from ignitetest.utils.version import DEV_BRANCH
+
+CACHE_NAME = "mdc-majority"
+
+# Keys generated from each DC; the whole data set is [0, KEYS_PER_DC * len(DCS_3)).
+KEYS_PER_DC = 100
+
+# Time for discovery to detect the partition and for every segment to complete PME.
+SPLIT_SETTLE_SECS = 20
+
+# Probe key ranges of the put admissibility checks. Well above the data set, and one range
+# per check, so that the writes an admissible check performs never collide.
+PROBE_BASE = 1_000_000
+PROBE_STRIDE = 1_000_000
+
+
+def _probe_offset(idx: int) -> int:
+    return PROBE_BASE + idx * PROBE_STRIDE
+
+
+class MdcMajorityPartitionTest(IgniteTest):
+    """
+    Tests for majority based MDC topology validation across three data centers.
+    """
+    @cluster(num_nodes=9)
+    @ignite_versions(str(DEV_BRANCH))
+    @parametrize(cross_dc_latency_ms=60, isolated_dc=DC_3)
+    @parametrize(cross_dc_latency_ms=100, isolated_dc=DC_1)
+    def test_minority_dc_isolation(self, ignite_version, cross_dc_latency_ms, isolated_dc):
+        """
+        One DC is cut off from the two others. The majority segment must stay writable and
+        the isolated one must go read-only while still serving every read.
+
+        Parametrized over the isolated DC to assert that no DC is privileged: isolating DC1
+        - the main DC of the equivalent two DC cluster, and the DC the others discovered
+        through - is just as survivable as isolating DC3.
+        """
+        mdc = MdcCluster(self, ignite_version, dcs=DCS_3, srv_per_dc=2, runners_per_dc=1, jmx_metrics=True,
+                         network_timeout=20_000, tcp_connect_timeout=10_000)
+
+        majority = tuple(dc for dc in mdc.dcs if dc != isolated_dc)
+
+        cut = isolation_pairs(isolated_dc, mdc.dcs)
+
+        with cross_dc_network(self.logger, mdc, delay_ms=cross_dc_latency_ms) as net:
+            mdc.start_servers()
+
+            total_keys = self._generate_data(mdc)
+
+            mdc.verify_cache_distribution(CACHE_NAME, copies_per_dc=1)
+
+            # The cluster's own verdict on the guarantees, next to the test's: the affinity
+            # metric reports the cache CONFIGURATION, the distribution one the assignment
+            # verify_cache_distribution() has just walked partition by partition.
+            mdc.verify_cache_mdc_metrics(CACHE_NAME, affinity_safe=True, distribution_safe=True)
+
+            # Both cuts land in the same SSH round-trip: rolling them out one after the
+            # other would briefly present the cluster with a two segment topology it would
+            # legitimately react to, which is not the scenario under test.
+            net.enable_network_partitions(*cut)
+
+            sleep(SPLIT_SETTLE_SECS)
+
+            mdc.verify_segments(majority, isolated_dc)
+
+            # One copy of every partition lives in every DC, so both segments read everything.
+            for dc in mdc.dcs:
+                mdc.check_data(dc, CACHE_NAME, 0, total_keys)
+
+            # Two DCs out of three are a majority - the segment keeps writing...
+            for idx, dc in enumerate(majority):
+                mdc.check_put_admissibility(dc, CACHE_NAME, True, key_offset=_probe_offset(idx))
+
+            # ...while the isolated DC sees one DC out of three and is rejected by the validator.
+            mdc.check_put_admissibility(isolated_dc, CACHE_NAME, False, key_offset=_probe_offset(2))
+
+            # The cache is configured for three DCs no matter what the segment can see, so the
+            # affinity metric holds. The distribution one is about what the segment CAN place
+            # a copy on and is only reported here - see the note on the constants.
+            mdc.verify_cache_mdc_metrics(CACHE_NAME, affinity_safe=True, dc=isolated_dc)
+
+            net.disable_network_partitions(*cut)
+
+            # The isolated DC may be the one the others discovered through, in which case the
+            # shared ip finder holds only its own addresses and a restart would seed it off
+            # itself. Point discovery at every DC so it rejoins the majority ring either way.
+            mdc.sync_service_discovery()
+
+            # Split-brain does not self-heal: the read-only segment rejoins via restart.
+            mdc.restart(isolated_dc)
+
+            self._verify_recovered(mdc, total_keys, probe_idx=3)
+
+            # Whole again, so the assignment is back to one copy per DC and the cluster says so.
+            mdc.verify_cache_mdc_metrics(CACHE_NAME, affinity_safe=True, distribution_safe=True)
+
+            mdc.stop_servers()
+
+    @cluster(num_nodes=9)
+    @ignite_versions(str(DEV_BRANCH))
+    @parametrize(cross_dc_latency_ms=100)
+    def test_three_way_split_blocks_all_writes(self, ignite_version, cross_dc_latency_ms):
+        """
+        Every cross-DC link drops at once, leaving three single-DC segments. No segment
+        holds a majority, so - unlike the two DC case, where one half always survives as
+        writable - the whole cluster goes read-only until the links come back.
+        """
+        mdc = MdcCluster(self, ignite_version, dcs=DCS_3, srv_per_dc=2, runners_per_dc=1,
+                         network_timeout=20_000, tcp_connect_timeout=10_000)
+
+        mesh = all_pairs(mdc.dcs)
+
+        with cross_dc_network(self.logger, mdc, delay_ms=cross_dc_latency_ms) as net:
+            mdc.start_servers()
+
+            total_keys = self._generate_data(mdc)
+
+            mdc.verify_cache_distribution(CACHE_NAME, copies_per_dc=1)
+
+            net.enable_network_partitions(*mesh)
+
+            sleep(SPLIT_SETTLE_SECS)
+
+            # Every DC ended up on its own, each a healthy single-DC segment.
+            mdc.verify_split_brain()
+
+            for idx, dc in enumerate(mdc.dcs):
+                mdc.check_data(dc, CACHE_NAME, 0, total_keys)
+
+                mdc.check_put_admissibility(dc, CACHE_NAME, False, key_offset=_probe_offset(idx))
+
+            net.disable_network_partitions(*mesh)
+
+            # The first DC keeps the ring, every other segment rejoins it via restart.
+            for dc in mdc.dcs[1:]:
+                mdc.restart(dc)
+
+            self._verify_recovered(mdc, total_keys, probe_idx=3)
+
+            mdc.stop_servers()
+
+    @cluster(num_nodes=9)
+    @ignite_versions(str(DEV_BRANCH))
+    def test_writes_survive_single_dc_loss(self, ignite_version):
+        """
+        Data centers go down one by one, no network impairments involved. Losing the first
+        one leaves a majority and writes continue; losing the second one drops the survivor
+        into a minority and it goes read-only; both returning restores writes everywhere.
+        """
+        mdc = MdcCluster(self, ignite_version, dcs=DCS_3, srv_per_dc=2, runners_per_dc=1)
+
+        with cross_dc_network(self.logger, mdc):
+            mdc.start_servers()
+
+            total_keys = self._generate_data(mdc)
+
+            mdc.verify_cache_distribution(CACHE_NAME, copies_per_dc=1)
+
+            mdc.stop_dcs(DC_3)
+
+            # Two DCs out of three remain visible - still a majority, still writable.
+            mdc.verify_segment_healthy((DC_1, DC_2))
+
+            mdc.check_data(DC_1, CACHE_NAME, 0, total_keys)
+
+            mdc.check_put_admissibility(DC_1, CACHE_NAME, True, key_offset=_probe_offset(0))
+            mdc.check_put_admissibility(DC_2, CACHE_NAME, True, key_offset=_probe_offset(1))
+
+            mdc.stop_dcs(DC_2)
+
+            # One DC out of three is a minority: everything is still readable...
+            mdc.check_data(DC_1, CACHE_NAME, 0, total_keys)
+
+            # ...but the validator rejects every write.
+            mdc.check_put_admissibility(DC_1, CACHE_NAME, False, key_offset=_probe_offset(2))
+
+            # DC1 never left, so the shared ip finder still points at a live ring. Both DCs
+            # are started before the first wait, so their joins are not serialized behind
+            # each other's rebalance.
+            mdc.start_dcs(DC_2, DC_3)
+
+            self._verify_recovered(mdc, total_keys, probe_idx=3)
+
+            mdc.stop_servers()
+
+    @staticmethod
+    def _generate_data(mdc: MdcCluster) -> int:
+        """
+        Populates the cache from every DC in turn, each with its own key range.
+
+        :return: The size of the whole data set, i.e. the exclusive upper key bound.
+        """
+        for idx, dc in enumerate(mdc.dcs):
+            mdc.generate_data(dc, CACHE_NAME, idx * KEYS_PER_DC, (idx + 1) * KEYS_PER_DC)
+
+        return len(mdc.dcs) * KEYS_PER_DC
+
+    @staticmethod
+    def _verify_recovered(mdc: MdcCluster, total_keys: int, probe_idx: int):
+        """
+        Verifies that the cluster is whole again: one ACTIVE cluster spanning every DC,
+        writes accepted everywhere, the whole data set intact and evenly spread, and no
+        segment left a suspicious trace in its logs.
+        """
+        mdc.verify_whole_cluster_healthy()
+
+        for idx, dc in enumerate(mdc.dcs):
+            mdc.check_put_admissibility(dc, CACHE_NAME, True, key_offset=_probe_offset(probe_idx + idx))
+
+        mdc.check_data(mdc.dcs[0], CACHE_NAME, 0, total_keys)
+
+        mdc.verify_cache_distribution(CACHE_NAME, copies_per_dc=1)
+
+        mdc.control().idle_verify(CACHE_NAME)
+
+        mdc.verify_servers_log_clean()
diff --git a/modules/ducktests/tests/ignitetest/tests/mdc/transactional_partition_test.py b/modules/ducktests/tests/ignitetest/tests/mdc/transactional_partition_test.py
index 4b1d2e3555cfe..6af2969f09ad0 100644
--- a/modules/ducktests/tests/ignitetest/tests/mdc/transactional_partition_test.py
+++ b/modules/ducktests/tests/ignitetest/tests/mdc/transactional_partition_test.py
@@ -88,10 +88,11 @@ def test_transactional_load_cut_on_partition(self, ignite_version, cross_dc_late
                 "keyTo": LOAD_KEY_TO_DC_2,
                 "stopOnError": True,
                 "resultPrefix": f"txLoad{DC_2}",
+                # The MDC topology parameters (validator mode, DC count) are injected by
+                # the fixture, so the cache this load creates matches the cluster's DC set.
                 "createCache": True,
                 "backups": BACKUPS,
                 "atomicity": "TRANSACTIONAL",
-                "mainDc": DC_1,
                 "txTimeout": 5_000
             })
 
diff --git a/modules/ducktests/tests/ignitetest/tests/rebalance/util.py b/modules/ducktests/tests/ignitetest/tests/rebalance/util.py
index 9a5351dd2388e..282a8f4ce786b 100644
--- a/modules/ducktests/tests/ignitetest/tests/rebalance/util.py
+++ b/modules/ducktests/tests/ignitetest/tests/rebalance/util.py
@@ -190,12 +190,12 @@ def get_rebalance_metrics(node, cache_group):
     :param cache_group: Cache group.
     :return: RebalanceMetrics instance.
     """
-    mbean = node.jmx_client().find_mbean('.*group=cacheGroups.*name="%s"' % cache_group)
-    start_time = int(next(mbean.RebalancingStartTime))
-    end_time = int(next(mbean.RebalancingEndTime))
+    mbean = node.metric_registry_mbean(f'cacheGroups.{cache_group}')
+    start_time = int(mbean.value("RebalancingStartTime"))
+    end_time = int(mbean.value("RebalancingEndTime"))
 
     return RebalanceMetrics(
-        received_bytes=int(next(mbean.RebalancingReceivedBytes)),
+        received_bytes=int(mbean.value("RebalancingReceivedBytes")),
         start_time=start_time,
         end_time=end_time,
         duration=(end_time - start_time) if start_time != -1 and end_time != -1 else 0,