IGNITE-27833 [ducktests] Add Multi-DC Integration Test for Cross-DC Network Partition Resilience

modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/dto/IndexedDataRecord.java

@@ -0,0 +1,96 @@
+/*
+ * 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.dto;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+
+/** */
+public class IndexedDataRecord {
+    /** The base index converted to a string. */
+    private final String strRepresentation;
+
+    /** The primitive index boxed as an Integer. */
+    private final Integer boxedIdx;
+
+    /** True if the base index is an even number, false otherwise. */
+    private final Boolean isEven;
+
+    /** The first character of the string representation. */
+    private final char leadingCharacter;
+
+    /** An array containing the index, its double, and its square. */
+    private final Integer[] seqArr;
+
+    /** A list view of the sequence array. */
+    private final List<Integer> seqList;
+
+    /**
+     * Constructs an immutable record by generating derived properties from the provided index.
+     *
+     * @param idx the base integer used to compute all internal fields
+     */
+    public IndexedDataRecord(int idx) {
+        this.strRepresentation = String.valueOf(idx);
+        this.boxedIdx = idx;
+        this.isEven = (idx % 2 == 0);
+        this.leadingCharacter = this.strRepresentation.charAt(0);
+        this.seqArr = new Integer[] {idx, idx * 2, idx * idx};
+        this.seqList = Arrays.asList(this.seqArr);
+    }
+
+    /** {@inheritDoc} */
+    @Override public boolean equals(Object o) {
+        if (this == o)
+            return true;
+
+        if (o == null || getClass() != o.getClass())
+            return false;
+
+        IndexedDataRecord that = (IndexedDataRecord)o;
+
+        return leadingCharacter == that.leadingCharacter
+            && Objects.equals(strRepresentation, that.strRepresentation)
+            && Objects.equals(boxedIdx, that.boxedIdx)
+            && Objects.equals(isEven, that.isEven)
+            && Arrays.equals(seqArr, that.seqArr)
+            && Objects.equals(seqList, that.seqList);
+    }
+
+    /** {@inheritDoc} */
+    @Override public int hashCode() {
+        int result = Objects.hash(strRepresentation, boxedIdx, isEven, leadingCharacter, seqList);
+
+        result = 31 * result + Arrays.hashCode(seqArr);
+
+        return result;
+    }
+
+    /** {@inheritDoc} */
+    @Override public String toString() {
+        return "IndexedDataRecord{" +
+            "stringRepresentation='" + strRepresentation + '\'' +
+            ", boxedIndex=" + boxedIdx +
+            ", isEven=" + isEven +
+            ", leadingCharacter=" + leadingCharacter +
+            ", sequenceArray=" + Arrays.toString(seqArr) +
+            ", sequenceList=" + seqList +
+            '}';
+    }
+}

modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/mdc/LoadMode.java

@@ -0,0 +1,46 @@
+/*
+ * 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;
+
+/** Operation modes shared by the MDC load applications. */
+public enum LoadMode {
+    /** Cache API reads with value verification. */
+    GET,
+
+    /** Cache API writes. */
+    PUT,
+
+    /** Transactional cache API writes (requires a TRANSACTIONAL cache). */
+    TX_PUT,
+
+    /** SQL reads with value verification (requires an SQL-enabled cache). */
+    SQL_SELECT,
+
+    /** SQL DML writes (requires an SQL-enabled cache). */
+    SQL_PUT;
+
+    /** @return Whether the mode writes to the cache. */
+    public boolean isWrite() {
+        return this == PUT || this == TX_PUT || this == SQL_PUT;
+    }
+
+    /** @return Whether the mode operates through SQL. */
+    public boolean isSql() {
+        return this == SQL_SELECT || this == SQL_PUT;
+    }
+}

modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/mdc/MdcCacheAwareApplication.java

@@ -0,0 +1,170 @@
+/*
+ * 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.Collections;
+import java.util.HashSet;
+import java.util.Set;
+import com.fasterxml.jackson.databind.JsonNode;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteSystemProperties;
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.cache.CacheWriteSynchronizationMode;
+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.configuration.CacheConfiguration;
+import org.apache.ignite.internal.ducktest.tests.dto.IndexedDataRecord;
+import org.apache.ignite.internal.ducktest.utils.IgniteAwareApplication;
+import org.apache.ignite.topology.MdcTopologyValidator;
+
+import static org.apache.ignite.IgniteSystemProperties.IGNITE_DATA_CENTER_ID;
+import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC;
+import static org.apache.ignite.cache.CacheMode.PARTITIONED;
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
+import static org.apache.ignite.internal.ducktest.utils.Utils.getEnum;
+
+/**
+ * Base class for MDC test applications.
+ * Encapsulates the cache configuration (with {@link MdcTopologyValidator} and
+ * {@link MdcAffinityBackupFilter}) so that the generator, the checkers and the load
+ * applications always operate on an identically configured cache.
+ * <p>
+ * Supported cache parameters (all optional unless stated otherwise):
+ * <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 mainDc} - main data center for the topology validator (2 DC mode);</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>
+ *     <li>{@code cacheMode} - {@link CacheMode}, default {@code PARTITIONED};</li>
+ *     <li>{@code atomicity} - {@link CacheAtomicityMode}, default {@code ATOMIC};</li>
+ *     <li>{@code writeSync} - {@link CacheWriteSynchronizationMode}, default {@code FULL_SYNC}.
+ *         Note: MDC-aware local reads require a mode other than {@code PRIMARY_SYNC};</li>
+ *     <li>{@code readFromBackup} - default {@code true}, required for DC-local reads;</li>
+ *     <li>{@code partitions} - affinity partitions number, default 512.</li>
+ * </ul>
+ */
+public abstract class MdcCacheAwareApplication extends IgniteAwareApplication {
+    /** Table name of the SQL-enabled MDC cache. The SQL schema is the quoted cache name. */
+    protected static final String SQL_TABLE = "LOAD";
+
+    /** */
+    protected static final String DFLT_CACHE_NAME = "default";
+
+    /** */
+    protected static final CacheMode DFLT_CACHE_MODE = PARTITIONED;
+
+    /** One backup: with the default 2 DCs, {@code (backups + 1) / dcsNum} = 1 copy per DC. */
+    protected static final int DFLT_BACKUPS = 1;
+
+    /** */
+    protected static final int DFLT_DCS_NUM = 2;
+
+    /** */
+    protected static final int DFLT_PARTITIONS = 512;
+
+    /** */
+    protected static final CacheAtomicityMode DFLT_ATOMICITY_MODE = ATOMIC;
+
+    /** */
+    protected static final CacheWriteSynchronizationMode DFLT_WRITE_SYNC = FULL_SYNC;
+
+    /**
+     * @param jNode Parameters.
+     * @return Cache configured with the MDC topology validator and backup filter.
+     */
+    protected IgniteCache<Integer, IndexedDataRecord> mdcCache(JsonNode jNode) {
+        return ignite.getOrCreateCache(this.<IndexedDataRecord>mdcCacheConfiguration(jNode));
+    }
+
+    /**
+     * Same MDC cache configuration with a {@link QueryEntity} on top, so the cache is
+     * queryable via SQL: {@code SELECT _VAL FROM "<cacheName>".LOAD WHERE _KEY = ?}.
+     *
+     * @param jNode Parameters.
+     * @return SQL-enabled cache configured with the MDC topology validator and backup filter.
+     */
+    protected IgniteCache<Integer, Integer> mdcSqlCache(JsonNode jNode) {
+        CacheConfiguration<Integer, Integer> cacheCfg = mdcCacheConfiguration(jNode);
+
+        cacheCfg.setQueryEntities(Collections.singletonList(
+            new QueryEntity(Integer.class, Integer.class).setTableName(SQL_TABLE)));
+
+        return ignite.getOrCreateCache(cacheCfg);
+    }
+
+    /**
+     * @param jNode Parameters.
+     * @return MDC cache configuration compiled from the application parameters.
+     */
+    protected <V> CacheConfiguration<Integer, V> mdcCacheConfiguration(JsonNode jNode) {
+        String cacheName = jNode.path("cacheName").asText(DFLT_CACHE_NAME);
+        int backups = jNode.path("backups").asInt(DFLT_BACKUPS);
+        int partitions = jNode.path("partitions").asInt(DFLT_PARTITIONS);
+
+        CacheAtomicityMode atomicity = getEnum(jNode, "atomicity", DFLT_ATOMICITY_MODE);
+        CacheWriteSynchronizationMode writeSync = getEnum(jNode, "writeSync", DFLT_WRITE_SYNC);
+        CacheMode cacheMode = getEnum(jNode, "cacheMode", DFLT_CACHE_MODE);
+
+        boolean readFromBackup = jNode.path("readFromBackup").asBoolean(true);
+
+        int dcsNum = jNode.path("dcsNum").asInt(DFLT_DCS_NUM);
+
+        MdcTopologyValidator topValidator = new MdcTopologyValidator();
+
+        if (jNode.hasNonNull("datacenters")) {
+            Set<String> dcs = new HashSet<>();
+
+            jNode.get("datacenters").forEach(dc -> dcs.add(dc.asText()));
+
+            topValidator.setDatacenters(dcs);
+        }
+        else
+            topValidator.setMainDatacenter(jNode.path("mainDc").asText());
+
+        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)));
+    }
+
+    /**
+     * @param cacheName Cache name.
+     * @return Existing cache. The cache must have been created by the generator beforehand.
+     */
+    protected IgniteCache<Integer, IndexedDataRecord> existingCache(String cacheName) {
+        return ignite.cache(cacheName);
+    }
+
+    /**
+     * @return Data center id this client belongs to (passed via {@link IgniteSystemProperties#IGNITE_DATA_CENTER_ID}).
+     */
+    protected static String dcId() {
+        return IgniteSystemProperties.getString(IGNITE_DATA_CENTER_ID);
+    }
+}