anton-vinogradov commented on code in PR #12740:
URL: https://github.com/apache/ignite/pull/12740#discussion_r3595553087
##########
modules/ducktests/tests/ignitetest/services/utils/ignite_configuration/communication.py:
##########
@@ -53,7 +53,8 @@ def __init__(self,
connections_per_node: int = None,
use_paired_connections: bool = None,
message_queue_limit: int = None,
- unacknowledged_messages_buffer_size: int = None):
+ unacknowledged_messages_buffer_size: int = None,
+ connect_timeout: int = 5_000):
Review Comment:
Defaulting `connect_timeout` to `5_000` (instead of `None`) changes behavior
of **all** existing ducktests, not only MDC ones: `communication_macro.j2`
renders every non-empty field, so every test now gets an explicit `<property
name="connectTimeout" value="5000"/>`, and
`TcpCommunicationSpi#setConnectTimeout()` disables the
failure-detection-timeout logic for the communication SPI
(`failureDetectionTimeoutEnabled(false)` in
`TcpCommunicationConfigInitializer`). Today ducktests run comm timeouts derived
from `failure_detection_timeout=10000` (the `IgniteConfiguration` default
here); with this change they silently switch to the
`connTimeout`/`maxConnectTimeout`/`reconCnt` scheme.
Quick fix: keep `connect_timeout: int = None` here — `MdcCluster` already
passes it explicitly
(`TcpCommunicationSpi(connect_timeout=tcp_connect_timeout)`), so MDC tests keep
their value while the rest of the suite stays untouched.
##########
modules/ducktests/tests/ignitetest/services/utils/control_utility.py:
##########
@@ -181,6 +181,83 @@ def idle_verify_dump(self, node=None):
return re.search(r'/.*.txt', data).group(0)
+ def cache_distribution(self, node_id=None, cache_names=None,
user_attributes=None):
+ """
+ Prints partition distribution.
+
+ :param node_id: Node id to get distribution for, all nodes if None.
+ :param cache_names: Cache name, or list of cache names, all caches if
None.
+ :param user_attributes: Node attribute name or list of names to add to
the output
+ (e.g. "IGNITE_DATA_CENTER_ID").
+ :return: CacheDistribution.
+ """
+ if isinstance(cache_names, str):
+ cache_names = [cache_names]
+
+ if isinstance(user_attributes, str):
+ user_attributes = [user_attributes]
+
+ cmd = f"--cache distribution {node_id if node_id else 'null'}"
+
+ if cache_names:
+ cmd += f" {','.join(cache_names)}"
+
+ if user_attributes:
+ cmd += f" --user-attributes {','.join(user_attributes)}"
+
+ result = self.__run(cmd)
+
+ return self.__parse_cache_distribution(result, user_attributes)
+
+ @staticmethod
+ def __parse_cache_distribution(output, user_attributes=None):
+ group_pattern = re.compile(r"\[next group: id=(?P<group_id>-?\d+),
name=(?P<name>[^\]]+)\]")
+
+ # Trailing attribute values appear after nodeAddresses,
comma-separated,
+ # in the same order they were passed to --user-attributes.
+ row_pattern = re.compile(r"(?P<group_id>-?\d+),"
+ r"(?P<partition>\d+),"
+ r"(?P<node_id>[0-9a-fA-F]+),"
+ r"(?P<primary>[PB]),"
+ r"(?P<state>[A-Z_]+),"
+ r"(?P<update_counter>\d+),"
+ r"(?P<partition_size>\d+),"
+ r"\[(?P<node_addresses>[^\]]*)\]"
+ r"(?:,(?P<attr_values>.*))?$")
+
+ groups = {}
+ cur_group = None
+
+ for line in output.splitlines():
+ line = line.strip()
+
+ match = group_pattern.search(line)
+ if match:
+ cur_group =
CacheGroupDistribution(group_id=int(match.group("group_id")),
+ name=match.group("name"),
+ partitions={})
+ groups[cur_group.name] = cur_group
+ continue
+
+ match = row_pattern.match(line)
+ if match and cur_group is not None:
+ attrs = {}
+ if user_attributes and match.group("attr_values") is not None:
+ values = [v.strip() for v in
match.group("attr_values").split(",")]
+ attrs = dict(zip(user_attributes, values))
Review Comment:
The task stores user attributes in a `TreeMap`
(`CacheDistributionTask#run`), so values are printed in *alphabetical* key
order, not in `--user-attributes` argument order. `dict(zip(user_attributes,
values))` mis-assigns values whenever 2+ attributes are requested in
non-alphabetical order (fine for the single-attribute MDC usage, but the helper
is generic).
Quick fix: `dict(zip(sorted(user_attributes), values))`, or parse the names
back from the header row, which prints them in output order.
##########
modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/utils/Utils.java:
##########
@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.ducktest.utils;
+
+import java.util.Locale;
+import java.util.function.Supplier;
+import com.fasterxml.jackson.databind.JsonNode;
+
+/** Small helpers shared by the ducktest applications. */
+public final class Utils {
+ /** */
+ private Utils() {
+ // No-op.
+ }
+
+ /**
+ * Runs the operation, records its latency into {@code stats} and returns
its result.
+ * If the operation throws, nothing is recorded and the exception
propagates.
+ */
+ public static <T> T timed(OpStats stats, Supplier<T> op) {
+ long startNs = System.nanoTime();
+
+ T val = op.get();
+
+ stats.record(System.nanoTime() - startNs);
+
+ return val;
+ }
+
+ /**
+ * Runs the void operation and records its latency into {@code stats}.
+ * If the operation throws, nothing is recorded and the exception
propagates.
+ */
+ public static void timed(OpStats stats, Runnable op) {
+ timed(stats, () -> {
+ op.run();
+
+ return null;
+ });
+ }
+
+ /** Formats a nanosecond latency as milliseconds with 3 decimal places. */
+ public static String fmtMs(double ns) {
+ return String.format(Locale.US, "%.3f", ns < 0 ? -1.0 : ns / 1e6);
+ }
+
+ /**
+ * Parses an optional enum-valued field. A missing, null or unrecognized
value falls back
+ * to {@code dfltVal}.
+ */
+ public static <E extends Enum<E>> E getEnum(JsonNode jNode, String
fieldName, E dfltVal) {
+ JsonNode field = jNode.path(fieldName);
+
+ if (field.isMissingNode() || field.isNull())
+ return dfltVal;
+
+ try {
+ return Enum.valueOf(dfltVal.getDeclaringClass(),
field.asText().toUpperCase(Locale.ROOT));
+ }
+ catch (IllegalArgumentException e) {
+ return dfltVal; // Fallback if string doesn't match any enum
constant.
Review Comment:
Silently falling back on an unrecognized value contradicts the rationale of
the required-field overload below ("a typo must fail the run, not pick a
default"): e.g. `"atomicity": "TRANSACTONAL"` would silently create an ATOMIC
cache and invalidate the scenario the test believes it is running, with no
signal anywhere. Suggest defaulting only on missing/null and letting
`Enum.valueOf` throw for a present-but-invalid value.
##########
modules/ducktests/tests/ignitetest/services/utils/control_utility.py:
##########
@@ -404,7 +483,19 @@ def __parse_cluster_state(output):
order=int(match.group("order")) if
match.group("order") else None)
baseline.append(node)
- return ClusterState(state=state, topology_version=topology,
baseline=baseline)
+ coordinator = None
+
+ match = coordinator_pattern.search(output)
+
+ if match:
+ order = int(match.group("order"))
+
+ coordinator = next((node for node in baseline if node.order ==
order), None)
+
+ if coordinator is None:
+ raise AssertionError(f"Coordinator with order={order} is not
found in baseline [baseline={baseline}]")
Review Comment:
This turns a working `cluster_state()` into a crash for legitimate
topologies. `--baseline` prints the coordinator as the min-order node among
**all** server nodes, including non-baseline ones (`AbstractBaselineCommand`),
while "Other nodes:" lines carry no `State=`, so they don't match
`baseline_pattern` and never make it into `baseline`. If the coordinator
happens to be a non-baseline server (joined, not yet in baseline), every
`cluster_state()`/`baseline()` call in any existing test now raises here.
Quick fix: leave `coordinator=None` in that case instead of raising —
`verify_split_brain()` already asserts coordinator presence on its own.
##########
.github/workflows/commit-check.yml:
##########
@@ -147,7 +147,6 @@ jobs:
fail-fast: false
matrix:
cfg:
- - { python: "3.8", toxenv: "py38" }
- { python: "3.9", toxenv: "py39" }
Review Comment:
`modules/ducktests/tests/tox.ini` still lists `py38` in `envlist` — with
`tcconfig==0.30.1` requiring Python >= 3.9, the py38 env can no longer install
the deps. Suggest dropping it from tox.ini in the same commit that drops it
here.
##########
modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/mdc/MdcContinuousLoadApplication.java:
##########
@@ -0,0 +1,301 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.ducktest.tests.mdc;
+
+import java.util.List;
+import java.util.Locale;
+import java.util.Objects;
+import javax.cache.CacheException;
+import com.fasterxml.jackson.databind.JsonNode;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.cache.query.SqlFieldsQuery;
+import org.apache.ignite.internal.ducktest.tests.dto.IndexedDataRecord;
+import org.apache.ignite.internal.ducktest.utils.OpStats;
+import org.apache.ignite.transactions.Transaction;
+import org.apache.ignite.transactions.TransactionConcurrency;
+import org.apache.ignite.transactions.TransactionIsolation;
+
+import static org.apache.ignite.internal.ducktest.utils.Utils.fmtMs;
+import static org.apache.ignite.internal.ducktest.utils.Utils.getEnum;
+import static org.apache.ignite.internal.ducktest.utils.Utils.timed;
+import static
org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC;
+import static
org.apache.ignite.transactions.TransactionIsolation.REPEATABLE_READ;
+
+/**
+ * Universal MDC load application. Runs a single-threaded synchronous load of
the
+ * requested {@link LoadMode} either for a fixed number of iterations (a
"burst") or until
+ * externally terminated (a "background" load spanning several test phases,
e.g. a
+ * network partition and its healing).
+ * <p>
+ * Parameters:
+ * <ul>
+ * <li>{@code mode} - one of {@code GET, PUT, TX_PUT, SQL_SELECT,
SQL_PUT};</li>
+ * <li>{@code cacheName} - cache to operate on;</li>
+ * <li>{@code createCache} - if {@code true}, (re)creates the cache from
the MDC
+ * configuration parameters (see {@link MdcCacheAwareApplication});
otherwise the
+ * cache must already exist;</li>
+ * <li>{@code keyFrom} / {@code keyTo} - key range. Read modes cycle
within the range;
+ * write modes advance sequentially from {@code keyFrom} on every
success, so on
+ * completion the keys {@code [keyFrom, keyFrom + opsCnt)} are
guaranteed written;</li>
+ * <li>{@code iterations} - number of operations; {@code 0} means "run
until terminated";</li>
+ * <li>{@code expectAdmissible} - write modes only. {@code true}: writes
must succeed
+ * (default); {@code false}: every write must be rejected by the
topology validator
+ * (read-only DC), any success fails the application;</li>
+ * <li>{@code tolerateErrors} - if {@code true}, operation failures are
counted instead of
+ * failing fast. Intended for background loads crossing a partition
boundary, where a
+ * short transient error window is possible;</li>
+ * <li>{@code stopOnError} - if {@code true}, the load stops gracefully on
the very first
+ * operation failure (rather than failing the application), records
the number of
+ * successful operations and a {@code StoppedOnError} flag, and
finishes. Takes precedence
+ * over {@code tolerateErrors}. Intended for a load that must be cut
off by the first
+ * exception a network partition triggers;</li>
+ * <li>{@code opPauseMs} - pause between operations, default 0;</li>
+ * <li>{@code resultPrefix} - prefix for recorded results, so that several
runs reusing one
+ * service produce uniquely named results;</li>
+ * <li>{@code txConcurrency} / {@code txIsolation} - transaction
parameters for {@code TX_PUT}.</li>
+ * </ul>
+ */
+public class MdcContinuousLoadApplication extends MdcCacheAwareApplication {
+ /** */
+ public static final TransactionConcurrency DFLT_TX_CONCURRENCY =
PESSIMISTIC;
+
+ /** */
+ public static final TransactionIsolation DFLT_TX_ISOLATION =
REPEATABLE_READ;
+
+ /** Cache for the cache API modes ({@code null} in SQL modes). */
+ private IgniteCache<Integer, IndexedDataRecord> cache;
+
+ /** Cache for the SQL modes ({@code null} in cache API modes). */
+ private IgniteCache<Integer, Integer> sqlCache;
+
+ /** Compiled DML statement for {@link LoadMode#SQL_PUT}. */
+ private String mergeSql;
+
+ /** Compiled query for {@link LoadMode#SQL_SELECT}. */
+ private String selectSql;
+
+ /** Latency of successful operations. */
+ private final OpStats stats = new OpStats();
+
+ /** */
+ private TransactionConcurrency txConcurrency;
+
+ /** */
+ private TransactionIsolation txIsolation;
+
+ /** {@inheritDoc} */
+ @Override public void run(JsonNode jNode) throws Exception {
+ LoadMode mode = getEnum(jNode, "mode", LoadMode.class);
+
+ String cacheName = jNode.path("cacheName").asText(DFLT_CACHE_NAME);
+ boolean createCache = jNode.path("createCache").asBoolean(false);
+
+ int keyFrom = jNode.path("keyFrom").asInt(0);
+ int keyTo = jNode.path("keyTo").asInt(Integer.MAX_VALUE);
+ long iterations = jNode.path("iterations").asLong(0);
+
+ boolean expectAdmissible =
jNode.path("expectAdmissible").asBoolean(true);
+ boolean tolerateErrors = jNode.path("tolerateErrors").asBoolean(false);
+ boolean stopOnError = jNode.path("stopOnError").asBoolean(false);
+
+ long opPauseMs = jNode.path("opPauseMs").asLong(0);
+
+ String pfx = jNode.path("resultPrefix").asText("");
+
+ txConcurrency = getEnum(jNode, "txConcurrency", DFLT_TX_CONCURRENCY);
+ txIsolation = getEnum(jNode, "txIsolation", DFLT_TX_ISOLATION);
+
+ markInitialized();
+ waitForActivation();
+
+ if (mode.isSql())
+ sqlCache = createCache ? mdcSqlCache(jNode) :
ignite.cache(cacheName);
+ else
+ cache = createCache ? mdcCache(jNode) : ignite.cache(cacheName);
+
+ mergeSql = String.format("MERGE INTO \"%s\".%s(_KEY, _VAL) VALUES(?,
?)", cacheName, SQL_TABLE);
+ selectSql = String.format("SELECT _VAL FROM \"%s\".%s WHERE _KEY = ?",
cacheName, SQL_TABLE);
+
+ log.info("MDC load started [dc=" + dcId() + ", mode=" + mode + ",
cache=" + cacheName +
+ ", keyFrom=" + keyFrom + ", keyTo=" + keyTo + ", iterations=" +
iterations +
+ ", expectAdmissible=" + expectAdmissible + ", tolerateErrors=" +
tolerateErrors + "]");
+
+ long opsCnt = 0;
+ long errCnt = 0;
+
+ boolean stoppedOnError = false;
+
+ long maxStallMs = 0;
+
+ long startTs = System.currentTimeMillis();
+ long lastOkTs = startTs;
+
+ int key = keyFrom;
+
+ while (!terminated() && (iterations == 0 || opsCnt + errCnt <
iterations)) {
+ boolean ok;
+
+ try {
+ ok = doOperation(mode, key);
+ }
+ catch (CacheException | IgniteException e) {
+ ok = false;
+
+ if (mode.isWrite() && !expectAdmissible)
+ log.info("Write rejected as expected [dc=" + dcId() + ",
key=" + key +
+ ", msg=" + e.getMessage() + "]");
+ else if (stopOnError) {
+ log.warn("Operation failed, cutting the load on first
error [dc=" + dcId() + ", mode=" + mode +
+ ", key=" + key + ", succeeded=" + opsCnt + ", msg=" +
e.getMessage() + "]", e);
+
+ errCnt++;
+ stoppedOnError = true;
+
+ break;
+ }
+ else if (!tolerateErrors)
+ throw new IllegalStateException("Operation failed [dc=" +
dcId() + ", mode=" + mode +
+ ", key=" + key + "]", e);
+ else
+ log.warn("Operation failed, tolerated [dc=" + dcId() + ",
mode=" + mode +
+ ", key=" + key + ", msg=" + e.getMessage() + "]");
+ }
+
+ if (ok) {
+ opsCnt++;
+
+ long now = System.currentTimeMillis();
+
+ maxStallMs = Math.max(maxStallMs, now - lastOkTs);
+ lastOkTs = now;
+ }
+ else
+ errCnt++;
+
+ // Writes advance on success only, so [keyFrom, keyFrom + opsCnt)
is guaranteed written.
+ // The inadmissible-probe mode advances always to probe distinct
keys. Reads cycle the range.
+ if (ok || (mode.isWrite() && !expectAdmissible)) {
+ key++;
+
+ if (key >= keyTo)
+ key = keyFrom;
+ }
+
+ if (opPauseMs > 0)
+ Thread.sleep(opPauseMs);
+ }
+
+ long durationMs = System.currentTimeMillis() - startTs;
+
+ if (mode.isWrite() && !expectAdmissible && opsCnt > 0) {
+ throw new IllegalStateException("Write load is admissible while
expected to be inadmissible [dc=" +
+ dcId() + ", mode=" + mode + ", succeeded=" + opsCnt + ",
rejected=" + errCnt + "]");
+ }
+
+ recordResult(pfx + "OpsCnt", String.valueOf(opsCnt));
+ recordResult(pfx + "ErrCnt", String.valueOf(errCnt));
+ recordResult(pfx + "StoppedOnError", String.valueOf(stoppedOnError));
+ recordResult(pfx + "DurationMs", String.valueOf(durationMs));
+
+ recordResult(pfx + "AvgOpMs", fmtMs(stats.avgNs()));
+ recordResult(pfx + "MinOpMs", fmtMs(stats.minNs()));
+ recordResult(pfx + "MaxOpMs", fmtMs(stats.maxNs()));
+
+ recordResult(pfx + "DerivedTps", String.format(Locale.US, "%.1f",
stats.tps()));
+
+ recordResult(pfx + "MaxStallMs", String.valueOf(maxStallMs));
+
+ log.info("MDC load finished [dc=" + dcId() + ", mode=" + mode + ",
ops=" + opsCnt +
+ ", errs=" + errCnt + ", stoppedOnError=" + stoppedOnError + ",
durationMs=" + durationMs +
+ ", avgOpMs=" + fmtMs(stats.avgNs()) + ", maxOpMs=" +
fmtMs(stats.maxNs()) +
+ ", maxStallMs=" + maxStallMs + "]");
+
+ markFinished();
+ }
+
+ /**
+ * Executes a single operation of the given mode against the given key.
Throws a
+ * {@link CacheException} or {@link IgniteException} on operation failure
(e.g. a write
+ * rejected by the topology validator).
+ *
+ * @return {@code true} if the operation succeeded and (for reads)
returned the expected value.
+ */
+ private boolean doOperation(LoadMode mode, int key) {
Review Comment:
A missed/corrupted read (GET / SQL_SELECT) returns `ok=false` and only bumps
`errCnt`: unlike an exception, it neither fails the application when
`tolerateErrors=false` nor cuts the load when `stopOnError=true`. So a
data-integrity failure — the worst outcome a read load can produce — is reduced
to a counter the python side may or may not assert on.
Suggest routing verification failures through the same policy as exceptions
(fail when `!tolerateErrors`, cut when `stopOnError`), or at least spelling the
distinction out in the class javadoc.
##########
modules/ducktests/tests/ignitetest/tests/mdc/thin_client_test.py:
##########
@@ -0,0 +1,149 @@
+# 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.
+
+"""
+Thin client data-center-aware routing.
+
+Every thin client is configured with the addresses of ALL server nodes of BOTH
DCs.
+A client pinned to a data center must route partition-aware reads to nodes of
its own DC:
+with a large cross-DC netem delay its average read latency stays small.
+
+The partition part verifies the thin client experience of a split-brain: a
client
+pinned to the main-DC half keeps writing; a client pinned to the read-only
half still
+reads cleanly (falling back to the reachable nodes from its address list) but
gets
+every write rejected; and writes resume after the heal and rejoin.
+
+The thin client services are registered into their DC's network group, so both
the
+netem delay and the iptables partition apply to their traffic as well.
+"""
+from time import sleep
+
+from ducktape.mark import parametrize
+
+from ignitetest.services.ignite_app import IgniteApplicationService
+from ignitetest.services.mdc.mdc_cluster import MdcCluster, dc_jvm_opts, DC_1,
DC_2, cross_dc_network, THIN_LOAD_APP
+from ignitetest.services.utils.ignite_configuration import
IgniteThinClientConfiguration
+from ignitetest.utils import cluster, ignite_versions
+from ignitetest.utils.ignite_test import IgniteTest
+from ignitetest.utils.version import DEV_BRANCH
+
+CACHE_NAME = "mdc-thin"
+
+KEYS = 100
+
+PINNED_GET_ITERS = 200
+PUT_ITERS = 30
+
+OFFSET_DURING = 1_000_000
+OFFSET_REJECTED_PROBES = 9_000_000
+OFFSET_AFTER = 2_000_000
+
+SPLIT_SETTLE_SECS = 15
+
+
+class MdcThinClientTest(IgniteTest):
+ """
+ Tests for thin client DC-aware routing and behavior through a partition.
+ """
+ @cluster(num_nodes=8)
+ @ignite_versions(str(DEV_BRANCH))
+ @parametrize(cross_dc_latency_ms=100)
+ def test_thin_client_dc_aware_routing_and_partition(self, ignite_version,
cross_dc_latency_ms):
+ """
+ DC-pinned thin client reads locally (latency far below the cross-DC
delay);
+ through a partition the pinned clients behave like their half-ring:
main half writes, read-only half reads
+ but rejects writes, and writes resume after the heal.
+ """
+ mdc = MdcCluster(self, ignite_version, srv_per_dc=2,
runners_per_dc={DC_1: 1},
+ client_connector=True)
+
+ # All thin clients get the full address list of both DCs: DC
preference must come
+ # from routing, not from the address list.
+ cli_dc1 = self._thin_client(mdc, dc_jvm_opts(DC_1))
+ cli_dc2 = self._thin_client(mdc, dc_jvm_opts(DC_2))
+
+ # Register the thin client hosts into the network groups, so netem and
the partition apply to them.
+ mdc.register(DC_1, cli_dc1)
+ mdc.register(DC_2, cli_dc2)
+
+ with cross_dc_network(self.logger, mdc, delay_ms=cross_dc_latency_ms)
as net:
+ mdc.start_servers()
+
+ mdc.generate_data(DC_1, CACHE_NAME, 0, KEYS, backups=1)
+
+ svc = mdc.run_service(cli_dc1, {"mode": "GET", "cacheName":
CACHE_NAME, "keyFrom": 0,
+ "keyTo": KEYS, "iterations":
PINNED_GET_ITERS,
+ "resultPrefix": "pinnedGet"})
+
+ avg_cli_dc_1 = mdc.result_float(svc, "pinnedGetAvgOpMs")
+
+ self.logger.info(f"Thin client routing latency
[delayMs={cross_dc_latency_ms}, getAvgOpMs={avg_cli_dc_1}]")
+
+ assert avg_cli_dc_1 < cross_dc_latency_ms, \
Review Comment:
Worth also asserting `pinnedGetErrCnt == 0`: missed/corrupted reads only
increment ErrCnt in `MdcThinClientLoadApplication`, and a broken read path (all
gets returning null) would still satisfy the latency assertion — likely even
faster than healthy local reads.
##########
modules/ducktests/tests/ignitetest/services/mdc/mdc_cluster.py:
##########
@@ -0,0 +1,531 @@
+# 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.
+
+"""
+MDC test fixture.
+
+ mdc = MdcCluster(self, ignite_version, srv_per_dc=3, 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_partition(DC_1, DC_2)
+ ...
+"""
+from typing import Dict, List, Optional, Union
+
+from ignitetest.services.ignite import IgniteService
+from ignitetest.services.ignite_app import IgniteApplicationService
+from ignitetest.services.network_group.configuration import NetworkGroupStore,
CrossNetworkGroupConfiguration
+from ignitetest.services.network_group.manager import NetworkGroupManager
+from ignitetest.services.utils.control_utility import ControlUtility
+from ignitetest.services.utils.ignite_configuration import
IgniteConfiguration, TcpCommunicationSpi
+from ignitetest.services.utils.ignite_configuration.discovery import
from_ignite_cluster, from_ignite_services
+from ignitetest.services.utils.ssl.client_connector_configuration import
ClientConnectorConfiguration
+from ignitetest.utils.version import IgniteVersion
+
+DC_1 = "DC1"
+DC_2 = "DC2"
+DCS = (DC_1, DC_2)
+
+IGNITE_STARTUP_TIMEOUT_SEC = 90
+
+DATA_CENTER_ATTR = "IGNITE_DATA_CENTER_ID"
+IGNITE_SQL_RETRY_TIMEOUT_ATTR = "IGNITE_SQL_RETRY_TIMEOUT"
+
+IGNITE_SQL_RETRY_TIMEOUT_MS = 1_000
+
+_APP_PKG = "org.apache.ignite.internal.ducktest.tests.mdc."
+
+GENERATOR_APP = _APP_PKG + "MdcDataGeneratorApplication"
+DATA_CHECKER_APP = _APP_PKG + "MdcDataCheckerApplication"
+LOAD_APP = _APP_PKG + "MdcContinuousLoadApplication"
+THIN_LOAD_APP = _APP_PKG + "MdcThinClientLoadApplication"
+
+# Suspicious server log patterns: none of them is expected in any MDC scenario,
+# partitioned or not. Matched against the node console capture.
+LRT_PATTERN = "long running transactions"
+PME_FREEZE_PATTERN = "Failed to wait for partition map exchange"
+LOST_PARTITIONS_PATTERN = "Detected lost partitions"
+
+
+def dc_jvm_opts(dc: str) -> List[str]:
+ """
+ :return: JVM options assigning a node to the given data center.
+ """
+ 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]:
+ """
+ Normalizes an int-or-dict per-DC count into a dict, e.g. 3 -> {DC1: 3,
DC2: 3}.
+ """
+ return dict(value) if isinstance(value, dict) else {dc: value for dc in
DCS}
+
+
+class MdcCluster:
+ """
+ Owns the per-DC Ignite services and reusable application services of an
MDC test,
+ plus the MDC-specific verification helpers.
+
+ :param test: The ducktape test instance.
+ :param ignite_version: Ignite version string.
+ :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.
+ """
+ def __init__(self, test, ignite_version: str, 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,
+ network_timeout: int = 5_000,
+ tcp_connect_timeout: int = 5_000):
+ self.test_context = test.test_context
+ self.logger = test.logger
+
+ cfg_kwargs = {
+ "version": IgniteVersion(ignite_version),
+ "network_timeout": network_timeout,
+ "communication_spi":
TcpCommunicationSpi(connect_timeout=tcp_connect_timeout)
+ }
+
+ if client_connector:
+ cfg_kwargs["client_connector_configuration"] =
ClientConnectorConfiguration()
+
+ self.ignite_config = IgniteConfiguration(**cfg_kwargs)
Review Comment:
The mechanism that makes two DCs form ONE cluster is completely invisible
here and worth an explicit comment (or an explicit shared
`discovery_spi=TcpDiscoverySpi()` in `cfg_kwargs`): both `IgniteService`s share
this single config object, hence one `TcpDiscoveryVmIpFinder`, and
`prepare_on_start()` memoizes the *first started* DC's addresses into it — DC2
then discovers through DC1's nodes, and `restart(DC_2)` re-joins the same way.
That is also exactly why `test_main_dc_loss_and_return` must call
`sync_service_discovery()` before restarting the seed DC. Making this
load-bearing detail explicit would save the next reader a debugging session.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
