You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by av...@apache.org on 2022/01/28 11:50:51 UTC
[ignite] branch master updated: IGNITE-15330 Read Repair should support strategies (#9661)
This is an automated email from the ASF dual-hosted git repository.
av pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new 7bbf2c3 IGNITE-15330 Read Repair should support strategies (#9661)
7bbf2c3 is described below
commit 7bbf2c3f60672110867ed6ec99316ddcaa12384e
Author: Anton Vinogradov <av...@apache.org>
AuthorDate: Fri Jan 28 14:50:13 2022 +0300
IGNITE-15330 Read Repair should support strategies (#9661)
---
docs/_docs/tools/control-script.adoc | 8 +-
.../consistency/ConsistencyCommand.java | 5 +-
.../util/GridCommandHandlerClusterByClassTest.java | 2 +-
.../util/GridCommandHandlerConsistencyTest.java | 91 ++--
.../ignite/util/KillCommandsCommandShTest.java | 8 +-
.../main/java/org/apache/ignite/IgniteCache.java | 4 +-
.../apache/ignite/cache/ReadRepairStrategy.java | 89 ++++
.../events/CacheConsistencyViolationEvent.java | 38 +-
.../processors/cache/CacheObjectUtils.java | 6 +-
.../processors/cache/CacheOperationContext.java | 39 +-
.../cache/GatewayProtectedCacheProxy.java | 14 +-
.../processors/cache/GridCacheAdapter.java | 111 ++---
.../processors/cache/GridCacheContext.java | 2 +-
.../processors/cache/GridCacheProxyImpl.java | 8 +-
.../processors/cache/IgniteCacheProxyImpl.java | 3 +-
.../cache/distributed/dht/GridDhtCacheAdapter.java | 7 +-
.../distributed/dht/atomic/GridDhtAtomicCache.java | 48 +-
.../dht/colocated/GridDhtColocatedCache.java | 15 +-
.../distributed/near/GridNearAtomicCache.java | 3 +-
.../near/GridNearTransactionalCache.java | 5 +-
.../cache/distributed/near/GridNearTxLocal.java | 72 +--
.../consistency/GridCompoundReadRepairFuture.java | 106 +++++
.../GridNearReadRepairAbstractFuture.java | 137 +++++-
.../GridNearReadRepairCheckOnlyFuture.java | 137 ++----
.../near/consistency/GridNearReadRepairFuture.java | 252 ++++++++--
.../IgniteConsistencyViolationException.java | 25 +-
...teIrreparableConsistencyViolationException.java | 63 +++
.../cache/local/atomic/GridLocalAtomicCache.java | 7 +-
.../datastructures/GridCacheQueueAdapter.java | 2 +-
.../consistency/VisorConsistencyRepairTask.java | 35 +-
.../consistency/VisorConsistencyRepairTaskArg.java | 16 +-
.../consistency/AbstractFullSetReadRepairTest.java | 237 +++++++---
.../cache/consistency/AbstractReadRepairTest.java | 519 +++++++++++++++------
.../cache/consistency/AtomicReadRepairTest.java | 207 --------
.../AtomicReadRepairTest.java} | 13 +-
.../ExplicitTransactionalReadRepairTest.java | 105 ++---
.../ImplicitTransactionalReadRepairTest.java | 50 +-
...licatedExplicitTransactionalReadRepairTest.java | 2 +-
...licatedImplicitTransactionalReadRepairTest.java | 2 +-
...eBackupExplicitTransactionalReadRepairTest.java | 2 +-
...eBackupImplicitTransactionalReadRepairTest.java | 2 +-
.../PdsAtomicReadRepairTest.java} | 10 +-
.../PdsExplicitTransactionalReadRepairTest.java} | 10 +-
.../PdsImplicitTransactionalReadRepairTest.java} | 10 +-
.../junits/multijvm/IgniteCacheProcessProxy.java | 3 +-
.../IgniteCacheConsistencySelfTestSuite.java | 55 +++
.../IgniteCacheDataStructuresSelfTestSuite.java | 17 -
.../tests/control_utility/consistency_test.py | 2 +-
.../internal/processors/query/h2/dml/DmlUtils.java | 2 +-
49 files changed, 1678 insertions(+), 928 deletions(-)
diff --git a/docs/_docs/tools/control-script.adoc b/docs/_docs/tools/control-script.adoc
index a1cce22..2f42109 100644
--- a/docs/_docs/tools/control-script.adoc
+++ b/docs/_docs/tools/control-script.adoc
@@ -1025,6 +1025,7 @@ tab:Windows[]
----
control.bat --property get --name 'statistics.usage.state'
----
+--
== Cache Consistency
@@ -1037,12 +1038,12 @@ The command allows to perform cache consistency check and repair (when possible)
tab:Unix[]
[source,shell]
----
-control.sh --enable-experimental --consistency repair cache-name partition
+control.sh --enable-experimental --consistency repair cache-name partition strategy
----
tab:Window[]
[source,shell]
----
-control.bat --enable-experimental --consistency repair cache-name partition
+control.bat --enable-experimental --consistency repair cache-name partition strategy
----
--
Parameters:
@@ -1050,8 +1051,9 @@ Parameters:
[cols="1,3",opts="header"]
|===
| Parameter | Description
-| `cache-name`| Cache to be checked/repaired..
+| `cache-name`| Cache to be checked/repaired.
| `partition`| Cache's partition to be checked/repaired.
+| `strategy`| Repair strategy [LWW, PRIMARY, RELATIVE_MAJORITY, REMOVE, CHECK_ONLY].
|===
=== Status
diff --git a/modules/control-utility/src/main/java/org/apache/ignite/internal/commandline/consistency/ConsistencyCommand.java b/modules/control-utility/src/main/java/org/apache/ignite/internal/commandline/consistency/ConsistencyCommand.java
index 46e2760..d00c29d 100644
--- a/modules/control-utility/src/main/java/org/apache/ignite/internal/commandline/consistency/ConsistencyCommand.java
+++ b/modules/control-utility/src/main/java/org/apache/ignite/internal/commandline/consistency/ConsistencyCommand.java
@@ -22,6 +22,7 @@ import java.util.LinkedHashMap;
import java.util.Map;
import java.util.logging.Logger;
import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.cache.ReadRepairStrategy;
import org.apache.ignite.internal.client.GridClient;
import org.apache.ignite.internal.client.GridClientConfiguration;
import org.apache.ignite.internal.commandline.AbstractCommand;
@@ -134,7 +135,9 @@ public class ConsistencyCommand extends AbstractCommand<Object> {
String cacheName = argIter.nextArg("Expected cache name.");
int part = argIter.nextNonNegativeIntArg("Expected partition.");
- cmdArg = new VisorConsistencyRepairTaskArg(cacheName, part);
+ ReadRepairStrategy strategy = ReadRepairStrategy.fromString(argIter.nextArg("Expected strategy."));
+
+ cmdArg = new VisorConsistencyRepairTaskArg(cacheName, part, strategy);
}
else if (cmd == STATUS)
cmdArg = null;
diff --git a/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerClusterByClassTest.java b/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerClusterByClassTest.java
index edddaa1..f4d712f 100644
--- a/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerClusterByClassTest.java
+++ b/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerClusterByClassTest.java
@@ -1661,7 +1661,7 @@ public class GridCommandHandlerClusterByClassTest extends GridCommandHandlerClus
cmdArgs.put(WAL, asList(new String[] {"print"}, new String[] {"delete"}));
cmdArgs.put(METADATA, asList(new String[] {"help"}, new String[] {"list"}));
cmdArgs.put(TRACING_CONFIGURATION, Collections.singletonList(new String[] {"get_all"}));
- cmdArgs.put(CONSISTENCY, Collections.singletonList(new String[] {"repair", "cache", "0"}));
+ cmdArgs.put(CONSISTENCY, Collections.singletonList(new String[] {"repair", "cache", "0", "LWW"}));
String warning = String.format(
"To use experimental command add --enable-experimental parameter for %s",
diff --git a/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerConsistencyTest.java b/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerConsistencyTest.java
index 6002439..6f4685b 100644
--- a/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerConsistencyTest.java
+++ b/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerConsistencyTest.java
@@ -23,6 +23,7 @@ import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.ignite.Ignite;
import org.apache.ignite.binary.BinaryObjectBuilder;
+import org.apache.ignite.cache.ReadRepairStrategy;
import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
@@ -37,10 +38,11 @@ import org.apache.ignite.internal.processors.dr.GridDrType;
import org.apache.ignite.internal.util.typedef.G;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC;
import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
-import static org.apache.ignite.events.EventType.EVT_CONSISTENCY_VIOLATION;
import static org.apache.ignite.internal.commandline.CommandHandler.EXIT_CODE_OK;
import static org.apache.ignite.internal.commandline.CommandHandler.EXIT_CODE_UNEXPECTED_ERROR;
import static org.apache.ignite.internal.visor.consistency.VisorConsistencyRepairTask.CONSISTENCY_VIOLATIONS_FOUND;
@@ -49,6 +51,7 @@ import static org.apache.ignite.testframework.GridTestUtils.assertContains;
/**
*
*/
+@RunWith(Parameterized.class)
public class GridCommandHandlerConsistencyTest extends GridCommandHandlerClusterPerMethodAbstractTest {
/** Default cache name atomic. */
private static final String DEFAULT_CACHE_NAME_ATOMIC = DEFAULT_CACHE_NAME + "Atomic";
@@ -62,6 +65,23 @@ public class GridCommandHandlerConsistencyTest extends GridCommandHandlerCluster
/** Partitions. */
private static final int PARTITIONS = 32;
+ /** */
+ @Parameterized.Parameters(name = "strategy={0}")
+ public static Iterable<Object[]> data() {
+ List<Object[]> res = new ArrayList<>();
+
+ for (ReadRepairStrategy strategy : ReadRepairStrategy.values())
+ res.add(new Object[] {strategy});
+
+ return res;
+ }
+
+ /**
+ *
+ */
+ @Parameterized.Parameter
+ public ReadRepairStrategy strategy;
+
/**
*
*/
@@ -81,7 +101,6 @@ public class GridCommandHandlerConsistencyTest extends GridCommandHandlerCluster
IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
cfg.setDataStorageConfiguration(null);
- cfg.setIncludeEventTypes(EVT_CONSISTENCY_VIOLATION);
return cfg;
}
@@ -122,13 +141,17 @@ public class GridCommandHandlerConsistencyTest extends GridCommandHandlerCluster
assertContains(log, testOut.toString(),
"conflict partitions has been found: [counterConflicts=0, hashConflicts=" + brokenParts.get());
- readRepairTx(brokenParts, txCacheName);
+ Integer fixesPerEntry = fixesPerEntry();
- assertEquals(PARTITIONS, brokenParts.get()); // Half fixed.
+ readRepair(brokenParts, txCacheName, fixesPerEntry);
- readRepaitAtomic(brokenParts, atomicCacheName);
+ if (fixesPerEntry != null && fixesPerEntry > 0)
+ assertEquals(PARTITIONS, brokenParts.get()); // Half fixed.
- assertEquals(PARTITIONS, brokenParts.get()); // Atomics still broken.
+ readRepair(brokenParts, atomicCacheName, fixesPerEntry != null ? 0 : null);
+
+ if (fixesPerEntry != null && fixesPerEntry > 0)
+ assertEquals(PARTITIONS, brokenParts.get()); // Atomics still broken.
}
/**
@@ -164,9 +187,12 @@ public class GridCommandHandlerConsistencyTest extends GridCommandHandlerCluster
assertContains(log, testOut.toString(),
"conflict partitions has been found: [counterConflicts=0, hashConflicts=" + brokenParts.get());
- readRepairTx(brokenParts, cacheName);
+ Integer fixesPerEntry = fixesPerEntry();
+
+ readRepair(brokenParts, cacheName, fixesPerEntry);
- assertEquals(0, brokenParts.get());
+ if (fixesPerEntry != null)
+ assertEquals(fixesPerEntry > 0 ? 0 : PARTITIONS, brokenParts.get());
}
/**
@@ -179,7 +205,8 @@ public class GridCommandHandlerConsistencyTest extends GridCommandHandlerCluster
injectTestSystemOut();
for (int i = 0; i < PARTITIONS; i++) {
- assertEquals(EXIT_CODE_UNEXPECTED_ERROR, execute("--consistency", "repair", "non-existent", String.valueOf(i)));
+ assertEquals(EXIT_CODE_UNEXPECTED_ERROR,
+ execute("--consistency", "repair", "non-existent", String.valueOf(i), strategy.toString()));
assertContains(log, testOut.toString(), "Cache not found");
}
}
@@ -187,36 +214,48 @@ public class GridCommandHandlerConsistencyTest extends GridCommandHandlerCluster
/**
*
*/
- private void readRepairTx(AtomicInteger brokenParts, String cacheName) {
+ private void readRepair(AtomicInteger brokenParts, String cacheName, Integer fixesPerEntry) {
for (int i = 0; i < PARTITIONS; i++) {
- assertEquals(EXIT_CODE_OK, execute("--consistency", "repair", cacheName, String.valueOf(i)));
+ assertEquals(EXIT_CODE_OK, execute("--consistency", "repair", cacheName, String.valueOf(i), strategy.toString()));
assertContains(log, testOut.toString(), CONSISTENCY_VIOLATIONS_FOUND);
- assertContains(log, testOut.toString(), "[found=1, fixed=1");
+ assertContains(log, testOut.toString(), "[found=1, fixed=" + (fixesPerEntry != null ? fixesPerEntry.toString() : ""));
assertEquals(EXIT_CODE_OK, execute("--cache", "idle_verify"));
- brokenParts.decrementAndGet();
+ if (fixesPerEntry != null)
+ if (fixesPerEntry > 0) {
+ brokenParts.decrementAndGet();
- if (brokenParts.get() > 0)
- assertContains(log, testOut.toString(),
- "conflict partitions has been found: [counterConflicts=0, hashConflicts=" + brokenParts);
- else
- assertContains(log, testOut.toString(), "no conflicts have been found");
+ if (brokenParts.get() > 0)
+ assertContains(log, testOut.toString(),
+ "conflict partitions has been found: [counterConflicts=0, hashConflicts=" + brokenParts);
+ else
+ assertContains(log, testOut.toString(), "no conflicts have been found");
+ }
+ else
+ assertContains(log, testOut.toString(),
+ "conflict partitions has been found: [counterConflicts=0, hashConflicts=" + brokenParts); // Nothing fixed.
}
}
/**
*
*/
- private void readRepaitAtomic(AtomicInteger brokenParts, String cacheName) {
- for (int i = 0; i < PARTITIONS; i++) { // This may be a copy of previous (tx case), implement atomic repair to make this happen :)
- assertEquals(EXIT_CODE_OK, execute("--consistency", "repair", cacheName, String.valueOf(i)));
- assertContains(log, testOut.toString(), CONSISTENCY_VIOLATIONS_FOUND);
- assertContains(log, testOut.toString(), "[found=1, fixed=0"); // Nothing fixed.
+ private Integer fixesPerEntry() {
+ switch (strategy) {
+ case PRIMARY:
+ case REMOVE:
+ return 1;
- assertEquals(EXIT_CODE_OK, execute("--cache", "idle_verify"));
- assertContains(log, testOut.toString(),
- "conflict partitions has been found: [counterConflicts=0, hashConflicts=" + brokenParts); // Nothing fixed.
+ case CHECK_ONLY:
+ return 0;
+
+ case RELATIVE_MAJORITY:
+ case LWW:
+ return null; // Who knows :)
+
+ default:
+ throw new UnsupportedOperationException("Unsupported strategy");
}
}
diff --git a/modules/control-utility/src/test/java/org/apache/ignite/util/KillCommandsCommandShTest.java b/modules/control-utility/src/test/java/org/apache/ignite/util/KillCommandsCommandShTest.java
index ed19f72..2a24d9a 100644
--- a/modules/control-utility/src/test/java/org/apache/ignite/util/KillCommandsCommandShTest.java
+++ b/modules/control-utility/src/test/java/org/apache/ignite/util/KillCommandsCommandShTest.java
@@ -24,6 +24,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.ReadRepairStrategy;
import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.internal.IgniteEx;
@@ -224,11 +225,11 @@ public class KillCommandsCommandShTest extends GridCommandHandlerClusterByClassA
/** */
@Test
public void testCancelConsistencyTask() throws InterruptedException {
- String consistencyCancheName = "consistencyCache";
+ String consistencyCacheName = "consistencyCache";
CacheConfiguration<Integer, Integer> cfg = new CacheConfiguration<>();
- cfg.setName(consistencyCancheName);
+ cfg.setName(consistencyCacheName);
cfg.setBackups(SERVER_NODE_CNT - 1);
cfg.setAffinity(new RendezvousAffinityFunction().setPartitions(1));
@@ -312,7 +313,8 @@ public class KillCommandsCommandShTest extends GridCommandHandlerClusterByClassA
injectTestSystemOut();
- assertEquals(EXIT_CODE_UNEXPECTED_ERROR, execute("--consistency", "repair", consistencyCancheName, "0"));
+ assertEquals(EXIT_CODE_UNEXPECTED_ERROR,
+ execute("--consistency", "repair", consistencyCacheName, "0", ReadRepairStrategy.LWW.toString()));
assertContains(log, testOut.toString(), "Operation execution cancelled.");
assertContains(log, testOut.toString(), VisorConsistencyRepairTask.NOTHING_FOUND);
diff --git a/modules/core/src/main/java/org/apache/ignite/IgniteCache.java b/modules/core/src/main/java/org/apache/ignite/IgniteCache.java
index 1a6e782..85587f9 100644
--- a/modules/core/src/main/java/org/apache/ignite/IgniteCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/IgniteCache.java
@@ -44,6 +44,7 @@ import org.apache.ignite.cache.CacheMetrics;
import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.cache.CachePeekMode;
import org.apache.ignite.cache.PartitionLossPolicy;
+import org.apache.ignite.cache.ReadRepairStrategy;
import org.apache.ignite.cache.query.FieldsQueryCursor;
import org.apache.ignite.cache.query.Query;
import org.apache.ignite.cache.query.QueryCursor;
@@ -188,10 +189,11 @@ public interface IgniteCache<K, V> extends javax.cache.Cache<K, V>, IgniteAsyncS
* <li>{@link IgniteCache#get} && {@link IgniteCache#getAsync}</li>
* <li>{@link IgniteCache#getAll} && {@link IgniteCache#getAllAsync}</li>
* </ul>
+ * @param strategy Read Repair strategy.
* @return Cache with explicit consistency check on each read and repair if necessary.
*/
@IgniteExperimental
- public IgniteCache<K, V> withReadRepair();
+ public IgniteCache<K, V> withReadRepair(ReadRepairStrategy strategy);
/**
* Returns cache that will operate with binary objects.
diff --git a/modules/core/src/main/java/org/apache/ignite/cache/ReadRepairStrategy.java b/modules/core/src/main/java/org/apache/ignite/cache/ReadRepairStrategy.java
new file mode 100644
index 0000000..4d80037
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/cache/ReadRepairStrategy.java
@@ -0,0 +1,89 @@
+/*
+ * 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.cache;
+
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.lang.IgniteExperimental;
+
+/**
+ * Read repair strategies.
+ *
+ * @see IgniteCache#withReadRepair(ReadRepairStrategy) for details.
+ */
+@IgniteExperimental
+public enum ReadRepairStrategy {
+ /** Last write (the newest entry) wins.
+ * <p>
+ * May cause {@link IgniteException} when fix is impossible (unable to detect the newest entry):
+ * <ul>
+ * <li>Null(s) found as well as non-null values for the same key.
+ * <p>
+ * Null (missed entry) has no version, so, it can not be compared with the versioned entry.</li>
+ * <li>Entries with the same version have different values.</li>
+ * </ul>
+ */
+ LWW("LWW"),
+
+ /** Value from the primary node wins. */
+ PRIMARY("PRIMARY"),
+
+ /** The relative majority, any value found more times than any other wins.
+ * <p>
+ * Works for an even number of copies (which is typical of Ignite) instead of an absolute majority.
+ * <p>
+ * May cause {@link IgniteException} when unable to detect value found more times than any other.
+ * <p>
+ * For example, when we have 5 copies (4 backups) and value `A` found twice, but `X`,`Y` and `Z` only once, `A` wins.
+ * But, when `A` found twice, as well as `B`, and `X` only once, the strategy unable to determine the winner.
+ * <p>
+ * When we have 4 copies (3 backups), any value found two or more times, when others are found only once, is the winner.
+ */
+ RELATIVE_MAJORITY("RELATIVE_MAJORITY"),
+
+ /** Inconsistent entries will be removed. */
+ REMOVE("REMOVE"),
+
+ /** Only check will be performed. */
+ CHECK_ONLY("CHECK_ONLY");
+
+ /** Strategy name. */
+ private final String name;
+
+ /**
+ * @param name Name.
+ */
+ ReadRepairStrategy(String name) {
+ this.name = name;
+ }
+
+ /**
+ * Provides strategy by name.
+ *
+ * @param name Text.
+ * @return Read Repair strategy.
+ */
+ public static ReadRepairStrategy fromString(String name) {
+ for (ReadRepairStrategy strategy : values()) {
+ if (strategy.name.equalsIgnoreCase(name))
+ return strategy;
+ }
+
+ throw new IllegalArgumentException("Unknown strategy: " + name);
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/events/CacheConsistencyViolationEvent.java b/modules/core/src/main/java/org/apache/ignite/events/CacheConsistencyViolationEvent.java
index 7512d8c..d02bf6e 100644
--- a/modules/core/src/main/java/org/apache/ignite/events/CacheConsistencyViolationEvent.java
+++ b/modules/core/src/main/java/org/apache/ignite/events/CacheConsistencyViolationEvent.java
@@ -19,6 +19,7 @@ package org.apache.ignite.events;
import java.util.Map;
import org.apache.ignite.cache.CacheEntryVersion;
+import org.apache.ignite.cache.ReadRepairStrategy;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.lang.IgniteExperimental;
@@ -67,28 +68,39 @@ public class CacheConsistencyViolationEvent extends EventAdapter {
private static final long serialVersionUID = 0L;
/** Represents original values of entries.*/
- final Map<Object, Map<ClusterNode, EntryInfo>> entries;
+ private final Map<Object, Map<ClusterNode, EntryInfo>> entries;
+
+ /** Fixed entries. */
+ private final Map<Object, Object> fixed;
/** Cache name. */
- final String cacheName;
+ private final String cacheName;
+
+ /** Strategy. */
+ private final ReadRepairStrategy strategy;
/**
* Creates a new instance of CacheConsistencyViolationEvent.
- *
* @param cacheName Cache name.
* @param node Local node.
* @param msg Event message.
* @param entries Collection of original entries.
+ * @param fixed Collection of fixed entries.
+ * @param strategy Strategy.
*/
public CacheConsistencyViolationEvent(
String cacheName,
ClusterNode node,
String msg,
- Map<Object, Map<ClusterNode, EntryInfo>> entries) {
+ Map<Object, Map<ClusterNode, EntryInfo>> entries,
+ Map<Object, Object> fixed,
+ ReadRepairStrategy strategy) {
super(node, msg, EVT_CONSISTENCY_VIOLATION);
this.cacheName = cacheName;
this.entries = entries;
+ this.fixed = fixed;
+ this.strategy = strategy;
}
/**
@@ -101,6 +113,15 @@ public class CacheConsistencyViolationEvent extends EventAdapter {
}
/**
+ * Returns a mapping of keys to a collection of fixed entries.
+ *
+ * @return Collection of fixed entries.
+ */
+ public Map<Object, Object> getFixedEntries() {
+ return fixed;
+ }
+
+ /**
* Returns cache name.
*
* @return Cache name.
@@ -110,6 +131,15 @@ public class CacheConsistencyViolationEvent extends EventAdapter {
}
/**
+ * Returns strategy.
+ *
+ * @return Strategy.
+ */
+ public ReadRepairStrategy getStrategy() {
+ return strategy;
+ }
+
+ /**
* Inconsistent entry info.
*/
public interface EntryInfo {
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheObjectUtils.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheObjectUtils.java
index 8ff3bc4..88eaeaf 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheObjectUtils.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheObjectUtils.java
@@ -83,7 +83,7 @@ public class CacheObjectUtils {
* @param keepBinary Keep binary flag.
* @return Unwrapped collection.
*/
- public static Collection<Object> unwrapBinariesIfNeeded(CacheObjectValueContext ctx, Collection<Object> col,
+ public static Collection<Object> unwrapBinariesIfNeeded(CacheObjectValueContext ctx, Collection<?> col,
boolean keepBinary) {
return unwrapBinariesIfNeeded(ctx, col, keepBinary, true);
}
@@ -94,7 +94,7 @@ public class CacheObjectUtils {
* @param cpy Copy flag.
* @return Unwrapped collection.
*/
- private static Collection<Object> unwrapKnownCollection(CacheObjectValueContext ctx, Collection<Object> col,
+ private static Collection<Object> unwrapKnownCollection(CacheObjectValueContext ctx, Collection<?> col,
boolean keepBinary, boolean cpy) {
Collection<Object> col0 = BinaryUtils.newKnownCollection(col);
@@ -135,7 +135,7 @@ public class CacheObjectUtils {
* @param cpy Copy value flag.
* @return Unwrapped collection.
*/
- private static Collection<Object> unwrapBinariesIfNeeded(CacheObjectValueContext ctx, Collection<Object> col,
+ private static Collection<Object> unwrapBinariesIfNeeded(CacheObjectValueContext ctx, Collection<?> col,
boolean keepBinary, boolean cpy) {
Collection<Object> col0 = BinaryUtils.newKnownCollection(col);
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheOperationContext.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheOperationContext.java
index f6786a9..7f22f538 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheOperationContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheOperationContext.java
@@ -20,6 +20,7 @@ package org.apache.ignite.internal.processors.cache;
import java.io.Serializable;
import javax.cache.expiry.ExpiryPolicy;
import org.apache.ignite.IgniteSystemProperties;
+import org.apache.ignite.cache.ReadRepairStrategy;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.jetbrains.annotations.Nullable;
@@ -49,8 +50,8 @@ public class CacheOperationContext implements Serializable {
/** */
private final boolean recovery;
- /** Read-repair flag. */
- private final boolean readRepair;
+ /** Read-repair strategy. */
+ private final ReadRepairStrategy readRepairStrategy;
/** Keep binary flag. */
private final boolean keepBinary;
@@ -73,7 +74,7 @@ public class CacheOperationContext implements Serializable {
expiryPlc = null;
noRetries = false;
recovery = false;
- readRepair = false;
+ readRepairStrategy = null;
dataCenterId = null;
allowAtomicOpsInTx = DFLT_ALLOW_ATOMIC_OPS_IN_TX;
}
@@ -83,7 +84,7 @@ public class CacheOperationContext implements Serializable {
* @param keepBinary Keep binary flag.
* @param expiryPlc Expiry policy.
* @param dataCenterId Data center id.
- * @param readRepair Read-repair flag.
+ * @param readRepairStrategy Read-repair strategy.
*/
public CacheOperationContext(
boolean skipStore,
@@ -92,7 +93,7 @@ public class CacheOperationContext implements Serializable {
boolean noRetries,
@Nullable Byte dataCenterId,
boolean recovery,
- boolean readRepair,
+ @Nullable ReadRepairStrategy readRepairStrategy,
boolean allowAtomicOpsInTx
) {
this.skipStore = skipStore;
@@ -101,7 +102,7 @@ public class CacheOperationContext implements Serializable {
this.noRetries = noRetries;
this.dataCenterId = dataCenterId;
this.recovery = recovery;
- this.readRepair = readRepair;
+ this.readRepairStrategy = readRepairStrategy;
this.allowAtomicOpsInTx = allowAtomicOpsInTx;
}
@@ -132,7 +133,7 @@ public class CacheOperationContext implements Serializable {
noRetries,
dataCenterId,
recovery,
- readRepair,
+ readRepairStrategy,
allowAtomicOpsInTx);
}
@@ -166,7 +167,7 @@ public class CacheOperationContext implements Serializable {
noRetries,
dataCenterId,
recovery,
- readRepair,
+ readRepairStrategy,
allowAtomicOpsInTx);
}
@@ -191,7 +192,7 @@ public class CacheOperationContext implements Serializable {
noRetries,
dataCenterId,
recovery,
- readRepair,
+ readRepairStrategy,
allowAtomicOpsInTx);
}
@@ -207,7 +208,7 @@ public class CacheOperationContext implements Serializable {
noRetries,
dataCenterId,
recovery,
- readRepair,
+ readRepairStrategy,
allowAtomicOpsInTx);
}
@@ -223,7 +224,7 @@ public class CacheOperationContext implements Serializable {
noRetries,
dataCenterId,
recovery,
- readRepair,
+ readRepairStrategy,
allowAtomicOpsInTx);
}
@@ -239,15 +240,15 @@ public class CacheOperationContext implements Serializable {
noRetries,
dataCenterId,
recovery,
- readRepair,
+ readRepairStrategy,
allowAtomicOpsInTx);
}
/**
- * @param readRepair Read Repair flag.
+ * @param readRepairStrategy Read Repair strategy.
* @return New instance of CacheOperationContext with Read Repair flag.
*/
- public CacheOperationContext setReadRepair(boolean readRepair) {
+ public CacheOperationContext setReadRepairStrategy(ReadRepairStrategy readRepairStrategy) {
return new CacheOperationContext(
skipStore,
keepBinary,
@@ -255,7 +256,7 @@ public class CacheOperationContext implements Serializable {
noRetries,
dataCenterId,
recovery,
- readRepair,
+ readRepairStrategy,
allowAtomicOpsInTx);
}
@@ -267,10 +268,10 @@ public class CacheOperationContext implements Serializable {
}
/**
- * @return Read Repair flag.
+ * @return Read Repair strategy.
*/
- public boolean readRepair() {
- return readRepair;
+ public ReadRepairStrategy readRepairStrategy() {
+ return readRepairStrategy;
}
/**
@@ -291,7 +292,7 @@ public class CacheOperationContext implements Serializable {
noRetries,
dataCenterId,
recovery,
- readRepair,
+ readRepairStrategy,
true);
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GatewayProtectedCacheProxy.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GatewayProtectedCacheProxy.java
index 75d47f7..1a4366c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GatewayProtectedCacheProxy.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GatewayProtectedCacheProxy.java
@@ -41,6 +41,7 @@ import org.apache.ignite.cache.CacheEntry;
import org.apache.ignite.cache.CacheEntryProcessor;
import org.apache.ignite.cache.CacheMetrics;
import org.apache.ignite.cache.CachePeekMode;
+import org.apache.ignite.cache.ReadRepairStrategy;
import org.apache.ignite.cache.query.FieldsQueryCursor;
import org.apache.ignite.cache.query.Query;
import org.apache.ignite.cache.query.QueryCursor;
@@ -52,6 +53,7 @@ import org.apache.ignite.internal.AsyncSupportAdapter;
import org.apache.ignite.internal.GridKernalState;
import org.apache.ignite.internal.processors.cache.mvcc.MvccUtils;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
+import org.apache.ignite.internal.util.typedef.internal.A;
import org.apache.ignite.lang.IgniteBiPredicate;
import org.apache.ignite.lang.IgniteClosure;
import org.apache.ignite.lang.IgniteFuture;
@@ -232,7 +234,9 @@ public class GatewayProtectedCacheProxy<K, V> extends AsyncSupportAdapter<Ignite
}
/** {@inheritDoc} */
- @Override public IgniteCache<K, V> withReadRepair() {
+ @Override public IgniteCache<K, V> withReadRepair(ReadRepairStrategy strategy) {
+ A.notNull(strategy, "strategy");
+
CacheOperationGate opGate = onEnter();
try {
@@ -245,7 +249,7 @@ public class GatewayProtectedCacheProxy<K, V> extends AsyncSupportAdapter<Ignite
throw new UnsupportedOperationException("Read-repair is incompatible with near caches.");
if (context().readThrough()) {
- // Read Repair get operation produces different versions for same entries loaded via readThrough feature.
+ // Entries loaded via readThrough feature have inconsistent versions by design.
throw new UnsupportedOperationException("Read-repair is incompatible with caches that use readThrough.");
}
@@ -257,12 +261,10 @@ public class GatewayProtectedCacheProxy<K, V> extends AsyncSupportAdapter<Ignite
"at least 1 backup configured for cache.");
}
- boolean readRepair = opCtx.readRepair();
-
- if (readRepair)
+ if (opCtx.readRepairStrategy() == strategy)
return this;
- return new GatewayProtectedCacheProxy<>(delegate, opCtx.setReadRepair(true), lock);
+ return new GatewayProtectedCacheProxy<>(delegate, opCtx.setReadRepairStrategy(strategy), lock);
}
finally {
onLeave(opGate);
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
index d3792a0..3fb52db 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
@@ -43,6 +43,7 @@ import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Function;
import java.util.function.Supplier;
import javax.cache.Cache;
import javax.cache.expiry.ExpiryPolicy;
@@ -59,6 +60,7 @@ import org.apache.ignite.cache.CacheEntry;
import org.apache.ignite.cache.CacheInterceptor;
import org.apache.ignite.cache.CacheMetrics;
import org.apache.ignite.cache.CachePeekMode;
+import org.apache.ignite.cache.ReadRepairStrategy;
import org.apache.ignite.cache.affinity.Affinity;
import org.apache.ignite.cluster.ClusterGroup;
import org.apache.ignite.cluster.ClusterNode;
@@ -94,6 +96,7 @@ import org.apache.ignite.internal.processors.cache.distributed.dht.topology.Grid
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtLocalPartition;
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionTopology;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLocal;
+import org.apache.ignite.internal.processors.cache.distributed.near.consistency.GridCompoundReadRepairFuture;
import org.apache.ignite.internal.processors.cache.distributed.near.consistency.IgniteConsistencyViolationException;
import org.apache.ignite.internal.processors.cache.dr.GridCacheDrInfo;
import org.apache.ignite.internal.processors.cache.mvcc.MvccSnapshot;
@@ -116,7 +119,6 @@ import org.apache.ignite.internal.transactions.IgniteTxRollbackCheckedException;
import org.apache.ignite.internal.transactions.IgniteTxTimeoutCheckedException;
import org.apache.ignite.internal.transactions.TransactionCheckedException;
import org.apache.ignite.internal.util.GridSerializableMap;
-import org.apache.ignite.internal.util.future.GridCompoundFuture;
import org.apache.ignite.internal.util.future.GridEmbeddedFuture;
import org.apache.ignite.internal.util.future.GridFinishedFuture;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
@@ -494,7 +496,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
false,
null,
false,
- false,
+ null,
DFLT_ALLOW_ATOMIC_OPS_IN_TX);
return new GridCacheProxyImpl<>(ctx, this, opCtx);
@@ -509,7 +511,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
false,
null,
false,
- false,
+ null,
DFLT_ALLOW_ATOMIC_OPS_IN_TX);
return new GridCacheProxyImpl<>((GridCacheContext<K1, V1>)ctx, (GridCacheAdapter<K1, V1>)this, opCtx);
@@ -531,7 +533,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
false,
null,
false,
- false,
+ null,
DFLT_ALLOW_ATOMIC_OPS_IN_TX);
return new GridCacheProxyImpl<>(ctx, this, opCtx);
@@ -546,7 +548,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
true,
null,
false,
- false,
+ null,
DFLT_ALLOW_ATOMIC_OPS_IN_TX);
return new GridCacheProxyImpl<>(ctx, this, opCtx);
@@ -561,7 +563,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
false,
null,
false,
- false,
+ null,
DFLT_ALLOW_ATOMIC_OPS_IN_TX);
return new GridCacheProxyImpl<>(ctx, this, opCtx);
@@ -699,12 +701,12 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
/*skip values*/true,
false);
- boolean readRepair = opCtx != null && opCtx.readRepair();
+ ReadRepairStrategy readRepairStrategy = opCtx != null ? opCtx.readRepairStrategy() : null;
- if (readRepair)
+ if (readRepairStrategy != null)
return getWithRepairAsync(
fut,
- () -> repairAsync(key, opCtx, true),
+ (ks) -> repairAsync(ks, opCtx, true),
() -> containsKeyAsync(key));
return fut;
@@ -733,7 +735,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
/*task name*/null,
/*deserialize binary*/false,
opCtx != null && opCtx.recovery(),
- opCtx != null && opCtx.readRepair(),
+ opCtx != null ? opCtx.readRepairStrategy() : null,
/*skip values*/true,
/*need ver*/false).chain(new CX1<IgniteInternalFuture<Map<K, V>>, Boolean>() {
@Override public Boolean applyx(IgniteInternalFuture<Map<K, V>> fut) throws IgniteCheckedException {
@@ -1384,7 +1386,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
taskName,
/*deserialize cache objects*/true,
opCtx != null && opCtx.recovery(),
- opCtx != null && opCtx.readRepair(),
+ opCtx != null ? opCtx.readRepairStrategy() : null,
/*skip values*/false,
/*need ver*/false).get().get(key);
}
@@ -1402,7 +1404,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
taskName,
true,
opCtx != null && opCtx.recovery(),
- opCtx != null && opCtx.readRepair(),
+ opCtx != null ? opCtx.readRepairStrategy() : null,
/*skip vals*/false,
/*can remap*/false).chain(new CX1<IgniteInternalFuture<Map<K, V>>, V>() {
@Override public V applyx(IgniteInternalFuture<Map<K, V>> e) throws IgniteCheckedException {
@@ -1433,7 +1435,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
taskName,
!(opCtx != null && opCtx.isKeepBinary()),
opCtx != null && opCtx.recovery(),
- opCtx != null && opCtx.readRepair(),
+ opCtx != null ? opCtx.readRepairStrategy() : null,
/*skip values*/false,
/*need ver*/false);
@@ -1536,7 +1538,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
key,
!keepBinary,
false,
- opCtx != null && opCtx.readRepair());
+ opCtx != null ? opCtx.readRepairStrategy() : null);
if (ctx.config().getInterceptor() != null)
fut = fut.chain(new CX1<IgniteInternalFuture<V>, V>() {
@@ -1576,7 +1578,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
key0,
!keepBinary,
true,
- opCtx != null && opCtx.readRepair());
+ opCtx != null ? opCtx.readRepairStrategy() : null);
final boolean intercept = ctx.config().getInterceptor() != null;
@@ -1629,7 +1631,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
!ctx.keepBinary(),
false,
opCtx != null && opCtx.recovery(),
- opCtx != null && opCtx.readRepair());
+ opCtx != null ? opCtx.readRepairStrategy() : null);
if (ctx.config().getInterceptor() != null)
map = interceptGet(keys, map);
@@ -1660,7 +1662,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
!ctx.keepBinary(),
true,
opCtx != null && opCtx.recovery(),
- opCtx != null && opCtx.readRepair());
+ opCtx != null ? opCtx.readRepairStrategy() : null);
Collection<CacheEntry<K, V>> res = new HashSet<>();
@@ -1699,7 +1701,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
taskName,
!(opCtx != null && opCtx.isKeepBinary()),
opCtx != null && opCtx.recovery(),
- opCtx != null && opCtx.readRepair(),
+ opCtx != null ? opCtx.readRepairStrategy() : null,
/*skip vals*/false,
/*need ver*/false);
@@ -1741,7 +1743,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
taskName,
!(opCtx != null && opCtx.isKeepBinary()),
opCtx != null && opCtx.recovery(),
- opCtx != null && opCtx.readRepair(),
+ opCtx != null ? opCtx.readRepairStrategy() : null,
/*skip vals*/false,
/*need ver*/true));
@@ -1885,7 +1887,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
taskName,
deserializeBinary,
opCtx != null && opCtx.recovery(),
- opCtx != null && opCtx.readRepair(),
+ opCtx != null ? opCtx.readRepairStrategy() : null,
skipVals,
needVer).chain(
new CX1<IgniteInternalFuture<Map<K, V>>, V>() {
@@ -1924,7 +1926,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
String taskName,
boolean deserializeBinary,
boolean recovery,
- boolean readRepair,
+ ReadRepairStrategy readRepairStrategy,
boolean skipVals,
final boolean needVer
) {
@@ -1937,7 +1939,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
taskName,
deserializeBinary,
opCtx != null && opCtx.recovery(),
- readRepair,
+ readRepairStrategy,
forcePrimary,
skipVals ? null : expiryPolicy(opCtx != null ? opCtx.expiry() : null),
skipVals,
@@ -1966,7 +1968,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
final String taskName,
final boolean deserializeBinary,
final boolean recovery,
- final boolean readRepair,
+ final ReadRepairStrategy readRepairStrategy,
final boolean forcePrimary,
@Nullable IgniteCacheExpiryPolicy expiry,
final boolean skipVals,
@@ -1986,7 +1988,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
skipVals,
/*keep cache objects*/false,
recovery,
- readRepair,
+ readRepairStrategy,
needVer,
null,
null); // TODO IGNITE-7371
@@ -2018,7 +2020,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
final boolean skipVals,
final boolean keepCacheObjects,
final boolean recovery,
- final boolean readRepair,
+ final ReadRepairStrategy readRepairStrategy,
final boolean needVer,
@Nullable String txLbl,
MvccSnapshot mvccSnapshot
@@ -2386,7 +2388,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
false,
!readThrough,
recovery,
- readRepair,
+ readRepairStrategy,
needVer);
}
}, ctx.operationContextPerCall(), /*retry*/false);
@@ -4836,7 +4838,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
needVer);
}
catch (IgniteConsistencyViolationException e) {
- repairAsync(key, ctx.operationContextPerCall(), false).get();
+ repairAsync(e.keys(), ctx.operationContextPerCall(), false).get();
return repairableGet(key, deserializeBinary, needVer);
}
@@ -4882,7 +4884,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
final K key,
boolean deserializeBinary,
final boolean needVer,
- boolean readRepair) {
+ ReadRepairStrategy readRepairStrategy) {
try {
checkJta();
}
@@ -4898,13 +4900,13 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
/*skip vals*/false,
needVer);
- if (readRepair) {
+ if (readRepairStrategy != null) {
CacheOperationContext opCtx = ctx.operationContextPerCall();
return getWithRepairAsync(
fut,
- () -> repairAsync(key, opCtx, false),
- () -> repairableGetAsync(key, deserializeBinary, needVer, readRepair));
+ (ks) -> repairAsync(ks, opCtx, false),
+ () -> repairableGetAsync(key, deserializeBinary, needVer, readRepairStrategy));
}
return fut;
@@ -4922,14 +4924,14 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
boolean deserializeBinary,
boolean needVer,
boolean recovery,
- boolean readRepair) throws IgniteCheckedException {
+ ReadRepairStrategy readRepairStrategy) throws IgniteCheckedException {
try {
- return getAll(keys, deserializeBinary, needVer, recovery, readRepair);
+ return getAll(keys, deserializeBinary, needVer, recovery, readRepairStrategy);
}
catch (IgniteConsistencyViolationException e) {
- repairAsync(keys, ctx.operationContextPerCall(), false).get();
+ repairAsync(e.keys(), ctx.operationContextPerCall(), false).get();
- return repairableGetAll(keys, deserializeBinary, needVer, recovery, readRepair);
+ return repairableGetAll(keys, deserializeBinary, needVer, recovery, readRepairStrategy);
}
}
@@ -4938,7 +4940,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
* @param deserializeBinary Deserialize binary flag.
* @param needVer Need version.
* @param recovery Recovery flag.
- * @param readRepair Read repair flag.
+ * @param readRepairStrategy Read repair strategy.
* @return Map of cached values.
* @throws IgniteCheckedException If read failed.
*/
@@ -4947,7 +4949,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
boolean deserializeBinary,
boolean needVer,
boolean recovery,
- boolean readRepair) throws IgniteCheckedException {
+ ReadRepairStrategy readRepairStrategy) throws IgniteCheckedException {
checkJta();
return getAllAsync(keys,
@@ -4956,7 +4958,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
ctx.kernalContext().job().currentTaskName(),
deserializeBinary,
recovery,
- readRepair,
+ readRepairStrategy,
/*skip vals*/false,
needVer).get();
}
@@ -4980,7 +4982,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
String taskName,
boolean deserializeBinary,
boolean recovery,
- boolean readRepair,
+ ReadRepairStrategy readRepairStrategy,
boolean skipVals,
final boolean needVer
) {
@@ -4991,16 +4993,16 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
taskName,
deserializeBinary,
recovery,
- readRepair,
+ readRepairStrategy,
skipVals,
needVer);
- if (readRepair) {
+ if (readRepairStrategy != null) {
CacheOperationContext opCtx = ctx.operationContextPerCall();
return getWithRepairAsync(
fut,
- () -> repairAsync(keys, opCtx, skipVals),
+ (ks) -> repairAsync(ks, opCtx, skipVals),
() -> repairableGetAllAsync(
keys,
forcePrimary,
@@ -5008,7 +5010,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
taskName,
deserializeBinary,
recovery,
- readRepair,
+ readRepairStrategy,
skipVals,
needVer));
}
@@ -5024,7 +5026,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
*/
private <R> IgniteInternalFuture<R> getWithRepairAsync(
IgniteInternalFuture<R> orig,
- Supplier<IgniteInternalFuture<Void>> repair,
+ Function<Collection<KeyCacheObject>, IgniteInternalFuture<Void>> repair,
Supplier<IgniteInternalFuture<R>> retry) {
final GridNearTxLocal tx = checkCurrentTx();
final CacheOperationContext opCtx = ctx.operationContextPerCall();
@@ -5036,7 +5038,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
fut.onDone(f.get());
}
catch (IgniteConsistencyViolationException e1) {
- repair.get().listen((repFut) -> {
+ repair.apply(e1.keys()).listen((repFut) -> {
if (repFut.error() != null)
fut.onDone(repFut.error());
else {
@@ -5074,13 +5076,13 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
* @param skipVals Skip values flag.
* @return Compound future that represents a result of repair action.
*/
- protected IgniteInternalFuture<Void> repairAsync(
- Collection<? extends K> keys,
+ private IgniteInternalFuture<Void> repairAsync(
+ Collection<KeyCacheObject> keys,
final CacheOperationContext opCtx,
boolean skipVals) {
- GridCompoundFuture fut = new GridCompoundFuture();
+ GridCompoundReadRepairFuture fut = new GridCompoundReadRepairFuture();
- for (K key : keys)
+ for (KeyCacheObject key : keys)
fut.add(repairAsync(key, opCtx, skipVals));
fut.markInitialized();
@@ -5096,16 +5098,17 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
* @param skipVals Skip values flag.
* @return Recover future.
*/
- protected IgniteInternalFuture<Void> repairAsync(
- final K key,
+ private IgniteInternalFuture<Void> repairAsync(
+ final KeyCacheObject key,
final CacheOperationContext opCtx,
boolean skipVals) {
assert ctx.transactional();
final GridNearTxLocal orig = checkCurrentTx();
- // Pessimistic non-read-committed 'get' should be fixed inside its own tx, the only exception is 'contains'.
- assert orig == null || orig.optimistic() || orig.readCommitted() || /*contains*/ skipVals;
+ assert orig == null || orig.optimistic() || orig.readCommitted() || /*contains*/ skipVals :
+ "Pessimistic non-read-committed 'get' should be fixed inside its own tx, the only exception is 'contains' " +
+ "[tx=" + orig + ", skipVals=" + skipVals + "]";
// Async check and recover if necessary.
return ctx.kernalContext().closure().callLocalSafe(new GridPlainCallable<Void>() {
@@ -5115,7 +5118,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
ctx.operationContextPerCall(opCtx);
try (Transaction tx = ctx.grid().transactions().txStart(PESSIMISTIC, SERIALIZABLE)) {
- get(key); // Repair.
+ get((K)key, null, false, false); // Repair.
final GridNearTxLocal tx0 = checkCurrentTx();
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java
index 281dc2a..670b5a6 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java
@@ -1771,7 +1771,7 @@ public class GridCacheContext<K, V> implements Externalizable {
* @param keepBinary Keep binary flag.
* @return Unwrapped collection.
*/
- public Collection<Object> unwrapBinariesIfNeeded(Collection<Object> col, boolean keepBinary) {
+ public Collection<Object> unwrapBinariesIfNeeded(Collection<?> col, boolean keepBinary) {
return CacheObjectUtils.unwrapBinariesIfNeeded(cacheObjCtx, col, keepBinary);
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProxyImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProxyImpl.java
index 0e2ff38..3d66277 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProxyImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProxyImpl.java
@@ -294,7 +294,7 @@ public class GridCacheProxyImpl<K, V> implements IgniteInternalCache<K, V>, Exte
false,
null,
false,
- false,
+ null,
DFLT_ALLOW_ATOMIC_OPS_IN_TX));
}
finally {
@@ -316,7 +316,7 @@ public class GridCacheProxyImpl<K, V> implements IgniteInternalCache<K, V>, Exte
false,
null,
false,
- false,
+ null,
DFLT_ALLOW_ATOMIC_OPS_IN_TX));
}
@@ -1566,7 +1566,7 @@ public class GridCacheProxyImpl<K, V> implements IgniteInternalCache<K, V>, Exte
false,
null,
false,
- false,
+ null,
DFLT_ALLOW_ATOMIC_OPS_IN_TX));
}
finally {
@@ -1587,7 +1587,7 @@ public class GridCacheProxyImpl<K, V> implements IgniteInternalCache<K, V>, Exte
true,
null,
false,
- false,
+ null,
DFLT_ALLOW_ATOMIC_OPS_IN_TX));
}
finally {
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxyImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxyImpl.java
index e607f17..a962b24 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxyImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxyImpl.java
@@ -56,6 +56,7 @@ import org.apache.ignite.cache.CacheEntryProcessor;
import org.apache.ignite.cache.CacheManager;
import org.apache.ignite.cache.CacheMetrics;
import org.apache.ignite.cache.CachePeekMode;
+import org.apache.ignite.cache.ReadRepairStrategy;
import org.apache.ignite.cache.query.AbstractContinuousQuery;
import org.apache.ignite.cache.query.ContinuousQuery;
import org.apache.ignite.cache.query.ContinuousQueryWithTransformer;
@@ -369,7 +370,7 @@ public class IgniteCacheProxyImpl<K, V> extends AsyncSupportAdapter<IgniteCache<
}
/** {@inheritDoc} */
- @Override public IgniteCache<K, V> withReadRepair() {
+ @Override public IgniteCache<K, V> withReadRepair(ReadRepairStrategy strategy) {
throw new UnsupportedOperationException();
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java
index 56550d1..f6744e8 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java
@@ -31,6 +31,7 @@ import javax.cache.Cache;
import javax.cache.expiry.ExpiryPolicy;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
+import org.apache.ignite.cache.ReadRepairStrategy;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.events.DiscoveryEvent;
import org.apache.ignite.events.Event;
@@ -659,7 +660,7 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
String taskName,
boolean deserializeBinary,
boolean recovery,
- boolean readRepair,
+ ReadRepairStrategy readRepairStrategy,
boolean skipVals,
boolean needVer
) {
@@ -672,7 +673,7 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
taskName,
deserializeBinary,
opCtx != null && opCtx.recovery(),
- readRepair,
+ readRepairStrategy,
forcePrimary,
null,
skipVals,
@@ -711,7 +712,7 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
skipVals,
/*keep cache objects*/true,
recovery,
- false,
+ null,
/*need version*/true,
txLbl,
mvccSnapshot);
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
index c7a24b2..2a73bc3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
@@ -37,6 +37,7 @@ import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.IgniteSystemProperties;
import org.apache.ignite.binary.BinaryInvalidTypeException;
+import org.apache.ignite.cache.ReadRepairStrategy;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.NodeStoppingException;
@@ -83,7 +84,6 @@ import org.apache.ignite.internal.processors.cache.distributed.near.GridNearGetR
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSingleGetRequest;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSingleGetResponse;
import org.apache.ignite.internal.processors.cache.distributed.near.consistency.GridNearReadRepairCheckOnlyFuture;
-import org.apache.ignite.internal.processors.cache.distributed.near.consistency.IgniteConsistencyViolationException;
import org.apache.ignite.internal.processors.cache.dr.GridCacheDrExpirationInfo;
import org.apache.ignite.internal.processors.cache.dr.GridCacheDrInfo;
import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
@@ -474,7 +474,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
final ExpiryPolicy expiryPlc = skipVals ? null : opCtx != null ? opCtx.expiry() : null;
final boolean skipStore = opCtx != null && opCtx.skipStore();
final boolean recovery = opCtx != null && opCtx.recovery();
- final boolean readRepair = opCtx != null && opCtx.readRepair();
+ final ReadRepairStrategy readRepairStrategy = opCtx != null ? opCtx.readRepairStrategy() : null;
return asyncOp(new CO<IgniteInternalFuture<V>>() {
@Override public IgniteInternalFuture<V> apply() {
@@ -483,7 +483,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
taskName,
deserializeBinary,
recovery,
- readRepair,
+ readRepairStrategy,
expiryPlc,
skipVals,
skipStore,
@@ -498,13 +498,13 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
boolean deserializeBinary,
boolean needVer,
boolean recovery,
- boolean readRepair) throws IgniteCheckedException {
+ ReadRepairStrategy readRepairStrategy) throws IgniteCheckedException {
return getAllAsyncInternal(keys,
!ctx.config().isReadFromBackup(),
ctx.kernalContext().job().currentTaskName(),
deserializeBinary,
recovery,
- readRepair,
+ readRepairStrategy,
false,
needVer,
false).get();
@@ -518,7 +518,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
final String taskName,
final boolean deserializeBinary,
final boolean recovery,
- final boolean readRepair,
+ final ReadRepairStrategy readRepairStrategy,
final boolean skipVals,
final boolean needVer
) {
@@ -527,7 +527,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
taskName,
deserializeBinary,
recovery,
- readRepair,
+ readRepairStrategy,
skipVals,
needVer,
true);
@@ -538,7 +538,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
* @param forcePrimary Force primary flag.
* @param taskName Task name.
* @param deserializeBinary Deserialize binary flag.
- * @param readRepair Read Repair flag.
+ * @param readRepairStrategy Read Repair strategy.
* @param skipVals Skip values flag.
* @param needVer Need version flag.
* @param asyncOp Async operation flag.
@@ -550,7 +550,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
final String taskName,
final boolean deserializeBinary,
final boolean recovery,
- final boolean readRepair,
+ final ReadRepairStrategy readRepairStrategy,
final boolean skipVals,
final boolean needVer,
boolean asyncOp
@@ -576,7 +576,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
taskName,
deserializeBinary,
recovery,
- readRepair,
+ readRepairStrategy,
expiryPlc,
skipVals,
skipStore,
@@ -590,7 +590,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
taskName,
deserializeBinary,
recovery,
- readRepair,
+ readRepairStrategy,
expiryPlc,
skipVals,
skipStore,
@@ -1374,7 +1374,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
* @param forcePrimary Force primary flag.
* @param taskName Task name.
* @param deserializeBinary Deserialize binary flag.
- * @param readRepair Read Repair flag.
+ * @param readRepairStrategy Read Repair strategy.
* @param expiryPlc Expiry policy.
* @param skipVals Skip values flag.
* @param skipStore Skip store flag.
@@ -1386,7 +1386,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
String taskName,
boolean deserializeBinary,
boolean recovery,
- boolean readRepair,
+ ReadRepairStrategy readRepairStrategy,
@Nullable ExpiryPolicy expiryPlc,
boolean skipVals,
boolean skipStore,
@@ -1396,10 +1396,11 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
IgniteCacheExpiryPolicy expiry = skipVals ? null : expiryPolicy(expiryPlc);
- if (readRepair) {
+ if (readRepairStrategy != null) {
return new GridNearReadRepairCheckOnlyFuture(
ctx,
Collections.singleton(ctx.toCacheKeyObject(key)),
+ readRepairStrategy,
!skipStore,
taskName,
deserializeBinary,
@@ -1449,7 +1450,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
String taskName,
boolean deserializeBinary,
boolean recovery,
- boolean readRepair,
+ ReadRepairStrategy readRepairStrategy,
@Nullable ExpiryPolicy expiryPlc,
boolean skipVals,
boolean skipStore,
@@ -1461,10 +1462,11 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
final boolean evt = !skipVals;
- if (readRepair) {
+ if (readRepairStrategy != null) {
return new GridNearReadRepairCheckOnlyFuture(
ctx,
ctx.cacheKeysView(keys),
+ readRepairStrategy,
!skipStore,
taskName,
deserializeBinary,
@@ -1712,20 +1714,6 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
}
}
- /** {@inheritDoc} */
- @Override protected IgniteInternalFuture<Void> repairAsync(Collection<? extends K> keys,
- final CacheOperationContext opCtx, boolean skipVals) {
- return new GridFinishedFuture<>(
- new IgniteConsistencyViolationException("Consistency violation detected during Read Repair procedure."));
- }
-
- /** {@inheritDoc} */
- @Override protected IgniteInternalFuture<Void> repairAsync(final K key, final CacheOperationContext opCtx,
- boolean skipVals) {
- return new GridFinishedFuture<>(
- new IgniteConsistencyViolationException("Consistency violation detected during Read Repair procedure."));
- }
-
/**
* @param nodeId Node ID.
* @param req Update request.
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java
index a120ea8..6af3fda 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java
@@ -25,6 +25,7 @@ import java.util.LinkedList;
import java.util.Map;
import java.util.UUID;
import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.cache.ReadRepairStrategy;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
@@ -191,7 +192,7 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
final CacheOperationContext opCtx = ctx.operationContextPerCall();
final boolean recovery = opCtx != null && opCtx.recovery();
- final boolean readRepair = opCtx != null && opCtx.readRepair();
+ final ReadRepairStrategy readRepairStrategy = opCtx != null ? opCtx.readRepairStrategy() : null;
// Get operation bypass Tx in Mvcc mode.
if (!ctx.mvccEnabled() && tx != null && !tx.implicit() && !skipTx) {
@@ -205,7 +206,7 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
false,
opCtx != null && opCtx.skipStore(),
recovery,
- readRepair,
+ readRepairStrategy,
needVer);
return fut.chain(new CX1<IgniteInternalFuture<Map<Object, Object>>, V>() {
@@ -257,10 +258,11 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
else
topVer = ctx.affinity().affinityTopologyVersion();
- if (readRepair) {
+ if (readRepairStrategy != null) {
return new GridNearReadRepairCheckOnlyFuture(
ctx,
Collections.singleton(ctx.toCacheKeyObject(key)),
+ readRepairStrategy,
opCtx == null || !opCtx.skipStore(),
taskName,
deserializeBinary,
@@ -311,7 +313,7 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
String taskName,
final boolean deserializeBinary,
final boolean recovery,
- final boolean readRepair,
+ final ReadRepairStrategy readRepairStrategy,
final boolean skipVals,
final boolean needVer
) {
@@ -339,7 +341,7 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
false,
opCtx != null && opCtx.skipStore(),
recovery,
- readRepair,
+ readRepairStrategy,
needVer);
}
}, opCtx, /*retry*/false);
@@ -374,10 +376,11 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
else
topVer = ctx.affinity().affinityTopologyVersion();
- if (readRepair) {
+ if (readRepairStrategy != null) {
return new GridNearReadRepairCheckOnlyFuture(
ctx,
ctx.cacheKeysView(keys),
+ readRepairStrategy,
opCtx == null || !opCtx.skipStore(),
taskName,
deserializeBinary,
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearAtomicCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearAtomicCache.java
index 114f87f..03099c3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearAtomicCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearAtomicCache.java
@@ -29,6 +29,7 @@ import javax.cache.processor.EntryProcessor;
import javax.cache.processor.EntryProcessorException;
import javax.cache.processor.EntryProcessorResult;
import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.cache.ReadRepairStrategy;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.CacheEntryPredicate;
@@ -414,7 +415,7 @@ public class GridNearAtomicCache<K, V> extends GridNearCacheAdapter<K, V> {
String taskName,
boolean deserializeBinary,
boolean recovery,
- boolean readRepair,
+ ReadRepairStrategy readRepairStrategy,
boolean skipVals,
boolean needVer
) {
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java
index 389d109..5887b8d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java
@@ -25,6 +25,7 @@ import java.util.List;
import java.util.Map;
import java.util.UUID;
import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.cache.ReadRepairStrategy;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
@@ -121,7 +122,7 @@ public class GridNearTransactionalCache<K, V> extends GridNearCacheAdapter<K, V>
String taskName,
final boolean deserializeBinary,
final boolean recovery,
- final boolean readRepair,
+ final ReadRepairStrategy readRepairStrategy,
final boolean skipVals,
final boolean needVer
) {
@@ -149,7 +150,7 @@ public class GridNearTransactionalCache<K, V> extends GridNearCacheAdapter<K, V>
false,
skipStore,
recovery,
- readRepair,
+ readRepairStrategy,
needVer);
}
}, opCtx, /*retry*/false);
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java
index 332bb3b..733ec86 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java
@@ -36,6 +36,7 @@ import javax.cache.CacheException;
import javax.cache.expiry.ExpiryPolicy;
import javax.cache.processor.EntryProcessor;
import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.cache.ReadRepairStrategy;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.NodeStoppingException;
@@ -2225,7 +2226,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
* @param skipVals Skip values flag.
* @param keepCacheObjects Keep cache objects
* @param skipStore Skip store flag.
- * @param readRepair Read Repair flag.
+ * @param readRepairStrategy Read Repair strategy.
* @return Future for this get.
*/
@SuppressWarnings("unchecked")
@@ -2238,7 +2239,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
final boolean keepCacheObjects,
final boolean skipStore,
final boolean recovery,
- final boolean readRepair,
+ final ReadRepairStrategy readRepairStrategy,
final boolean needVer) {
if (F.isEmpty(keys))
return new GridFinishedFuture<>(Collections.<K, V>emptyMap());
@@ -2283,7 +2284,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
keepCacheObjects,
skipStore,
recovery,
- readRepair,
+ readRepairStrategy,
needVer);
}
catch (IgniteCheckedException e) {
@@ -2448,16 +2449,17 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
keepCacheObjects,
skipStore,
recovery,
- readRepair,
+ readRepairStrategy,
needVer,
expiryPlc0);
}
- if (readRepair) {
+ if (readRepairStrategy != null) { // Checking and repairing each locked entry (if necessary).
return new GridNearReadRepairFuture(
topVer != null ? topVer : topologyVersion(),
cacheCtx,
keys,
+ readRepairStrategy,
!skipStore,
taskName,
deserializeBinary,
@@ -2469,12 +2471,13 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
// For every fixed entry.
for (Map.Entry<KeyCacheObject, EntryGetResult> entry : fut.get().entrySet()) {
EntryGetResult getRes = entry.getValue();
+ KeyCacheObject key = entry.getKey();
- enlistWrite(
+ enlistWrite( // Fixing inconsistency on commit.
cacheCtx,
entryTopVer,
- entry.getKey(),
- getRes.value(),
+ key,
+ getRes != null ? getRes.value() : null,
expiryPlc0,
null,
null,
@@ -2489,19 +2492,25 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
null);
// Rewriting fixed, initially filled by explicit lock operation.
- cacheCtx.addResult(retMap,
- entry.getKey(),
- getRes.value(),
- skipVals,
- keepCacheObjects,
- deserializeBinary,
- false,
- getRes,
- getRes.version(),
- 0,
- 0,
- needVer,
- U.deploymentClassLoader(cctx.kernalContext(), deploymentLdrId));
+ if (getRes != null)
+ cacheCtx.addResult(retMap,
+ key,
+ getRes.value(),
+ skipVals,
+ keepCacheObjects,
+ deserializeBinary,
+ false,
+ getRes,
+ getRes.version(),
+ 0,
+ 0,
+ needVer,
+ U.deploymentClassLoader(cctx.kernalContext(), deploymentLdrId));
+ else
+ retMap.remove(keepCacheObjects ? key :
+ cacheCtx.cacheObjectContext().unwrapBinaryIfNeeded(
+ key, !deserializeBinary, false,
+ U.deploymentClassLoader(cctx.kernalContext(), deploymentLdrId)));
}
return Collections.emptyMap();
@@ -2585,7 +2594,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
keepCacheObjects,
skipStore,
recovery,
- readRepair,
+ readRepairStrategy,
needVer,
expiryPlc);
}
@@ -2629,7 +2638,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
boolean keepCacheObjects,
boolean skipStore,
boolean recovery,
- boolean readRepair,
+ ReadRepairStrategy readRepairStrategy,
final boolean needVer
) throws IgniteCheckedException {
assert !F.isEmpty(keys);
@@ -2802,7 +2811,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
GridCacheVersion readVer = null;
EntryGetResult getRes = null;
- if ((!pessimistic() || (readCommitted() && !skipVals)) && !readRepair) {
+ if ((!pessimistic() || (readCommitted() && !skipVals)) && readRepairStrategy == null) {
IgniteCacheExpiryPolicy accessPlc =
optimistic() ? accessPolicy(cacheCtx, txKey, expiryPlc) : null;
@@ -3024,7 +3033,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
needReadVer,
keepBinary,
recovery,
- false,
+ null,
expiryPlc,
c);
}
@@ -3126,7 +3135,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
* @param skipVals Skip values flag.
* @param needVer If {@code true} version is required for loaded values.
* @param c Closure to be applied for loaded values.
- * @param readRepair Read Repair flag.
+ * @param readRepairStrategy Read Repair strategy.
* @param expiryPlc Expiry policy.
* @return Future with {@code True} value if loading took place.
*/
@@ -3140,7 +3149,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
final boolean needVer,
boolean keepBinary,
boolean recovery,
- boolean readRepair,
+ ReadRepairStrategy readRepairStrategy,
final ExpiryPolicy expiryPlc,
final GridInClosure3<KeyCacheObject, Object, GridCacheVersion> c
) {
@@ -3176,10 +3185,11 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
});
}
else if (cacheCtx.isColocated()) {
- if (readRepair) {
+ if (readRepairStrategy != null) {
return new GridNearReadRepairCheckOnlyFuture(
cacheCtx,
keys,
+ readRepairStrategy,
readThrough,
taskName,
false,
@@ -4894,7 +4904,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
* @param skipVals Skip values flag.
* @param keepCacheObjects Keep cache objects flag.
* @param skipStore Skip store flag.
- * @param readRepair Read Repair flag.
+ * @param readRepairStrategy Read Repair strategy.
* @param expiryPlc Expiry policy.
* @return Loaded key-value pairs.
*/
@@ -4908,7 +4918,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
final boolean keepCacheObjects,
final boolean skipStore,
final boolean recovery,
- final boolean readRepair,
+ final ReadRepairStrategy readRepairStrategy,
final boolean needVer,
final ExpiryPolicy expiryPlc
) {
@@ -4945,7 +4955,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
needReadVer,
!deserializeBinary,
recovery,
- readRepair,
+ readRepairStrategy,
expiryPlc,
new GridInClosure3<KeyCacheObject, Object, GridCacheVersion>() {
@Override public void apply(KeyCacheObject key, Object val, GridCacheVersion loadVer) {
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/consistency/GridCompoundReadRepairFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/consistency/GridCompoundReadRepairFuture.java
new file mode 100644
index 0000000..3eaacdc
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/consistency/GridCompoundReadRepairFuture.java
@@ -0,0 +1,106 @@
+/*
+ * 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.processors.cache.distributed.near.consistency;
+
+import java.util.Collection;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.util.future.GridFutureAdapter;
+import org.apache.ignite.lang.IgniteInClosure;
+
+/**
+ * Compound future that represents the result of the external fixes for some keys.
+ */
+public class GridCompoundReadRepairFuture extends GridFutureAdapter<Void> implements IgniteInClosure<IgniteInternalFuture<Void>> {
+ /** Listener calls updater. */
+ private static final AtomicIntegerFieldUpdater<GridCompoundReadRepairFuture> LSNR_CALLS_UPD =
+ AtomicIntegerFieldUpdater.newUpdater(GridCompoundReadRepairFuture.class, "lsnrCalls");
+
+ /** Initialized. */
+ private volatile boolean inited;
+
+ /** Listener calls. */
+ private volatile int lsnrCalls;
+
+ /** Count of compounds in the future. */
+ private volatile int size;
+
+ /** Irreparable Keys. */
+ private volatile Collection<Object> irreparableKeys;
+
+ /**
+ * @param fut Future.
+ */
+ public void add(IgniteInternalFuture<Void> fut) {
+ size++; // All additions are from the same thread.
+
+ fut.listen(this);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void apply(IgniteInternalFuture<Void> fut) {
+ Throwable e = fut.error();
+
+ if (e != null) {
+ if (e instanceof IgniteIrreparableConsistencyViolationException) {
+ Collection<Object> repairableKey = ((IgniteIrreparableConsistencyViolationException)e).repairableKeys();
+ Collection<Object> irreparableKeys = ((IgniteIrreparableConsistencyViolationException)e).irreparableKeys();
+
+ assert repairableKey == null || repairableKey.isEmpty() : repairableKey.size();
+ assert irreparableKeys.size() == 1 : irreparableKeys.size(); // Single key fix.
+
+ synchronized (this) {
+ if (this.irreparableKeys == null)
+ this.irreparableKeys = ConcurrentHashMap.newKeySet();
+ }
+
+ this.irreparableKeys.addAll(irreparableKeys);
+ }
+ else
+ onDone(e);
+ }
+
+ LSNR_CALLS_UPD.incrementAndGet(this);
+
+ checkComplete();
+ }
+
+ /**
+ * Mark this future as initialized.
+ */
+ public final void markInitialized() {
+ inited = true;
+
+ checkComplete();
+ }
+
+ /**
+ * Check completeness of the future.
+ */
+ private void checkComplete() {
+ assert lsnrCalls <= size;
+
+ if (inited && !isDone() && lsnrCalls == size) {
+ if (irreparableKeys == null)
+ onDone();
+ else
+ onDone(new IgniteIrreparableConsistencyViolationException(null, irreparableKeys));
+ }
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/consistency/GridNearReadRepairAbstractFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/consistency/GridNearReadRepairAbstractFuture.java
index 61fd137..c0cdb3d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/consistency/GridNearReadRepairAbstractFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/consistency/GridNearReadRepairAbstractFuture.java
@@ -17,7 +17,9 @@
package org.apache.ignite.internal.processors.cache.distributed.near.consistency;
+import java.util.Arrays;
import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -25,7 +27,9 @@ import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.cache.CacheEntryVersion;
+import org.apache.ignite.cache.ReadRepairStrategy;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.events.CacheConsistencyViolationEvent;
import org.apache.ignite.internal.IgniteInternalFuture;
@@ -33,6 +37,7 @@ import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.cluster.ClusterTopologyServerNotFoundException;
import org.apache.ignite.internal.managers.eventstorage.GridEventStorageManager;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.cache.CacheObject;
import org.apache.ignite.internal.processors.cache.EntryGetResult;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.IgniteCacheExpiryPolicy;
@@ -88,13 +93,16 @@ public abstract class GridNearReadRepairAbstractFuture extends GridFutureAdapter
protected final IgniteInternalTx tx;
/** Primaries per key. */
- private volatile Map<KeyCacheObject, ClusterNode> primaries;
+ protected volatile Map<KeyCacheObject, ClusterNode> primaries;
+
+ /** Strategy. */
+ protected final ReadRepairStrategy strategy;
/** Remap flag. */
private final boolean canRemap;
/** Latest mapped topology version. */
- private AffinityTopologyVersion topVer;
+ private volatile AffinityTopologyVersion topVer;
/**
* Creates a new instance of GridNearReadRepairAbstractFuture.
@@ -102,6 +110,7 @@ public abstract class GridNearReadRepairAbstractFuture extends GridFutureAdapter
* @param topVer Topology version.
* @param ctx Cache context.
* @param keys Keys.
+ * @param strategy Read repair strategy.
* @param readThrough Read-through flag.
* @param taskName Task name.
* @param deserializeBinary Deserialize binary flag.
@@ -113,6 +122,7 @@ public abstract class GridNearReadRepairAbstractFuture extends GridFutureAdapter
AffinityTopologyVersion topVer,
GridCacheContext<KeyCacheObject, EntryGetResult> ctx,
Collection<KeyCacheObject> keys,
+ ReadRepairStrategy strategy,
boolean readThrough,
String taskName,
boolean deserializeBinary,
@@ -128,17 +138,26 @@ public abstract class GridNearReadRepairAbstractFuture extends GridFutureAdapter
this.expiryPlc = expiryPlc;
this.tx = tx;
+ assert strategy != null;
+
+ this.strategy = strategy;
+
canRemap = topVer == null;
- map(canRemap ? ctx.affinity().affinityTopologyVersion() : topVer);
+ this.topVer = canRemap ? ctx.affinity().affinityTopologyVersion() : topVer;
}
/**
- * @param topVer Affinity topology version.
+ *
*/
- protected synchronized void map(AffinityTopologyVersion topVer) {
- this.topVer = topVer;
+ protected final void init() {
+ map();
+ }
+ /**
+ *
+ */
+ private synchronized void map() {
assert futs.isEmpty() : "Remapping started without the clean-up.";
Map<KeyCacheObject, ClusterNode> primaryNodes = new HashMap<>();
@@ -199,10 +218,12 @@ public abstract class GridNearReadRepairAbstractFuture extends GridFutureAdapter
/**
* @param topVer Topology version.
*/
- protected void remap(AffinityTopologyVersion topVer) {
+ protected final void remap(AffinityTopologyVersion topVer) {
futs.clear();
- map(topVer);
+ this.topVer = topVer;
+
+ map();
}
/**
@@ -210,7 +231,7 @@ public abstract class GridNearReadRepairAbstractFuture extends GridFutureAdapter
*
* @param finished Future represents a result of GET operation.
*/
- protected synchronized void onResult(IgniteInternalFuture<Map<KeyCacheObject, EntryGetResult>> finished) {
+ protected final synchronized void onResult(IgniteInternalFuture<Map<KeyCacheObject, EntryGetResult>> finished) {
if (isDone() // All subfutures (including currently processing) were successfully finished at previous future processing.
|| (topVer == null) // Remapping, ignoring any updates until remapped.
|| !futs.containsValue((GridPartitionedGetFuture<KeyCacheObject, EntryGetResult>)finished)) // Remapped.
@@ -255,50 +276,116 @@ public abstract class GridNearReadRepairAbstractFuture extends GridFutureAdapter
protected abstract void reduce();
/**
+ * Checks consistency.
+ *
+ * @return Regular `get` result when data is consistent.
+ */
+ protected final Map<KeyCacheObject, EntryGetResult> check() throws IgniteCheckedException {
+ Map<KeyCacheObject, EntryGetResult> resMap = new HashMap<>(keys.size());
+ Set<KeyCacheObject> inconsistentKeys = new HashSet<>();
+
+ for (GridPartitionedGetFuture<KeyCacheObject, EntryGetResult> fut : futs.values()) {
+ for (KeyCacheObject key : fut.keys()) {
+ EntryGetResult curRes = fut.result().get(key);
+
+ if (!resMap.containsKey(key)) {
+ resMap.put(key, curRes);
+
+ continue;
+ }
+
+ EntryGetResult prevRes = resMap.get(key);
+
+ if (curRes != null) {
+ if (prevRes == null || prevRes.version().compareTo(curRes.version()) != 0)
+ inconsistentKeys.add(key);
+ else {
+ CacheObject curVal = curRes.value();
+ CacheObject prevVal = prevRes.value();
+
+ byte[] curBytes = curVal.valueBytes(ctx.cacheObjectContext());
+ byte[] prevBytes = prevVal.valueBytes(ctx.cacheObjectContext());
+
+ if (!Arrays.equals(curBytes, prevBytes))
+ inconsistentKeys.add(key);
+
+ }
+ }
+ else if (prevRes != null)
+ inconsistentKeys.add(key);
+ }
+ }
+
+ if (!inconsistentKeys.isEmpty())
+ throw new IgniteConsistencyViolationException(inconsistentKeys);
+
+ return resMap;
+ }
+
+ /**
* @param fixedEntries Fixed map.
*/
- protected void recordConsistencyViolation(
- Set<KeyCacheObject> inconsistentKeys,
- Map<KeyCacheObject, EntryGetResult> fixedEntries
+ protected final void recordConsistencyViolation(
+ Collection<KeyCacheObject> inconsistentKeys,
+ Map<KeyCacheObject, EntryGetResult> fixedEntries,
+ ReadRepairStrategy strategy
) {
GridEventStorageManager evtMgr = ctx.gridEvents();
if (!evtMgr.isRecordable(EVT_CONSISTENCY_VIOLATION))
return;
- Map<Object, Map<ClusterNode, CacheConsistencyViolationEvent.EntryInfo>> originalMap = new HashMap<>();
+ Map<Object, Map<ClusterNode, CacheConsistencyViolationEvent.EntryInfo>> entries = new HashMap<>();
for (Map.Entry<ClusterNode, GridPartitionedGetFuture<KeyCacheObject, EntryGetResult>> pair : futs.entrySet()) {
ClusterNode node = pair.getKey();
GridPartitionedGetFuture<KeyCacheObject, EntryGetResult> fut = pair.getValue();
- for (Map.Entry<KeyCacheObject, EntryGetResult> entry : fut.result().entrySet()) {
- KeyCacheObject key = entry.getKey();
-
+ for (KeyCacheObject key : fut.keys()) {
if (inconsistentKeys.contains(key)) {
- EntryGetResult res = entry.getValue();
- CacheEntryVersion ver = res.version();
+ Map<ClusterNode, CacheConsistencyViolationEvent.EntryInfo> map =
+ entries.computeIfAbsent(
+ ctx.unwrapBinaryIfNeeded(key, !deserializeBinary, false, null), k -> new HashMap<>());
- Object val = ctx.unwrapBinaryIfNeeded(res.value(), !deserializeBinary, false, null);
+ EntryGetResult res = fut.result().get(key);
+ CacheEntryVersion ver = res != null ? res.version() : null;
- Map<ClusterNode, CacheConsistencyViolationEvent.EntryInfo> map =
- originalMap.computeIfAbsent(
- ctx.unwrapBinaryIfNeeded(key, false, false, null), k -> new HashMap<>());
+ Object val = res != null ? ctx.unwrapBinaryIfNeeded(res.value(), !deserializeBinary, false, null) : null;
boolean primary = primaries.get(key).equals(fut.affNode());
- boolean correct = fixedEntries != null && fixedEntries.get(key).equals(res);
+ boolean correct = fixedEntries != null &&
+ ((fixedEntries.get(key) != null && fixedEntries.get(key).equals(res)) ||
+ (fixedEntries.get(key) == null && res == null));
map.put(node, new EventEntryInfo(val, ver, primary, correct));
}
}
}
+ Map<Object, Object> fixed;
+
+ if (fixedEntries == null)
+ fixed = Collections.emptyMap();
+ else {
+ fixed = new HashMap<>();
+
+ for (Map.Entry<KeyCacheObject, EntryGetResult> entry : fixedEntries.entrySet()) {
+ Object key = ctx.unwrapBinaryIfNeeded(entry.getKey(), !deserializeBinary, false, null);
+ Object val = entry.getValue() != null ?
+ ctx.unwrapBinaryIfNeeded(entry.getValue().value(), !deserializeBinary, false, null) : null;
+
+ fixed.put(key, val);
+ }
+ }
+
evtMgr.record(new CacheConsistencyViolationEvent(
ctx.name(),
ctx.discovery().localNode(),
- "Consistency violation fixed.",
- originalMap));
+ "Consistency violation was " + (fixed == null ? "NOT " : "") + "fixed.",
+ entries,
+ fixed,
+ strategy));
}
/**
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/consistency/GridNearReadRepairCheckOnlyFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/consistency/GridNearReadRepairCheckOnlyFuture.java
index 48d7ce1..5f2b57c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/consistency/GridNearReadRepairCheckOnlyFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/consistency/GridNearReadRepairCheckOnlyFuture.java
@@ -17,21 +17,16 @@
package org.apache.ignite.internal.processors.cache.distributed.near.consistency;
-import java.util.Arrays;
import java.util.Collection;
-import java.util.HashMap;
-import java.util.HashSet;
import java.util.Map;
-import java.util.Objects;
import java.util.Set;
import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.cache.ReadRepairStrategy;
import org.apache.ignite.internal.IgniteInternalFuture;
-import org.apache.ignite.internal.processors.cache.CacheObjectAdapter;
import org.apache.ignite.internal.processors.cache.EntryGetResult;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.IgniteCacheExpiryPolicy;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
-import org.apache.ignite.internal.processors.cache.distributed.dht.GridPartitionedGetFuture;
import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
import org.apache.ignite.internal.util.lang.GridClosureException;
import org.apache.ignite.internal.util.typedef.F;
@@ -56,6 +51,7 @@ public class GridNearReadRepairCheckOnlyFuture extends GridNearReadRepairAbstrac
*
* @param ctx Cache context.
* @param keys Keys.
+ * @param strategy Read repair strategy.
* @param readThrough Read-through flag.
* @param taskName Task name.
* @param deserializeBinary Deserialize binary flag.
@@ -69,6 +65,7 @@ public class GridNearReadRepairCheckOnlyFuture extends GridNearReadRepairAbstrac
public GridNearReadRepairCheckOnlyFuture(
GridCacheContext ctx,
Collection<KeyCacheObject> keys,
+ ReadRepairStrategy strategy,
boolean readThrough,
String taskName,
boolean deserializeBinary,
@@ -81,6 +78,7 @@ public class GridNearReadRepairCheckOnlyFuture extends GridNearReadRepairAbstrac
super(null,
ctx,
keys,
+ strategy,
readThrough,
taskName,
deserializeBinary,
@@ -91,68 +89,34 @@ public class GridNearReadRepairCheckOnlyFuture extends GridNearReadRepairAbstrac
this.skipVals = skipVals;
this.needVer = needVer;
this.keepCacheObjects = keepCacheObjects;
+
+ init();
}
/** {@inheritDoc} */
@Override protected void reduce() {
- Map<KeyCacheObject, EntryGetResult> resMap = new HashMap<>(keys.size());
- Set<KeyCacheObject> inconsistentKeys = new HashSet<>();
-
- for (GridPartitionedGetFuture<KeyCacheObject, EntryGetResult> fut : futs.values()) {
- for (KeyCacheObject key : fut.keys()) {
- EntryGetResult curRes = fut.result().get(key);
-
- if (!resMap.containsKey(key)) {
- resMap.put(key, curRes);
-
- continue;
- }
-
- EntryGetResult prevRes = resMap.get(key);
-
- if (curRes != null) {
- if (prevRes == null || prevRes.version().compareTo(curRes.version()) != 0)
- inconsistentKeys.add(key);
- else {
- CacheObjectAdapter curVal = curRes.value();
- CacheObjectAdapter prevVal = prevRes.value();
-
- try {
- byte[] curBytes = curVal.valueBytes(ctx.cacheObjectContext());
- byte[] prevBytes = prevVal.valueBytes(ctx.cacheObjectContext());
-
- if (!Arrays.equals(curBytes, prevBytes))
- inconsistentKeys.add(key);
- }
- catch (IgniteCheckedException e) {
- onDone(e);
-
- return;
- }
- }
- }
- else if (prevRes != null)
- inconsistentKeys.add(key);
- }
+ try {
+ onDone(check());
}
+ catch (IgniteConsistencyViolationException e) {
+ Set<KeyCacheObject> inconsistentKeys = e.keys();
- if (!inconsistentKeys.isEmpty()) {
if (REMAP_CNT_UPD.incrementAndGet(this) > MAX_REMAP_CNT) {
- if (!ctx.transactional()) // Will not be fixed, should be recorded as is.
- recordConsistencyViolation(inconsistentKeys, /*nothing fixed*/ null);
+ if (ctx.atomic() || strategy == ReadRepairStrategy.CHECK_ONLY) { // Will not be fixed, should be recorded as is.
+ recordConsistencyViolation(inconsistentKeys, /*nothing fixed*/ null, ReadRepairStrategy.CHECK_ONLY);
- onDone(new IgniteConsistencyViolationException("Distributed cache consistency violation detected."));
+ onDone(new IgniteIrreparableConsistencyViolationException(null,
+ ctx.unwrapBinariesIfNeeded(inconsistentKeys, !deserializeBinary)));
+ }
+ else // Should be fixed by concurrent tx(s).
+ onDone(e);
}
else
remap(ctx.affinity().affinityTopologyVersion()); // Rechecking possible "false positive" case.
-
- return;
}
-
- // Misses recorded to detect partial misses, but should not be propagated when the key is null at each node.
- resMap.values().removeIf(Objects::isNull);
-
- onDone(resMap);
+ catch (IgniteCheckedException e) {
+ onDone(e);
+ }
}
/**
@@ -165,23 +129,7 @@ public class GridNearReadRepairCheckOnlyFuture extends GridNearReadRepairAbstrac
try {
final Map<K, V> map = new IgniteBiTuple<>();
- for (Map.Entry<KeyCacheObject, EntryGetResult> entry : fut.get().entrySet()) {
- EntryGetResult getRes = entry.getValue();
-
- ctx.addResult(map,
- entry.getKey(),
- getRes.value(),
- skipVals,
- keepCacheObjects,
- deserializeBinary,
- false,
- getRes,
- getRes.version(),
- 0,
- 0,
- needVer,
- null);
- }
+ addResult(fut, map);
if (skipVals) {
Boolean val = map.isEmpty() ? false : (Boolean)F.firstValue(map);
@@ -207,23 +155,7 @@ public class GridNearReadRepairCheckOnlyFuture extends GridNearReadRepairAbstrac
try {
final Map<K, V> map = U.newHashMap(keys.size());
- for (Map.Entry<KeyCacheObject, EntryGetResult> entry : fut.get().entrySet()) {
- EntryGetResult getRes = entry.getValue();
-
- ctx.addResult(map,
- entry.getKey(),
- getRes.value(),
- skipVals,
- keepCacheObjects,
- deserializeBinary,
- false,
- getRes,
- getRes.version(),
- 0,
- 0,
- needVer,
- null);
- }
+ addResult(fut, map);
return map;
}
@@ -232,4 +164,29 @@ public class GridNearReadRepairCheckOnlyFuture extends GridNearReadRepairAbstrac
}
});
}
+
+ /**
+ *
+ */
+ private <K, V> void addResult(IgniteInternalFuture<Map<KeyCacheObject, EntryGetResult>> fut,
+ Map<K, V> map) throws IgniteCheckedException {
+ for (Map.Entry<KeyCacheObject, EntryGetResult> entry : fut.get().entrySet()) {
+ EntryGetResult getRes = entry.getValue();
+
+ if (getRes != null)
+ ctx.addResult(map,
+ entry.getKey(),
+ getRes.value(),
+ skipVals,
+ keepCacheObjects,
+ deserializeBinary,
+ false,
+ getRes,
+ getRes.version(),
+ 0,
+ 0,
+ needVer,
+ U.deploymentClassLoader(ctx.kernalContext(), U.contextDeploymentClassLoaderId(ctx.kernalContext())));
+ }
+ }
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/consistency/GridNearReadRepairFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/consistency/GridNearReadRepairFuture.java
index 1bb064e..2f6c15b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/consistency/GridNearReadRepairFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/consistency/GridNearReadRepairFuture.java
@@ -19,17 +19,25 @@ package org.apache.ignite.internal.processors.cache.distributed.near.consistency
import java.util.Arrays;
import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.Map;
+import java.util.Set;
import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.cache.ReadRepairStrategy;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
-import org.apache.ignite.internal.processors.cache.CacheObjectAdapter;
+import org.apache.ignite.internal.processors.cache.CacheObject;
import org.apache.ignite.internal.processors.cache.EntryGetResult;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.IgniteCacheExpiryPolicy;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridPartitionedGetFuture;
import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
+import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
+import org.apache.ignite.internal.util.typedef.T2;
+import org.apache.ignite.lang.IgniteBiTuple;
import org.apache.ignite.transactions.TransactionState;
/**
@@ -43,6 +51,7 @@ public class GridNearReadRepairFuture extends GridNearReadRepairAbstractFuture {
* @param topVer Affinity topology version.
* @param ctx Cache context.
* @param keys Keys.
+ * @param strategy Read repair strategy.
* @param readThrough Read-through flag.
* @param taskName Task name.
* @param deserializeBinary Deserialize binary flag.
@@ -54,6 +63,7 @@ public class GridNearReadRepairFuture extends GridNearReadRepairAbstractFuture {
AffinityTopologyVersion topVer,
GridCacheContext ctx,
Collection<KeyCacheObject> keys,
+ ReadRepairStrategy strategy,
boolean readThrough,
String taskName,
boolean deserializeBinary,
@@ -63,6 +73,7 @@ public class GridNearReadRepairFuture extends GridNearReadRepairAbstractFuture {
super(topVer,
ctx,
keys,
+ strategy,
readThrough,
taskName,
deserializeBinary,
@@ -71,18 +82,81 @@ public class GridNearReadRepairFuture extends GridNearReadRepairAbstractFuture {
tx);
assert ctx.transactional() : "Atomic cache should not be recovered using this future";
+
+ init();
}
/** {@inheritDoc} */
@Override protected void reduce() {
- Map<KeyCacheObject, EntryGetResult> newestMap = new HashMap<>(keys.size()); // Newest entries (by version).
- Map<KeyCacheObject, EntryGetResult> fixedMap = new HashMap<>(); // Newest entries required to be re-committed.
+ assert strategy != null;
+
+ try {
+ check();
+
+ onDone(Collections.emptyMap()); // Everything is fine.
+ }
+ catch (IgniteConsistencyViolationException e) { // Inconsistent entries found.
+ try {
+ Map<KeyCacheObject, EntryGetResult> fixedMap; // Entries required to be re-committed.
+
+ if (strategy == ReadRepairStrategy.LWW)
+ fixedMap = fixWithLww(e.keys());
+ else if (strategy == ReadRepairStrategy.PRIMARY)
+ fixedMap = fixWithPrimary(e.keys());
+ else if (strategy == ReadRepairStrategy.RELATIVE_MAJORITY)
+ fixedMap = fixWithMajority(e.keys());
+ else if (strategy == ReadRepairStrategy.REMOVE)
+ fixedMap = fixWithRemove(e.keys());
+ else if (strategy == ReadRepairStrategy.CHECK_ONLY)
+ throw new IgniteIrreparableConsistencyViolationException(null,
+ ctx.unwrapBinariesIfNeeded(e.keys(), !deserializeBinary));
+ else
+ throw new UnsupportedOperationException("Unsupported strategy: " + strategy);
+
+ if (!fixedMap.isEmpty()) {
+ tx.finishFuture().listen(future -> {
+ TransactionState state = tx.state();
+
+ if (state == TransactionState.COMMITTED) // Explicit tx may fix the values but become rolled back later.
+ recordConsistencyViolation(fixedMap.keySet(), fixedMap, strategy);
+ });
+ }
+
+ onDone(fixedMap);
+ }
+ catch (IgniteIrreparableConsistencyViolationException ie) { // Unable to repair all entries.
+ recordConsistencyViolation(e.keys(), /*nothing fixed*/ null, strategy);
+
+ onDone(ie);
+ }
+ catch (IgniteCheckedException ce) {
+ onDone(ce);
+ }
+ }
+ catch (IgniteCheckedException e) {
+ onDone(e);
+ }
+ }
+
+ /**
+ *
+ */
+ public Map<KeyCacheObject, EntryGetResult> fixWithLww(Set<KeyCacheObject> inconsistentKeys) throws IgniteCheckedException {
+ Map<KeyCacheObject, EntryGetResult> newestMap = new HashMap<>(inconsistentKeys.size()); // Newest entries (by version).
+ Map<KeyCacheObject, EntryGetResult> fixedMap = new HashMap<>(inconsistentKeys.size());
+
+ Set<KeyCacheObject> irreparableSet = new HashSet<>();
for (GridPartitionedGetFuture<KeyCacheObject, EntryGetResult> fut : futs.values()) {
- for (KeyCacheObject key : fut.keys()) {
+ for (KeyCacheObject key : inconsistentKeys) {
+ if (!fut.keys().contains(key))
+ continue;
+
EntryGetResult candidateRes = fut.result().get(key);
- if (!newestMap.containsKey(key)) {
+ boolean hasNewest = newestMap.containsKey(key);
+
+ if (!hasNewest) {
newestMap.put(key, candidateRes);
continue;
@@ -91,9 +165,13 @@ public class GridNearReadRepairFuture extends GridNearReadRepairAbstractFuture {
EntryGetResult newestRes = newestMap.get(key);
if (candidateRes != null) {
- if (newestRes == null) { // Existing data wins.
- newestMap.put(key, candidateRes);
- fixedMap.put(key, candidateRes);
+ if (newestRes == null) {
+ if (hasNewest) // Newest is null.
+ irreparableSet.add(key);
+ else { // Existing data wins.
+ newestMap.put(key, candidateRes);
+ fixedMap.put(key, candidateRes);
+ }
}
else {
int compareRes = candidateRes.version().compareTo(newestRes.version());
@@ -105,40 +183,156 @@ public class GridNearReadRepairFuture extends GridNearReadRepairAbstractFuture {
else if (compareRes < 0)
fixedMap.put(key, newestRes);
else if (compareRes == 0) {
- CacheObjectAdapter candidateVal = candidateRes.value();
- CacheObjectAdapter newestVal = newestRes.value();
-
- try {
- byte[] candidateBytes = candidateVal.valueBytes(ctx.cacheObjectContext());
- byte[] newestBytes = newestVal.valueBytes(ctx.cacheObjectContext());
+ CacheObject candidateVal = candidateRes.value();
+ CacheObject newestVal = newestRes.value();
- if (!Arrays.equals(candidateBytes, newestBytes))
- fixedMap.put(key, newestRes); // Same version, fixing values inconsistency.
- }
- catch (IgniteCheckedException e) {
- onDone(e);
+ byte[] candidateBytes = candidateVal.valueBytes(ctx.cacheObjectContext());
+ byte[] newestBytes = newestVal.valueBytes(ctx.cacheObjectContext());
- return;
- }
+ if (!Arrays.equals(candidateBytes, newestBytes))
+ irreparableSet.add(key);
}
}
}
else if (newestRes != null)
- fixedMap.put(key, newestRes); // Existing data wins.
+ irreparableSet.add(key); // Impossible to detect latest between existing and null.
}
}
assert !fixedMap.containsValue(null) : "null should never be considered as a fix";
- if (!fixedMap.isEmpty()) {
- tx.finishFuture().listen(future -> {
- TransactionState state = tx.state();
+ if (!irreparableSet.isEmpty())
+ throwIrreparable(inconsistentKeys, irreparableSet);
+
+ return fixedMap;
+ }
+
+ /**
+ *
+ */
+ public Map<KeyCacheObject, EntryGetResult> fixWithPrimary(Collection<KeyCacheObject> inconsistentKeys) {
+ Map<KeyCacheObject, EntryGetResult> fixedMap = new HashMap<>(inconsistentKeys.size());
+
+ for (GridPartitionedGetFuture<KeyCacheObject, EntryGetResult> fut : futs.values()) {
+ for (KeyCacheObject key : inconsistentKeys) {
+ if (fut.keys().contains(key) && primaries.get(key).equals(fut.affNode()))
+ fixedMap.put(key, fut.result().get(key));
+ }
+ }
+
+ return fixedMap;
+ }
+
+ /**
+ *
+ */
+ public Map<KeyCacheObject, EntryGetResult> fixWithRemove(Collection<KeyCacheObject> inconsistentKeys) {
+ Map<KeyCacheObject, EntryGetResult> fixedMap = new HashMap<>(inconsistentKeys.size());
+
+ for (KeyCacheObject key : inconsistentKeys)
+ fixedMap.put(key, null);
+
+ return fixedMap;
+ }
+
+ /**
+ *
+ */
+ public Map<KeyCacheObject, EntryGetResult> fixWithMajority(Collection<KeyCacheObject> inconsistentKeys)
+ throws IgniteCheckedException {
+ /** */
+ class ByteArrayWrapper {
+ final byte[] arr;
+
+ /** */
+ public ByteArrayWrapper(byte[] arr) {
+ this.arr = arr;
+ }
+
+ /** */
+ @Override public boolean equals(Object o) {
+ return Arrays.equals(arr, ((ByteArrayWrapper)o).arr);
+ }
+
+ /** */
+ @Override public int hashCode() {
+ return Arrays.hashCode(arr);
+ }
+ }
+
+ Set<KeyCacheObject> irreparableSet = new HashSet<>(inconsistentKeys.size());
+ Map<KeyCacheObject, EntryGetResult> fixedMap = new HashMap<>(inconsistentKeys.size());
+
+ for (KeyCacheObject key : inconsistentKeys) {
+ Map<T2<ByteArrayWrapper, GridCacheVersion>, T2<EntryGetResult, Integer>> cntMap = new HashMap<>();
+
+ for (GridPartitionedGetFuture<KeyCacheObject, EntryGetResult> fut : futs.values()) {
+ if (!fut.keys().contains(key))
+ continue;
+
+ EntryGetResult res = fut.result().get(key);
+
+ ByteArrayWrapper wrapped;
+ GridCacheVersion ver;
+
+ if (res != null) {
+ CacheObject val = res.value();
+
+ wrapped = new ByteArrayWrapper(val.valueBytes(ctx.cacheObjectContext()));
+ ver = res.version();
+ }
+ else {
+ wrapped = new ByteArrayWrapper(null);
+ ver = null;
+ }
+
+ T2<ByteArrayWrapper, GridCacheVersion> keyVer = new T2<>(wrapped, ver);
+
+ cntMap.putIfAbsent(keyVer, new T2<>(res, 0));
+
+ cntMap.compute(keyVer, (kv, ri) -> new T2<>(ri.getKey(), ri.getValue() + 1));
+ }
+
+ int[] sorted = cntMap.values().stream()
+ .map(IgniteBiTuple::getValue)
+ .sorted(Comparator.reverseOrder())
+ .mapToInt(v -> v)
+ .toArray();
+
+ int max = sorted[0];
+
+ assert max > 0;
+
+ if (sorted.length > 1 && sorted[1] == max) { // Majority was not found.
+ irreparableSet.add(key);
+
+ continue;
+ }
+
+ for (Map.Entry<T2<ByteArrayWrapper, GridCacheVersion>, T2<EntryGetResult, Integer>> count : cntMap.entrySet())
+ if (count.getValue().getValue().equals(max)) {
+ fixedMap.put(key, count.getValue().getKey());
- if (state == TransactionState.COMMITTED) // Explicit tx may fix the values but become rolled back later.
- recordConsistencyViolation(fixedMap.keySet(), fixedMap);
- });
+ break;
+ }
}
- onDone(fixedMap);
+ if (!irreparableSet.isEmpty())
+ throwIrreparable(inconsistentKeys, irreparableSet);
+
+ return fixedMap;
+ }
+
+ /**
+ *
+ */
+ private void throwIrreparable(Collection<KeyCacheObject> inconsistentKeys, Set<KeyCacheObject> irreparableSet)
+ throws IgniteIrreparableConsistencyViolationException {
+ Set<KeyCacheObject> repairableKeys = new HashSet<>(inconsistentKeys);
+
+ repairableKeys.removeAll(irreparableSet);
+
+ throw new IgniteIrreparableConsistencyViolationException(ctx.unwrapBinariesIfNeeded(repairableKeys, !deserializeBinary),
+ ctx.unwrapBinariesIfNeeded(irreparableSet, !deserializeBinary));
}
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/consistency/IgniteConsistencyViolationException.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/consistency/IgniteConsistencyViolationException.java
index db3d448..7d102d6 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/consistency/IgniteConsistencyViolationException.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/consistency/IgniteConsistencyViolationException.java
@@ -17,20 +17,35 @@
package org.apache.ignite.internal.processors.cache.distributed.near.consistency;
+import java.util.Set;
import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.processors.cache.KeyCacheObject;
/**
- * Possible consistency violation exception.
- * Each such case should be rechecked under locks.
+ * Consistency violation exception.
*/
public class IgniteConsistencyViolationException extends IgniteCheckedException {
/** */
private static final long serialVersionUID = 0L;
+ /** Inconsistent entries keys. */
+ private final Set<KeyCacheObject> keys;
+
+ /**
+ * @param keys Keys.
+ */
+ public IgniteConsistencyViolationException(Set<KeyCacheObject> keys) {
+ super("Distributed cache consistency violation detected.");
+
+ assert keys != null && !keys.isEmpty();
+
+ this.keys = keys;
+ }
+
/**
- * @param msg Message.
+ * Inconsistent entries keys.
*/
- public IgniteConsistencyViolationException(String msg) {
- super(msg);
+ public Set<KeyCacheObject> keys() {
+ return keys;
}
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/consistency/IgniteIrreparableConsistencyViolationException.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/consistency/IgniteIrreparableConsistencyViolationException.java
new file mode 100644
index 0000000..32ee368
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/consistency/IgniteIrreparableConsistencyViolationException.java
@@ -0,0 +1,63 @@
+/*
+ * 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.processors.cache.distributed.near.consistency;
+
+import java.util.Collection;
+import org.apache.ignite.IgniteCheckedException;
+
+/**
+ * Irreparable consistency violation exception.
+ */
+public class IgniteIrreparableConsistencyViolationException extends IgniteCheckedException {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** Repairable keys. */
+ private final Collection<Object> repairableKeys;
+
+ /** Irreparable keys. */
+ private final Collection<Object> irreparableKeys;
+
+ /**
+ * @param repairableKeys Repairable keys.
+ * @param irreparableKeys Irreparable keys.
+ */
+ public IgniteIrreparableConsistencyViolationException(Collection<Object> repairableKeys,
+ Collection<Object> irreparableKeys) {
+ super("Irreparable distributed cache consistency violation detected.");
+
+ assert irreparableKeys != null && !irreparableKeys.isEmpty() : irreparableKeys;
+
+ this.repairableKeys = repairableKeys;
+ this.irreparableKeys = irreparableKeys;
+ }
+
+ /**
+ * Inconsistent keys found but can not be fixed using the specified strategy.
+ */
+ public Collection<Object> irreparableKeys() {
+ return irreparableKeys;
+ }
+
+ /**
+ * Inconsistent keys found but can be fixed using the specified strategy.
+ */
+ public Collection<Object> repairableKeys() {
+ return repairableKeys;
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java
index 06490a8..b088a56 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java
@@ -33,6 +33,7 @@ import javax.cache.processor.EntryProcessor;
import javax.cache.processor.EntryProcessorException;
import javax.cache.processor.EntryProcessorResult;
import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.cache.ReadRepairStrategy;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.CacheEntryPredicate;
@@ -319,7 +320,7 @@ public class GridLocalAtomicCache<K, V> extends GridLocalCache<K, V> {
boolean deserializeBinary,
boolean needVer,
boolean recovery,
- boolean readRepair) throws IgniteCheckedException {
+ ReadRepairStrategy readRepairStrategy) throws IgniteCheckedException {
A.notNull(keys, "keys");
String taskName = ctx.kernalContext().job().currentTaskName();
@@ -341,7 +342,7 @@ public class GridLocalAtomicCache<K, V> extends GridLocalCache<K, V> {
final String taskName,
final boolean deserializeBinary,
boolean recovery,
- boolean readRepair,
+ ReadRepairStrategy readRepairStrategy,
final boolean skipVals,
final boolean needVer
) {
@@ -545,7 +546,7 @@ public class GridLocalAtomicCache<K, V> extends GridLocalCache<K, V> {
taskName,
deserializeBinary,
opCtx != null && opCtx.recovery(),
- false,
+ null,
/*force primary*/false,
expiry,
skipVals,
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/GridCacheQueueAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/GridCacheQueueAdapter.java
index 3a0b09f..a4e761a0e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/GridCacheQueueAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/GridCacheQueueAdapter.java
@@ -441,7 +441,7 @@ public abstract class GridCacheQueueAdapter<T> extends AbstractCollection<T> imp
false,
null,
false,
- false,
+ null,
DFLT_ALLOW_ATOMIC_OPS_IN_TX)
: opCtx.keepBinary();
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/visor/consistency/VisorConsistencyRepairTask.java b/modules/core/src/main/java/org/apache/ignite/internal/visor/consistency/VisorConsistencyRepairTask.java
index 1380bee..0948c30 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/visor/consistency/VisorConsistencyRepairTask.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/visor/consistency/VisorConsistencyRepairTask.java
@@ -25,13 +25,14 @@ import org.apache.ignite.IgniteCache;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.cache.ReadRepairStrategy;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.events.CacheConsistencyViolationEvent;
import org.apache.ignite.internal.processors.cache.CacheGroupContext;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.IgniteInternalCache;
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtLocalPartition;
-import org.apache.ignite.internal.processors.cache.distributed.near.consistency.IgniteConsistencyViolationException;
+import org.apache.ignite.internal.processors.cache.distributed.near.consistency.IgniteIrreparableConsistencyViolationException;
import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
import org.apache.ignite.internal.util.GridConcurrentHashSet;
import org.apache.ignite.internal.util.lang.GridCursor;
@@ -87,6 +88,8 @@ public class VisorConsistencyRepairTask extends AbstractConsistencyTask<VisorCon
/** {@inheritDoc} */
@Override protected String run(VisorConsistencyRepairTaskArg arg) throws IgniteException {
String cacheName = arg.cacheName();
+ ReadRepairStrategy strategy = arg.strategy();
+
int p = arg.part();
int batchSize = 1024;
int statusDelay = 60_000; // Every minute.
@@ -111,7 +114,8 @@ public class VisorConsistencyRepairTask extends AbstractConsistencyTask<VisorCon
if (part == null)
return null; // Partition does not belong to the node.
- log.info("Consistency check started [grp=" + grpCtx.cacheOrGroupName() + ", part=" + p + "]");
+ log.info("Consistency check started " +
+ "[grp=" + grpCtx.cacheOrGroupName() + ", part=" + p + ", strategy=" + strategy + "]");
VisorConsistencyStatusTask.MAP.put(arg, "0/" + part.fullSize());
@@ -130,7 +134,7 @@ public class VisorConsistencyRepairTask extends AbstractConsistencyTask<VisorCon
GridCursor<? extends CacheDataRow> cursor = grpCtx.offheap().dataStore(part).cursor(cctx.cacheId());
- IgniteCache<Object, Object> cache = ignite.cache(cacheName).withKeepBinary().withReadRepair();
+ IgniteCache<Object, Object> cache = ignite.cache(cacheName).withKeepBinary().withReadRepair(strategy);
do {
keys.clear();
@@ -152,7 +156,7 @@ public class VisorConsistencyRepairTask extends AbstractConsistencyTask<VisorCon
cache.getAll(keys); // Repair.
}
catch (CacheException e) {
- if (!(e.getCause() instanceof IgniteConsistencyViolationException) // Found but not fixed.
+ if (!(e.getCause() instanceof IgniteIrreparableConsistencyViolationException) // Found but not fixed.
&& !isCancelled())
throw new IgniteException("Read repair attempt failed.", e);
}
@@ -200,7 +204,7 @@ public class VisorConsistencyRepairTask extends AbstractConsistencyTask<VisorCon
StringBuilder sb = new StringBuilder();
for (CacheConsistencyViolationEvent evt : evts) {
- for (Map.Entry<Object, Map<ClusterNode, CacheConsistencyViolationEvent.EntryInfo>> entry : evt.getEntries().entrySet()) {
+ for (Map.Entry<?, Map<ClusterNode, CacheConsistencyViolationEvent.EntryInfo>> entry : evt.getEntries().entrySet()) {
Object key = entry.getKey();
if (cctx.affinity().partition(key) != part)
@@ -209,7 +213,11 @@ public class VisorConsistencyRepairTask extends AbstractConsistencyTask<VisorCon
found++;
sb.append("Key: ").append(key)
- .append(" (Cache: ").append(evt.getCacheName()).append(")").append("\n");
+ .append(" (cache: ").append(evt.getCacheName())
+ .append(", strategy: ").append(evt.getStrategy()).append(")").append("\n");
+
+ if (evt.getFixedEntries().containsKey(key))
+ sb.append(" Fixed: ").append(evt.getFixedEntries().get(key)).append("\n");
for (Map.Entry<ClusterNode, CacheConsistencyViolationEvent.EntryInfo> mapping : entry.getValue().entrySet()) {
ClusterNode node = mapping.getKey();
@@ -217,14 +225,19 @@ public class VisorConsistencyRepairTask extends AbstractConsistencyTask<VisorCon
sb.append(" Node: ").append(node).append("\n")
.append(" Value: ").append(info.getValue()).append("\n")
- .append(" Version: ").append(info.getVersion()).append("\n")
- .append(" Other cluster version: ").append(info.getVersion().otherClusterVersion()).append("\n")
- .append(" On primary: ").append(info.isPrimary()).append("\n")
- .append(" Considered as a correct value: ").append(info.isCorrect()).append("\n");
+ .append(" Version: ").append(info.getVersion()).append("\n");
+
+ if (info.getVersion() != null)
+ sb.append(" Other cluster version: ").append(info.getVersion().otherClusterVersion()).append("\n");
+
+ sb.append(" On primary: ").append(info.isPrimary()).append("\n");
if (info.isCorrect())
- fixed++;
+ sb.append(" Considered as a CORRECT value!").append("\n");
}
+
+ if (evt.getFixedEntries().containsKey(key))
+ fixed++;
}
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/visor/consistency/VisorConsistencyRepairTaskArg.java b/modules/core/src/main/java/org/apache/ignite/internal/visor/consistency/VisorConsistencyRepairTaskArg.java
index 134cda2..50c8b3f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/visor/consistency/VisorConsistencyRepairTaskArg.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/visor/consistency/VisorConsistencyRepairTaskArg.java
@@ -21,6 +21,7 @@ import java.io.IOException;
import java.io.ObjectInput;
import java.io.ObjectOutput;
import java.util.Objects;
+import org.apache.ignite.cache.ReadRepairStrategy;
import org.apache.ignite.internal.dto.IgniteDataTransferObject;
import org.apache.ignite.internal.util.typedef.internal.U;
@@ -37,6 +38,9 @@ public class VisorConsistencyRepairTaskArg extends IgniteDataTransferObject {
/** Partition. */
private int part;
+ /** Strategy. */
+ ReadRepairStrategy strategy;
+
/**
* Default constructor.
*/
@@ -47,15 +51,17 @@ public class VisorConsistencyRepairTaskArg extends IgniteDataTransferObject {
* @param cacheName Cache name.
* @param part Part.
*/
- public VisorConsistencyRepairTaskArg(String cacheName, int part) {
+ public VisorConsistencyRepairTaskArg(String cacheName, int part, ReadRepairStrategy strategy) {
this.cacheName = cacheName;
this.part = part;
+ this.strategy = strategy;
}
/** {@inheritDoc} */
@Override protected void writeExternalData(ObjectOutput out) throws IOException {
U.writeString(out, cacheName);
out.writeInt(part);
+ U.writeEnum(out, strategy);
}
/** {@inheritDoc} */
@@ -63,6 +69,7 @@ public class VisorConsistencyRepairTaskArg extends IgniteDataTransferObject {
ObjectInput in) throws IOException, ClassNotFoundException {
cacheName = U.readString(in);
part = in.readInt();
+ strategy = U.readEnum(in, ReadRepairStrategy.class);
}
/**
@@ -79,6 +86,13 @@ public class VisorConsistencyRepairTaskArg extends IgniteDataTransferObject {
return part;
}
+ /**
+ *
+ */
+ public ReadRepairStrategy strategy() {
+ return strategy;
+ }
+
/** {@inheritDoc} */
@Override public boolean equals(Object o) {
if (this == o)
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/AbstractFullSetReadRepairTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/AbstractFullSetReadRepairTest.java
index 01c0446..9897c30 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/AbstractFullSetReadRepairTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/AbstractFullSetReadRepairTest.java
@@ -18,12 +18,18 @@
package org.apache.ignite.internal.processors.cache.consistency;
import java.util.Collection;
+import java.util.Collections;
import java.util.Map;
import java.util.Set;
+import java.util.TreeSet;
+import java.util.function.BiConsumer;
import java.util.function.Consumer;
+import java.util.stream.Collectors;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.cache.CacheEntry;
+import org.apache.ignite.cache.ReadRepairStrategy;
+import org.apache.ignite.internal.processors.cache.distributed.near.consistency.IgniteIrreparableConsistencyViolationException;
import org.apache.ignite.internal.util.typedef.G;
import org.junit.Test;
@@ -39,24 +45,29 @@ public abstract class AbstractFullSetReadRepairTest extends AbstractReadRepairTe
Set<Integer> keys = data.data.keySet();
boolean raw = data.raw;
boolean async = data.async;
+ ReadRepairStrategy strategy = data.strategy;
assert keys.size() == 1;
for (Map.Entry<Integer, InconsistentMapping> entry : data.data.entrySet()) { // Once.
Integer key = entry.getKey();
- Integer latest = entry.getValue().latest;
+ Integer fixed = entry.getValue().fixed;
- Integer res =
- raw ?
- async ?
- cache.withReadRepair().getEntryAsync(key).get().getValue() :
- cache.withReadRepair().getEntry(key).getValue() :
- async ?
- cache.withReadRepair().getAsync(key).get() :
- cache.withReadRepair().get(key);
+ Integer res;
+
+ if (raw) {
+ CacheEntry<Integer, Integer> rawEntry = async ?
+ cache.withReadRepair(strategy).getEntryAsync(key).get() :
+ cache.withReadRepair(strategy).getEntry(key);
- if (latest != null)
- assertEquals(latest, res);
+ res = rawEntry != null ? rawEntry.getValue() : null;
+ }
+ else
+ res = async ?
+ cache.withReadRepair(strategy).getAsync(key).get() :
+ cache.withReadRepair(strategy).get(key);
+
+ assertEquals(fixed, res);
}
};
@@ -68,33 +79,32 @@ public abstract class AbstractFullSetReadRepairTest extends AbstractReadRepairTe
Set<Integer> keys = data.data.keySet();
boolean raw = data.raw;
boolean async = data.async;
+ ReadRepairStrategy strategy = data.strategy;
assert !keys.isEmpty();
if (raw) {
Collection<CacheEntry<Integer, Integer>> res =
async ?
- cache.withReadRepair().getEntriesAsync(keys).get() :
- cache.withReadRepair().getEntries(keys);
+ cache.withReadRepair(strategy).getEntriesAsync(keys).get() :
+ cache.withReadRepair(strategy).getEntries(keys);
for (CacheEntry<Integer, Integer> entry : res) {
- Integer latest = data.data.get(entry.getKey()).latest;
+ Integer fixed = data.data.get(entry.getKey()).fixed;
- if (latest != null)
- assertEquals(latest, entry.getValue());
+ assertEquals(fixed, entry.getValue());
}
}
else {
Map<Integer, Integer> res =
async ?
- cache.withReadRepair().getAllAsync(keys).get() :
- cache.withReadRepair().getAll(keys);
+ cache.withReadRepair(strategy).getAllAsync(keys).get() :
+ cache.withReadRepair(strategy).getAll(keys);
for (Map.Entry<Integer, Integer> entry : res.entrySet()) {
- Integer latest = data.data.get(entry.getKey()).latest;
+ Integer fixed = data.data.get(entry.getKey()).fixed;
- if (latest != null)
- assertEquals(latest, entry.getValue());
+ assertEquals(fixed, entry.getValue());
}
}
};
@@ -107,20 +117,21 @@ public abstract class AbstractFullSetReadRepairTest extends AbstractReadRepairTe
Set<Integer> keys = data.data.keySet();
boolean raw = data.raw;
boolean async = data.async;
+ ReadRepairStrategy strategy = data.strategy;
assert keys.size() == 1;
for (Integer key : data.data.keySet()) { // Once.
- Integer missed = key * -1; // Negative to gain null.
+ Integer missed = -key; // Negative to gain null.
Object res =
raw ?
async ?
- cache.withReadRepair().getEntryAsync(missed).get() :
- cache.withReadRepair().getEntry(missed) :
+ cache.withReadRepair(strategy).getEntryAsync(missed).get() :
+ cache.withReadRepair(strategy).getEntry(missed) :
async ?
- cache.withReadRepair().getAsync(missed).get() :
- cache.withReadRepair().get(missed);
+ cache.withReadRepair(strategy).getAsync(missed).get() :
+ cache.withReadRepair(strategy).get(missed);
assertEquals(null, res);
}
@@ -129,19 +140,53 @@ public abstract class AbstractFullSetReadRepairTest extends AbstractReadRepairTe
/**
*
*/
+ protected static final Consumer<ReadRepairData> GET_ALL_NULL = (data) -> {
+ IgniteCache<Integer, Integer> cache = data.cache;
+ Set<Integer> keys = data.data.keySet();
+ boolean raw = data.raw;
+ boolean async = data.async;
+ ReadRepairStrategy strategy = data.strategy;
+
+ Set<Integer> missed = keys.stream().map(key -> -key).collect(Collectors.toCollection(TreeSet::new)); // Negative to gain null.
+
+ if (raw) {
+ Collection<CacheEntry<Integer, Integer>> res =
+ async ?
+ cache.withReadRepair(strategy).getEntriesAsync(missed).get() :
+ cache.withReadRepair(strategy).getEntries(missed);
+
+ assertTrue(res.isEmpty());
+ }
+ else {
+ Map<Integer, Integer> res =
+ async ?
+ cache.withReadRepair(strategy).getAllAsync(missed).get() :
+ cache.withReadRepair(strategy).getAll(missed);
+
+ assertTrue(res.isEmpty());
+ }
+ };
+
+ /**
+ *
+ */
protected static final Consumer<ReadRepairData> CONTAINS_CHECK_AND_FIX = (data) -> {
IgniteCache<Integer, Integer> cache = data.cache;
Set<Integer> keys = data.data.keySet();
boolean async = data.async;
+ ReadRepairStrategy strategy = data.strategy;
assert keys.size() == 1;
- for (Integer key : data.data.keySet()) { // Once.
+ for (Map.Entry<Integer, InconsistentMapping> entry : data.data.entrySet()) { // Once.
+ Integer key = entry.getKey();
+ Integer fixed = entry.getValue().fixed;
+
boolean res = async ?
- cache.withReadRepair().containsKeyAsync(key).get() :
- cache.withReadRepair().containsKey(key);
+ cache.withReadRepair(strategy).containsKeyAsync(key).get() :
+ cache.withReadRepair(strategy).containsKey(key);
- assertEquals(true, res);
+ assertEquals(fixed != null, res);
}
};
@@ -152,45 +197,103 @@ public abstract class AbstractFullSetReadRepairTest extends AbstractReadRepairTe
IgniteCache<Integer, Integer> cache = data.cache;
Set<Integer> keys = data.data.keySet();
boolean async = data.async;
+ ReadRepairStrategy strategy = data.strategy;
boolean res = async ?
- cache.withReadRepair().containsKeysAsync(keys).get() :
- cache.withReadRepair().containsKeys(keys);
+ cache.withReadRepair(strategy).containsKeysAsync(keys).get() :
+ cache.withReadRepair(strategy).containsKeys(keys);
+
+ boolean allFixed = true;
- assertEquals(true, res);
+ for (Integer key : keys) {
+ Integer fixed = data.data.get(key).fixed;
+
+ if (fixed == null)
+ allFixed = false;
+ }
+
+ assertEquals(allFixed, res);
};
/**
*
*/
- protected static final Consumer<ReadRepairData> CHECK_FIXED = (data) -> {
- IgniteCache<Integer, Integer> cache = data.cache;
- boolean raw = data.raw;
+ protected static final BiConsumer<ReadRepairData, IgniteIrreparableConsistencyViolationException> CHECK_FIXED =
+ (data, e) -> {
+ IgniteCache<Integer, Integer> cache = data.cache;
+ boolean raw = data.raw;
- for (Map.Entry<Integer, InconsistentMapping> entry : data.data.entrySet()) {
- Integer key = entry.getKey();
- Integer latest = entry.getValue().latest;
+ for (Map.Entry<Integer, InconsistentMapping> entry : data.data.entrySet()) {
+ Integer key = entry.getKey();
+
+ // Checking only fixed entries, while entries listed at exception were not fixed.
+ if (e != null && (e.irreparableKeys().contains(key) ||
+ (e.repairableKeys() != null && e.repairableKeys().contains(key))))
+ continue;
+
+ Integer fixed = entry.getValue().fixed;
- Integer res = raw ?
- cache.getEntry(key).getValue() :
- cache.get(key);
+ Integer res;
+
+ if (raw) {
+ CacheEntry<Integer, Integer> rawEntry = cache.getEntry(key);
+
+ res = rawEntry != null ? rawEntry.getValue() : null;
+ }
+ else
+ res = cache.get(key);
+
+ assertEquals(fixed, res);
+ }
+ };
+
+ /**
+ *
+ */
+ protected final BiConsumer<ReadRepairData, Runnable> repairIfRepairable = (data, r) -> {
+ try {
+ r.run();
- if (latest != null)
- assertEquals(latest, res);
+ assertTrue(data.repairable());
+ }
+ catch (Exception e) {
+ Throwable cause = e.getCause();
+
+ assertTrue(cause instanceof IgniteIrreparableConsistencyViolationException);
+ assertFalse(data.repairable());
+
+ check(data, (IgniteIrreparableConsistencyViolationException)cause, true);
}
};
/**
* @param data Data.
*/
- protected void check(ReadRepairData data, boolean checkEvtRecorded, boolean checkFixed) {
- if (checkEvtRecorded)
- checkEvent(data, checkFixed);
+ protected void check(ReadRepairData data, IgniteIrreparableConsistencyViolationException e, boolean evtRecorded) {
+ Collection<?> irreparableKeys = e != null ? e.irreparableKeys() : null;
+
+ if (e != null) {
+ Collection<?> repairableKeys = e.repairableKeys();
+
+ if (repairableKeys != null)
+ assertTrue(Collections.disjoint(repairableKeys, irreparableKeys));
+
+ Collection<?> expectedToBeIrreparableKeys = data.data.entrySet().stream()
+ .filter(entry -> !entry.getValue().repairable)
+ .map(Map.Entry::getKey)
+ .collect(Collectors.toSet());
+
+ assertEquals(expectedToBeIrreparableKeys, irreparableKeys);
+ }
+
+ assertEquals(irreparableKeys == null, data.repairable());
+
+ if (evtRecorded)
+ checkEvent(data, e);
else
checkEventMissed();
- if (checkFixed)
- CHECK_FIXED.accept(data);
+ CHECK_FIXED.accept(data, e);
}
/**
@@ -198,35 +301,23 @@ public abstract class AbstractFullSetReadRepairTest extends AbstractReadRepairTe
*/
@Test
public void test() throws Exception {
- for (Ignite node : G.allGrids()) {
- testGetVariations(node);
- testGetNull(node);
- testContainsVariations(node);
+ for (Ignite initiator : G.allGrids()) {
+ test(initiator, 1, false); // just get
+ test(initiator, 1, true); // 1 (all keys available at primary)
+ test(initiator, 2, true); // less than backups
+ test(initiator, 3, true); // equals to backups
+ test(initiator, 4, true); // equals to backups + primary
+ test(initiator, 10, true); // more than backups + primary
}
}
/**
*
*/
- private void testGetVariations(Ignite initiator) throws Exception {
- testGet(initiator, 1, false); // just get
- testGet(initiator, 1, true); // 1 (all keys available at primary)
- testGet(initiator, 2, true); // less than backups
- testGet(initiator, 3, true); // equals to backups
- testGet(initiator, 4, true); // equals to backups + primary
- testGet(initiator, 10, true); // more than backups
- }
-
- /**
- *
- */
- private void testContainsVariations(Ignite initiator) throws Exception {
- testContains(initiator, 1, false); // just contains
- testContains(initiator, 1, true); // 1 (all keys available at primary)
- testContains(initiator, 2, true); // less than backups
- testContains(initiator, 3, true); // equals to backups
- testContains(initiator, 4, true); // equals to backups + primary
- testContains(initiator, 10, true); // more than backups
+ private void test(Ignite initiator, Integer cnt, boolean all) throws Exception {
+ testGet(initiator, cnt, all);
+ testGetNull(initiator, cnt, all);
+ testContains(initiator, cnt, all);
}
/**
@@ -237,7 +328,7 @@ public abstract class AbstractFullSetReadRepairTest extends AbstractReadRepairTe
/**
*
*/
- protected abstract void testGetNull(Ignite initiator) throws Exception;
+ protected abstract void testGetNull(Ignite initiator, Integer cnt, boolean all) throws Exception;
/**
*
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/AbstractReadRepairTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/AbstractReadRepairTest.java
index 7ded65a..72358b4 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/AbstractReadRepairTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/AbstractReadRepairTest.java
@@ -18,26 +18,33 @@
package org.apache.ignite.internal.processors.cache.consistency;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
+import java.util.Comparator;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.ThreadLocalRandom;
import java.util.function.Consumer;
+import java.util.stream.Collectors;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.IgniteEvents;
import org.apache.ignite.cache.CacheAtomicityMode;
import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.cache.ReadRepairStrategy;
+import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.cluster.ClusterState;
import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.DataStorageConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.events.CacheConsistencyViolationEvent;
import org.apache.ignite.events.Event;
import org.apache.ignite.events.EventType;
+import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.CacheObjectImpl;
@@ -45,18 +52,20 @@ import org.apache.ignite.internal.processors.cache.GridCacheAdapter;
import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
import org.apache.ignite.internal.processors.cache.GridCacheOperation;
import org.apache.ignite.internal.processors.cache.IgniteInternalCache;
+import org.apache.ignite.internal.processors.cache.distributed.near.consistency.IgniteIrreparableConsistencyViolationException;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersionManager;
import org.apache.ignite.internal.processors.dr.GridDrType;
import org.apache.ignite.internal.util.typedef.G;
+import org.apache.ignite.internal.util.typedef.T2;
import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.lang.IgnitePredicate;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
-import org.junit.Before;
import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC;
import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
import static org.apache.ignite.cache.CacheMode.PARTITIONED;
+import static org.apache.ignite.cache.CacheMode.REPLICATED;
import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
import static org.apache.ignite.events.EventType.EVT_CONSISTENCY_VIOLATION;
@@ -81,17 +90,28 @@ public abstract class AbstractReadRepairTest extends GridCommonAbstractTest {
}
/** Atomicy mode. */
- protected CacheAtomicityMode atomicyMode() {
+ protected CacheAtomicityMode atomicityMode() {
return TRANSACTIONAL;
}
+ /** Persistence enabled. */
+ protected boolean persistenceEnabled() {
+ return false;
+ }
+
+ /** Server nodes count. */
+ private int serverNodesCount() {
+ return backupsCount() + 1/*primary*/ + 1/*not an owner*/;
+ }
+
/** {@inheritDoc} */
@Override protected void beforeTestsStarted() throws Exception {
super.beforeTestsStarted();
- Ignite ignite = startGrids(7); // Server nodes.
+ if (persistenceEnabled())
+ cleanPersistenceDir();
- grid(0).getOrCreateCache(cacheConfiguration());
+ Ignite ignite = startGrids(serverNodesCount()); // Server nodes.
startClientGrid(G.allGrids().size() + 1); // Client node 1.
startClientGrid(G.allGrids().size() + 1); // Client node 2.
@@ -108,15 +128,12 @@ public abstract class AbstractReadRepairTest extends GridCommonAbstractTest {
},
EVT_CONSISTENCY_VIOLATION);
- awaitPartitionMapExchange();
- }
+ if (persistenceEnabled())
+ ignite.cluster().state(ClusterState.ACTIVE);
- /**
- *
- */
- @Before
- public void before() {
- evtDeq.clear();
+ ignite.getOrCreateCache(cacheConfiguration());
+
+ awaitPartitionMapExchange();
}
/** {@inheritDoc} */
@@ -126,6 +143,9 @@ public abstract class AbstractReadRepairTest extends GridCommonAbstractTest {
log.info("Checked " + iterableKey + " keys");
stopAllGrids();
+
+ if (persistenceEnabled())
+ cleanPersistenceDir();
}
/**
@@ -136,9 +156,11 @@ public abstract class AbstractReadRepairTest extends GridCommonAbstractTest {
cfg.setWriteSynchronizationMode(FULL_SYNC);
cfg.setCacheMode(cacheMode());
- cfg.setAtomicityMode(atomicyMode());
+ cfg.setAtomicityMode(atomicityMode());
cfg.setBackups(backupsCount());
+ cfg.setAffinity(new RendezvousAffinityFunction().setPartitions(32));
+
return cfg;
}
@@ -148,6 +170,12 @@ public abstract class AbstractReadRepairTest extends GridCommonAbstractTest {
cfg.setIncludeEventTypes(EventType.EVTS_ALL);
+ if (persistenceEnabled()) {
+ cfg.setDataStorageConfiguration(new DataStorageConfiguration());
+ cfg.getDataStorageConfiguration().getDefaultDataRegionConfiguration().setMaxSize(100 * 1024 * 1024);
+ cfg.getDataStorageConfiguration().getDefaultDataRegionConfiguration().setPersistenceEnabled(true);
+ }
+
return cfg;
}
@@ -161,49 +189,65 @@ public abstract class AbstractReadRepairTest extends GridCommonAbstractTest {
/**
*
*/
- protected void checkEvent(ReadRepairData data, boolean checkFixed) {
- Map<Object, Map<ClusterNode, CacheConsistencyViolationEvent.EntryInfo>> entries = new HashMap<>();
+ protected void checkEvent(ReadRepairData data, IgniteIrreparableConsistencyViolationException e) {
+ Map<Object, Map<ClusterNode, CacheConsistencyViolationEvent.EntryInfo>> evtEntries = new HashMap<>();
+ Map<Object, Object> evtFixed = new HashMap<>();
- while (!entries.keySet().equals(data.data.keySet())) {
+ Map<Integer, InconsistentMapping> inconsistent = data.data.entrySet().stream()
+ .filter(entry -> !entry.getValue().consistent)
+ .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+
+ while (!evtEntries.keySet().equals(inconsistent.keySet())) {
if (!evtDeq.isEmpty()) {
CacheConsistencyViolationEvent evt = evtDeq.remove();
- entries.putAll(evt.getEntries()); // Optimistic and read committed transactions produce per key fixes.
+ assertEquals(atomicityMode() == TRANSACTIONAL ? data.strategy : ReadRepairStrategy.CHECK_ONLY, evt.getStrategy());
+
+ // Optimistic and read committed transactions produce per key fixes.
+ evtEntries.putAll(evt.getEntries());
+ evtFixed.putAll(evt.getFixedEntries());
}
}
- int fixedCnt = 0;
-
- for (Map.Entry<Integer, InconsistentMapping> mapping : data.data.entrySet()) {
+ for (Map.Entry<Integer, InconsistentMapping> mapping : inconsistent.entrySet()) {
Integer key = mapping.getKey();
- Integer latest = mapping.getValue().latest;
+ Integer fixed = mapping.getValue().fixed;
+ Integer primary = mapping.getValue().primary;
+ boolean repairable = mapping.getValue().repairable;
- Object fixedVal = null;
+ if (!repairable)
+ assertNotNull(e);
- Map<ClusterNode, CacheConsistencyViolationEvent.EntryInfo> values = entries.get(key);
-
- if (values != null)
- for (Map.Entry<ClusterNode, CacheConsistencyViolationEvent.EntryInfo> infoEntry : values.entrySet()) {
- ClusterNode node = infoEntry.getKey();
- CacheConsistencyViolationEvent.EntryInfo info = infoEntry.getValue();
+ if (e == null) {
+ assertTrue(repairable);
+ assertTrue(evtFixed.containsKey(key));
+ assertEquals(fixed, evtFixed.get(key));
+ }
+ // Repairable but not repaired (because of irreparable entry at the same tx) entries.
+ else if (e.irreparableKeys().contains(key) || (e.repairableKeys() != null && e.repairableKeys().contains(key)))
+ assertFalse(evtFixed.containsKey(key));
- if (info.isCorrect()) {
- assertNull(fixedVal);
+ Map<ClusterNode, CacheConsistencyViolationEvent.EntryInfo> evtEntryInfos = evtEntries.get(key);
- fixedVal = info.getValue();
+ if (evtEntryInfos != null)
+ for (Map.Entry<ClusterNode, CacheConsistencyViolationEvent.EntryInfo> evtEntryInfo : evtEntryInfos.entrySet()) {
+ ClusterNode node = evtEntryInfo.getKey();
+ CacheConsistencyViolationEvent.EntryInfo info = evtEntryInfo.getValue();
- fixedCnt++;
- }
+ if (info.isCorrect())
+ assertEquals(fixed, info.getValue());
- if (info.isPrimary())
+ if (info.isPrimary()) {
+ assertEquals(primary, info.getValue());
assertEquals(node, primaryNode(key, DEFAULT_CACHE_NAME).cluster().localNode());
+ }
}
-
- if (latest != null)
- assertEquals(checkFixed ? latest : null, fixedVal);
}
- assertEquals(checkFixed ? data.data.size() : 0, fixedCnt);
+ int expectedFixedCnt = inconsistent.size() -
+ (e != null ? (e.repairableKeys() != null ? e.repairableKeys().size() : 0) + e.irreparableKeys().size() : 0);
+
+ assertEquals(expectedFixedCnt, evtFixed.size());
assertTrue(evtDeq.isEmpty());
}
@@ -222,42 +266,87 @@ public abstract class AbstractReadRepairTest extends GridCommonAbstractTest {
throws Exception {
IgniteCache<Integer, Integer> cache = initiator.getOrCreateCache(DEFAULT_CACHE_NAME);
- for (int i = 0; i < ThreadLocalRandom.current().nextInt(1, 10); i++) {
+ ThreadLocalRandom rnd = ThreadLocalRandom.current();
+
+ for (int i = 0; i < rnd.nextInt(1, 10); i++) {
+ ReadRepairStrategy[] strategies = ReadRepairStrategy.values();
+
+ ReadRepairStrategy strategy = strategies[rnd.nextInt(strategies.length)];
+
Map<Integer, InconsistentMapping> results = new TreeMap<>(); // Sorted to avoid warning.
- for (int j = 0; j < cnt; j++) {
- InconsistentMapping res = setDifferentValuesForSameKey(++iterableKey, misses, nulls);
+ try {
+ for (int j = 0; j < cnt; j++) {
+ InconsistentMapping res = setDifferentValuesForSameKey(++iterableKey, misses, nulls, strategy);
- results.put(iterableKey, res);
- }
+ results.put(iterableKey, res);
+ }
+
+ for (Ignite node : G.allGrids()) { // Check that cache filled properly.
+ Map<Integer, Integer> all =
+ node.<Integer, Integer>getOrCreateCache(DEFAULT_CACHE_NAME).getAll(results.keySet());
- for (Ignite node : G.allGrids()) { // Check that cache filled properly.
- Map<Integer, Integer> all =
- node.<Integer, Integer>getOrCreateCache(DEFAULT_CACHE_NAME).getAll(results.keySet());
+ for (Map.Entry<Integer, Integer> entry : all.entrySet()) {
+ Integer key = entry.getKey();
+ Integer val = entry.getValue();
- for (Map.Entry<Integer, Integer> entry : all.entrySet()) {
- Integer key = entry.getKey();
- Integer val = entry.getValue();
+ T2<Integer, GridCacheVersion> valVer = results.get(key).mapping.get(node);
- Integer exp = results.get(key).mapping.get(node); // Should read from itself (backup or primary).
+ Integer exp;
- if (exp == null)
- exp = results.get(key).primary; // Should read from primary (not a partition owner).
+ if (valVer != null)
+ exp = valVer.get1(); // Should read from itself (backup or primary).
+ else
+ exp = results.get(key).primary; // Or read from primary (when not a partition owner).
- assertEquals(exp, val);
+ assertEquals(exp, val);
+ }
}
+
+ c.accept(new ReadRepairData(cache, results, raw, async, strategy));
}
+ catch (Throwable th) {
+ StringBuilder sb = new StringBuilder();
+
+ sb.append("Read Repair test failed [")
+ .append("cache=").append(cache.getName())
+ .append(", strategy=").append(strategy)
+ .append("]\n");
+
+ for (Map.Entry<Integer, InconsistentMapping> entry : results.entrySet()) {
+ sb.append("Key: ").append(entry.getKey()).append("\n");
- c.accept(new ReadRepairData(cache, results, raw, async));
+ InconsistentMapping mapping = entry.getValue();
+
+ sb.append(" Random data [primary=").append(mapping.primary)
+ .append(", fixed=").append(mapping.fixed)
+ .append(", repairable=").append(mapping.repairable)
+ .append(", consistent=").append(mapping.consistent)
+ .append("]\n");
+
+ sb.append(" Distribution: \n");
+
+ for (Map.Entry<Ignite, T2<Integer, GridCacheVersion>> dist : mapping.mapping.entrySet()) {
+ sb.append(" Node: ").append(dist.getKey().name()).append("\n");
+ sb.append(" Value: ").append(dist.getValue().get1()).append("\n");
+ sb.append(" Version: ").append(dist.getValue().get2()).append("\n");
+ }
+
+ sb.append("\n");
+ }
+
+ throw new Exception(sb.toString(), th);
+ }
}
}
/**
*
*/
- private InconsistentMapping setDifferentValuesForSameKey(int key, boolean misses, boolean nulls) throws Exception {
+ private InconsistentMapping setDifferentValuesForSameKey(int key, boolean misses, boolean nulls,
+ ReadRepairStrategy strategy) throws Exception {
List<Ignite> nodes = new ArrayList<>();
- Map<Ignite, Integer> mapping = new HashMap<>();
+ Map<Ignite, T2<Integer, GridCacheVersion>> mapping = new HashMap<>();
Ignite primary = primaryNode(key, DEFAULT_CACHE_NAME);
@@ -275,116 +364,223 @@ public abstract class AbstractReadRepairTest extends GridCommonAbstractTest {
if (rnd.nextBoolean()) // Random order.
Collections.shuffle(nodes);
- GridCacheVersionManager mgr =
- ((GridCacheAdapter)(grid(1)).cachex(DEFAULT_CACHE_NAME).cache()).context().shared().versions();
+ IgniteInternalCache<Integer, Integer> internalCache = (grid(1)).cachex(DEFAULT_CACHE_NAME);
+
+ GridCacheVersionManager mgr = ((GridCacheAdapter)internalCache.cache()).context().shared().versions();
int incVal = 0;
Integer primVal = null;
+ Collection<T2<Integer, GridCacheVersion>> vals = new ArrayList<>();
- if (misses)
- nodes = nodes.subList(0, rnd.nextInt(1, nodes.size()));
+ if (misses) {
+ List<Ignite> keeped = nodes.subList(0, rnd.nextInt(1, nodes.size()));
- int rmvLimit = rnd.nextInt(nodes.size());
+ nodes.stream()
+ .filter(node -> !keeped.contains(node))
+ .forEach(node -> {
+ T2<Integer, GridCacheVersion> nullT2 = new T2<>(null, null);
+
+ vals.add(nullT2);
+ mapping.put(node, nullT2);
+ }); // Recording nulls (missed values).
+
+ nodes = keeped;
+ }
+
+ boolean rmvd = false;
boolean incVer = rnd.nextBoolean();
GridCacheVersion ver = null;
for (Ignite node : nodes) {
- IgniteInternalCache cache = ((IgniteEx)node).cachex(DEFAULT_CACHE_NAME);
+ IgniteInternalCache<Integer, Integer> cache = ((IgniteEx)node).cachex(DEFAULT_CACHE_NAME);
- GridCacheAdapter adapter = ((GridCacheAdapter)cache.cache());
+ GridCacheAdapter<Integer, Integer> adapter = (GridCacheAdapter)cache.cache();
GridCacheEntryEx entry = adapter.entryEx(key);
if (ver == null || incVer)
ver = mgr.next(entry.context().kernalContext().discovery().topologyVersion()); // Incremental version.
- boolean rmv = nulls && rnd.nextBoolean() && --rmvLimit >= 0;
-
- Integer val = rmv ? null : ++incVal;
-
- boolean init = entry.initialValue(
- new CacheObjectImpl(rmv ? -1 : val, null), // Incremental value.
- ver,
- 0,
- 0,
- false,
- AffinityTopologyVersion.NONE,
- GridDrType.DR_NONE,
- false,
- false);
-
- if (rmv) {
- if (cache.configuration().getAtomicityMode() == ATOMIC)
- entry.innerUpdate(
- ver,
- ((IgniteEx)node).localNode().id(),
- ((IgniteEx)node).localNode().id(),
- GridCacheOperation.DELETE,
- null,
- null,
- false,
- false,
- false,
- false,
- null,
- false,
- false,
- false,
- false,
- AffinityTopologyVersion.NONE,
- null,
- GridDrType.DR_NONE,
- 0,
- 0,
- null,
- false,
- false,
- null,
- null,
- null,
- null,
- false);
+ boolean rmv = nulls && (!rmvd || rnd.nextBoolean());
+
+ Integer val = rmv ? null : rnd.nextBoolean()/*increment or same as previously*/ ? ++incVal : incVal;
+
+ T2<Integer, GridCacheVersion> valVer = new T2<>(val, val != null ? ver : null);
+
+ vals.add(valVer);
+ mapping.put(node, valVer);
+
+ GridKernalContext kctx = ((IgniteEx)node).context();
+
+ byte[] bytes = kctx.cacheObjects().marshal(entry.context().cacheObjectContext(), rmv ? -1 : val); // Incremental value.
+
+ try {
+ kctx.cache().context().database().checkpointReadLock();
+
+ boolean init = entry.initialValue(
+ new CacheObjectImpl(null, bytes),
+ ver,
+ 0,
+ 0,
+ false,
+ AffinityTopologyVersion.NONE,
+ GridDrType.DR_NONE,
+ false,
+ false);
+
+ if (rmv) {
+ if (cache.configuration().getAtomicityMode() == ATOMIC)
+ entry.innerUpdate(
+ ver,
+ ((IgniteEx)node).localNode().id(),
+ ((IgniteEx)node).localNode().id(),
+ GridCacheOperation.DELETE,
+ null,
+ null,
+ false,
+ false,
+ false,
+ false,
+ null,
+ false,
+ false,
+ false,
+ false,
+ AffinityTopologyVersion.NONE,
+ null,
+ GridDrType.DR_NONE,
+ 0,
+ 0,
+ null,
+ false,
+ false,
+ null,
+ null,
+ null,
+ null,
+ false);
+ else
+ entry.innerRemove(
+ null,
+ ((IgniteEx)node).localNode().id(),
+ ((IgniteEx)node).localNode().id(),
+ false,
+ false,
+ false,
+ false,
+ false,
+ null,
+ AffinityTopologyVersion.NONE,
+ CU.empty0(),
+ GridDrType.DR_NONE,
+ null,
+ null,
+ null,
+ 1L);
+
+ rmvd = true;
+
+ assertFalse(entry.hasValue());
+ }
else
- entry.innerRemove(
- null,
- ((IgniteEx)node).localNode().id(),
- ((IgniteEx)node).localNode().id(),
- false,
- false,
- false,
- false,
- false,
- null,
- AffinityTopologyVersion.NONE,
- CU.empty0(),
- GridDrType.DR_NONE,
- null,
- null,
- null,
- 1L);
-
- assertFalse(entry.hasValue());
- }
- else
- assertTrue(entry.hasValue());
-
- assertTrue("iterableKey " + key + " already inited", init);
+ assertTrue(entry.hasValue());
- mapping.put(node, val);
+ assertTrue("iterableKey " + key + " already inited", init);
- if (node.equals(primary))
- primVal = val;
+ if (node.equals(primary))
+ primVal = val;
+ }
+ finally {
+ ((IgniteEx)node).context().cache().context().database().checkpointReadUnlock();
+ }
}
- if (!nulls)
- assertEquals(nodes.size(), new HashSet<>(mapping.values()).size()); // Each node have unique value.
+ assertEquals(vals.size(), mapping.size());
+ assertEquals(vals.size(),
+ internalCache.configuration().getCacheMode() == REPLICATED ? serverNodesCount() : backupsCount() + 1);
if (!misses && !nulls)
assertTrue(primVal != null); // Primary value set.
- return new InconsistentMapping(mapping, primVal, incVer ? incVal : null /*Any*/);
+ Integer fixed;
+
+ boolean consistent;
+ boolean repairable;
+
+ if (vals.stream().distinct().count() == 1) { // Consistent value.
+ consistent = true;
+ repairable = true;
+ fixed = vals.iterator().next().getKey();
+ }
+ else {
+ consistent = false;
+ repairable = atomicityMode() != ATOMIC; // Currently, Atomic caches can not be repaired.
+
+ switch (strategy) {
+ case LWW:
+ if (misses || rmvd || !incVer) {
+ repairable = false;
+
+ fixed = Integer.MIN_VALUE; // Should never be returned.
+ }
+ else
+ fixed = incVal;
+
+ break;
+
+ case PRIMARY:
+ fixed = primVal;
+
+ break;
+
+ case RELATIVE_MAJORITY:
+ fixed = Integer.MIN_VALUE; // Should never be returned.
+
+ Map<T2<Integer, GridCacheVersion>, Integer> counts = new HashMap<>();
+
+ for (T2<Integer, GridCacheVersion> val : vals) {
+ counts.putIfAbsent(val, 0);
+
+ counts.compute(val, (k, v) -> v + 1);
+ }
+
+ int[] sorted = counts.values().stream().sorted(Comparator.reverseOrder()).mapToInt(v -> v).toArray();
+
+ int max = sorted[0];
+
+ if (sorted.length > 1 && sorted[1] == max)
+ repairable = false;
+
+ if (repairable)
+ for (Map.Entry<T2<Integer, GridCacheVersion>, Integer> count : counts.entrySet())
+ if (count.getValue().equals(max)) {
+ fixed = count.getKey().getKey();
+
+ break;
+ }
+
+ break;
+
+ case REMOVE:
+ fixed = null;
+
+ break;
+
+ case CHECK_ONLY:
+ repairable = false;
+
+ fixed = Integer.MIN_VALUE; // Should never be returned.
+
+ break;
+
+ default:
+ throw new UnsupportedOperationException(strategy.toString());
+ }
+ }
+
+ return new InconsistentMapping(mapping, primVal, fixed, repairable, consistent);
}
/**
@@ -392,16 +588,19 @@ public abstract class AbstractReadRepairTest extends GridCommonAbstractTest {
*/
protected static final class ReadRepairData {
/** Initiator's cache. */
- IgniteCache<Integer, Integer> cache;
+ final IgniteCache<Integer, Integer> cache;
/** Generated data across topology per key mapping. */
- Map<Integer, InconsistentMapping> data;
+ final Map<Integer, InconsistentMapping> data;
/** Raw read flag. True means required GetEntry() instead of get(). */
- boolean raw;
+ final boolean raw;
/** Async read flag. */
- boolean async;
+ final boolean async;
+
+ /** Strategy. */
+ final ReadRepairStrategy strategy;
/**
*
@@ -410,11 +609,20 @@ public abstract class AbstractReadRepairTest extends GridCommonAbstractTest {
IgniteCache<Integer, Integer> cache,
Map<Integer, InconsistentMapping> data,
boolean raw,
- boolean async) {
+ boolean async,
+ ReadRepairStrategy strategy) {
this.cache = cache;
this.data = data;
this.raw = raw;
this.async = async;
+ this.strategy = strategy;
+ }
+
+ /**
+ *
+ */
+ boolean repairable() {
+ return data.values().stream().allMatch(mapping -> mapping.repairable);
}
}
@@ -423,21 +631,30 @@ public abstract class AbstractReadRepairTest extends GridCommonAbstractTest {
*/
protected static final class InconsistentMapping {
/** Value per node. */
- Map<Ignite, Integer> mapping;
+ final Map<Ignite, T2<Integer, GridCacheVersion>> mapping;
/** Primary node's value. */
- Integer primary;
+ final Integer primary;
+
+ /** Expected fix result. */
+ final Integer fixed;
+
+ /** Inconsistency can be repaired using the specified strategy. */
+ final boolean repairable;
- /** Latest value across the topology. */
- Integer latest;
+ /** Consistent value. */
+ final boolean consistent;
/**
*
*/
- public InconsistentMapping(Map<Ignite, Integer> mapping, Integer primary, Integer latest) {
+ public InconsistentMapping(Map<Ignite, T2<Integer, GridCacheVersion>> mapping, Integer primary, Integer fixed,
+ boolean repairable, boolean consistent) {
this.mapping = new HashMap<>(mapping);
this.primary = primary;
- this.latest = latest;
+ this.fixed = fixed;
+ this.repairable = repairable;
+ this.consistent = consistent;
}
}
}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/AtomicReadRepairTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/AtomicReadRepairTest.java
deleted file mode 100644
index ac65b95..0000000
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/AtomicReadRepairTest.java
+++ /dev/null
@@ -1,207 +0,0 @@
-/*
- * 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.processors.cache.consistency;
-
-import java.util.Collection;
-import java.util.Map;
-import java.util.Set;
-import java.util.function.Consumer;
-import javax.cache.CacheException;
-import org.apache.ignite.Ignite;
-import org.apache.ignite.IgniteCache;
-import org.apache.ignite.cache.CacheAtomicityMode;
-import org.apache.ignite.cache.CacheEntry;
-import org.apache.ignite.internal.processors.cache.distributed.near.consistency.IgniteConsistencyViolationException;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
-
-/**
- * Ideally, this test should be equal to {@link ImplicitTransactionalReadRepairTest} except atomicy mode
- * configuration.
- *
- * Implement repair on atomic caches to make this happen :)
- */
-@RunWith(Parameterized.class)
-public class AtomicReadRepairTest extends ImplicitTransactionalReadRepairTest {
- /**
- *
- */
- private static final Consumer<ReadRepairData> GET_CHECK_AND_FAIL = (data) -> {
- IgniteCache<Integer, Integer> cache = data.cache;
- Set<Integer> keys = data.data.keySet();
- boolean raw = data.raw;
- boolean async = data.async;
-
- assert keys.size() == 1;
-
- for (Map.Entry<Integer, InconsistentMapping> entry : data.data.entrySet()) { // Once.
- try {
- Integer key = entry.getKey();
-
- Integer res =
- raw ?
- async ?
- cache.withReadRepair().getEntryAsync(key).get().getValue() :
- cache.withReadRepair().getEntry(key).getValue() :
- async ?
- cache.withReadRepair().getAsync(key).get() :
- cache.withReadRepair().get(key);
-
- fail("Should not happen.");
- }
- catch (CacheException e) {
- assertTrue(e.getCause() instanceof IgniteConsistencyViolationException);
- }
- }
- };
-
- /**
- *
- */
- private static final Consumer<ReadRepairData> GETALL_CHECK_AND_FAIL = (data) -> {
- IgniteCache<Integer, Integer> cache = data.cache;
- Set<Integer> keys = data.data.keySet();
- boolean raw = data.raw;
- boolean async = data.async;
-
- assert !keys.isEmpty();
-
- try {
- if (raw) {
- Collection<CacheEntry<Integer, Integer>> res =
- async ?
- cache.withReadRepair().getEntriesAsync(keys).get() :
- cache.withReadRepair().getEntries(keys);
-
- for (CacheEntry<Integer, Integer> entry : res) {
- Integer latest = data.data.get(entry.getKey()).latest;
-
- if (latest != null)
- assertEquals(latest, entry.getValue());
- }
- }
- else {
- Map<Integer, Integer> res =
- async ?
- cache.withReadRepair().getAllAsync(keys).get() :
- cache.withReadRepair().getAll(keys);
-
- for (Map.Entry<Integer, Integer> entry : res.entrySet()) {
- Integer latest = data.data.get(entry.getKey()).latest;
-
- if (latest != null)
- assertEquals(latest, entry.getValue());
- }
- }
-
- fail("Should not happen.");
- }
- catch (CacheException e) {
- assertTrue(e.getCause() instanceof IgniteConsistencyViolationException);
- }
- };
-
- /**
- *
- */
- private static final Consumer<ReadRepairData> CONTAINS_CHECK_AND_FAIL = (data) -> {
- IgniteCache<Integer, Integer> cache = data.cache;
- Set<Integer> keys = data.data.keySet();
- boolean async = data.async;
-
- assert keys.size() == 1;
-
- for (Map.Entry<Integer, InconsistentMapping> entry : data.data.entrySet()) { // Once.
- try {
- Integer key = entry.getKey();
-
- boolean res = async ?
- cache.withReadRepair().containsKeyAsync(key).get() :
- cache.withReadRepair().containsKey(key);
-
- fail("Should not happen.");
- }
- catch (Exception e) {
- assertTrue(e.getCause() instanceof IgniteConsistencyViolationException);
- }
- }
- };
-
- /**
- *
- */
- private static final Consumer<ReadRepairData> CONTAINS_ALL_CHECK_AND_FAIL = (data) -> {
- IgniteCache<Integer, Integer> cache = data.cache;
- Set<Integer> keys = data.data.keySet();
- boolean async = data.async;
-
- try {
- boolean res = async ?
- cache.withReadRepair().containsKeysAsync(keys).get() :
- cache.withReadRepair().containsKeys(keys);
-
- fail("Should not happen.");
- }
- catch (Exception e) {
- assertTrue(e.getCause() instanceof IgniteConsistencyViolationException);
- }
- };
-
- /** {@inheritDoc} */
- @Override protected CacheAtomicityMode atomicyMode() {
- return CacheAtomicityMode.ATOMIC;
- }
-
- /** {@inheritDoc} */
- @Override protected void testGet(Ignite initiator, Integer cnt, boolean all) throws Exception {
- prepareAndCheck(
- initiator,
- cnt,
- raw,
- async,
- misses,
- nulls,
- (ReadRepairData data) -> {
- if (all)
- GETALL_CHECK_AND_FAIL.accept(data);
- else
- GET_CHECK_AND_FAIL.accept(data);
-
- check(data, true, false);
- });
- }
-
- /** {@inheritDoc} */
- @Override protected void testContains(Ignite initiator, Integer cnt, boolean all) throws Exception {
- prepareAndCheck(
- initiator,
- cnt,
- raw,
- async,
- misses,
- nulls,
- (ReadRepairData data) -> {
- if (all)
- CONTAINS_ALL_CHECK_AND_FAIL.accept(data);
- else
- CONTAINS_CHECK_AND_FAIL.accept(data);
-
- check(data, true, false);
- });
- }
-}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/SingleBackupImplicitTransactionalReadRepairTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/inmem/AtomicReadRepairTest.java
similarity index 67%
copy from modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/SingleBackupImplicitTransactionalReadRepairTest.java
copy to modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/inmem/AtomicReadRepairTest.java
index 9d77fd4..e4cc2b8 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/SingleBackupImplicitTransactionalReadRepairTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/inmem/AtomicReadRepairTest.java
@@ -15,14 +15,19 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.processors.cache.consistency;
+package org.apache.ignite.internal.processors.cache.consistency.inmem;
+
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
/**
*
*/
-public class SingleBackupImplicitTransactionalReadRepairTest extends ImplicitTransactionalReadRepairTest {
+@RunWith(Parameterized.class)
+public class AtomicReadRepairTest extends ImplicitTransactionalReadRepairTest {
/** {@inheritDoc} */
- @Override protected Integer backupsCount() {
- return 1; // Single backup possible optimisations check.
+ @Override protected CacheAtomicityMode atomicityMode() {
+ return CacheAtomicityMode.ATOMIC;
}
}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/ExplicitTransactionalReadRepairTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/inmem/ExplicitTransactionalReadRepairTest.java
similarity index 61%
rename from modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/ExplicitTransactionalReadRepairTest.java
rename to modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/inmem/ExplicitTransactionalReadRepairTest.java
index 2935bc9..0314f53 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/ExplicitTransactionalReadRepairTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/inmem/ExplicitTransactionalReadRepairTest.java
@@ -15,12 +15,14 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.processors.cache.consistency;
+package org.apache.ignite.internal.processors.cache.consistency.inmem;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
+import java.util.function.Consumer;
import org.apache.ignite.Ignite;
+import org.apache.ignite.internal.processors.cache.consistency.AbstractFullSetReadRepairTest;
import org.apache.ignite.transactions.Transaction;
import org.apache.ignite.transactions.TransactionConcurrency;
import org.apache.ignite.transactions.TransactionIsolation;
@@ -85,62 +87,31 @@ public class ExplicitTransactionalReadRepairTest extends AbstractFullSetReadRepa
async,
misses,
nulls,
- (ReadRepairData data) -> {
+ (ReadRepairData data) -> repairIfRepairable.accept(data, () -> {
boolean fixByOtherTx = concurrency == TransactionConcurrency.OPTIMISTIC ||
isolation == TransactionIsolation.READ_COMMITTED;
- try (Transaction tx = initiator.transactions().txStart(concurrency, isolation)) {
- // Recovery (inside tx).
- if (all)
- GETALL_CHECK_AND_FIX.accept(data);
- else
- GET_CHECK_AND_FIX.accept(data);
-
- check(data, fixByOtherTx, true); // Checks (inside tx).
-
- try {
- tx.commit();
- }
- catch (TransactionRollbackException e) {
- fail("Should not happen. " + e);
- }
- }
-
- check(data, !fixByOtherTx, true); // Checks (outside tx).
- });
+ testReadRepair(initiator, data, all ? GETALL_CHECK_AND_FIX : GET_CHECK_AND_FIX, fixByOtherTx, true);
+ }));
}
/** {@inheritDoc} */
- @Override protected void testGetNull(Ignite initiator) throws Exception {
+ @Override protected void testContains(Ignite initiator, Integer cnt, boolean all) throws Exception {
prepareAndCheck(
initiator,
- 1,
+ cnt,
raw,
async,
misses,
nulls,
- (ReadRepairData data) -> {
- try (Transaction tx = initiator.transactions().txStart(concurrency, isolation)) {
- GET_NULL.accept(data);
-
- checkEventMissed();
-
- try {
- tx.commit();
- }
- catch (TransactionRollbackException e) {
- fail("Should not happen. " + e);
- }
- }
-
- GET_NULL.accept(data); // Checks (outside tx).
-
- checkEventMissed();
- });
+ (ReadRepairData data) -> repairIfRepairable.accept(data, () -> {
+ // "Contains" works like optimistic() || readCommitted() and always fixed by other tx.
+ testReadRepair(initiator, data, all ? CONTAINS_ALL_CHECK_AND_FIX : CONTAINS_CHECK_AND_FIX, true, true);
+ }));
}
/** {@inheritDoc} */
- @Override protected void testContains(Ignite initiator, Integer cnt, boolean all) throws Exception {
+ @Override protected void testGetNull(Ignite initiator, Integer cnt, boolean all) throws Exception {
prepareAndCheck(
initiator,
cnt,
@@ -148,28 +119,34 @@ public class ExplicitTransactionalReadRepairTest extends AbstractFullSetReadRepa
async,
misses,
nulls,
- (ReadRepairData data) -> {
- // "Contains" works like optimistic() || readCommitted() and always fixed by other tx.
- boolean fixByOtherTx = true;
-
- try (Transaction tx = initiator.transactions().txStart(concurrency, isolation)) {
- // Recovery (inside tx).
- if (all)
- CONTAINS_ALL_CHECK_AND_FIX.accept(data);
- else
- CONTAINS_CHECK_AND_FIX.accept(data);
-
- check(data, fixByOtherTx, true); // Checks (inside tx).
-
- try {
- tx.commit();
- }
- catch (TransactionRollbackException e) {
- fail("Should not happen. " + e);
- }
- }
+ (ReadRepairData data) -> testReadRepair(initiator, data, all ? GET_ALL_NULL : GET_NULL, false, false));
+ }
+
+ /**
+ *
+ */
+ private void testReadRepair(Ignite initiator, ReadRepairData data, Consumer<ReadRepairData> readOp,
+ boolean fixByOtherTx, boolean hit) {
+ try (Transaction tx = initiator.transactions().txStart(concurrency, isolation)) {
+ // Recovery (inside tx).
+ readOp.accept(data);
+
+ if (hit) // Checks (inside tx).
+ check(data, null, fixByOtherTx); // Hit.
+ else
+ checkEventMissed(); // Miss.
+
+ try {
+ tx.commit();
+ }
+ catch (TransactionRollbackException e) {
+ fail("Should not happen. " + e);
+ }
+ }
- check(data, !fixByOtherTx, true); // Checks (outside tx).
- });
+ if (hit) // Checks (outside tx).
+ check(data, null, !fixByOtherTx); // Hit.
+ else
+ checkEventMissed(); // Miss.
}
}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/ImplicitTransactionalReadRepairTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/inmem/ImplicitTransactionalReadRepairTest.java
similarity index 72%
rename from modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/ImplicitTransactionalReadRepairTest.java
rename to modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/inmem/ImplicitTransactionalReadRepairTest.java
index a6019fd..dae70a2 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/ImplicitTransactionalReadRepairTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/inmem/ImplicitTransactionalReadRepairTest.java
@@ -15,12 +15,14 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.processors.cache.consistency;
+package org.apache.ignite.internal.processors.cache.consistency.inmem;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
+import java.util.function.Consumer;
import org.apache.ignite.Ignite;
+import org.apache.ignite.internal.processors.cache.consistency.AbstractFullSetReadRepairTest;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
@@ -69,38 +71,25 @@ public class ImplicitTransactionalReadRepairTest extends AbstractFullSetReadRepa
async,
misses,
nulls,
- (ReadRepairData data) -> {
- if (all)
- GETALL_CHECK_AND_FIX.accept(data);
- else
- GET_CHECK_AND_FIX.accept(data);
-
- check(data, true, true);
- });
+ (ReadRepairData data) -> repairIfRepairable.accept(data,
+ () -> testReadRepair(data, all ? GETALL_CHECK_AND_FIX : GET_CHECK_AND_FIX, true)));
}
/** {@inheritDoc} */
- @Override protected void testGetNull(Ignite initiator) throws Exception {
+ @Override protected void testContains(Ignite initiator, Integer cnt, boolean all) throws Exception {
prepareAndCheck(
initiator,
- 1,
+ cnt,
raw,
async,
misses,
nulls,
- (ReadRepairData data) -> {
- GET_NULL.accept(data); // first attempt.
-
- checkEventMissed();
-
- GET_NULL.accept(data); // second attempt (checks first attempt causes no changes/fixes/etc).
-
- checkEventMissed();
- });
+ (ReadRepairData data) -> repairIfRepairable.accept(data,
+ () -> testReadRepair(data, all ? CONTAINS_ALL_CHECK_AND_FIX : CONTAINS_CHECK_AND_FIX, true)));
}
/** {@inheritDoc} */
- @Override protected void testContains(Ignite initiator, Integer cnt, boolean all) throws Exception {
+ @Override protected void testGetNull(Ignite initiator, Integer cnt, boolean all) throws Exception {
prepareAndCheck(
initiator,
cnt,
@@ -108,13 +97,18 @@ public class ImplicitTransactionalReadRepairTest extends AbstractFullSetReadRepa
async,
misses,
nulls,
- (ReadRepairData data) -> {
- if (all)
- CONTAINS_ALL_CHECK_AND_FIX.accept(data);
- else
- CONTAINS_CHECK_AND_FIX.accept(data);
+ (ReadRepairData data) -> testReadRepair(data, all ? GET_ALL_NULL : GET_NULL, false));
+ }
+
+ /**
+ *
+ */
+ private void testReadRepair(ReadRepairData data, Consumer<ReadRepairData> readOp, boolean hit) {
+ readOp.accept(data);
- check(data, true, true);
- });
+ if (hit)
+ check(data, null, true); // Hit.
+ else
+ checkEventMissed(); // Miss.
}
}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/ReplicatedExplicitTransactionalReadRepairTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/inmem/ReplicatedExplicitTransactionalReadRepairTest.java
similarity index 93%
rename from modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/ReplicatedExplicitTransactionalReadRepairTest.java
rename to modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/inmem/ReplicatedExplicitTransactionalReadRepairTest.java
index c5dcd28..aad3f15 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/ReplicatedExplicitTransactionalReadRepairTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/inmem/ReplicatedExplicitTransactionalReadRepairTest.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.processors.cache.consistency;
+package org.apache.ignite.internal.processors.cache.consistency.inmem;
import org.apache.ignite.cache.CacheMode;
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/ReplicatedImplicitTransactionalReadRepairTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/inmem/ReplicatedImplicitTransactionalReadRepairTest.java
similarity index 93%
rename from modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/ReplicatedImplicitTransactionalReadRepairTest.java
rename to modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/inmem/ReplicatedImplicitTransactionalReadRepairTest.java
index a244658..8f84c57 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/ReplicatedImplicitTransactionalReadRepairTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/inmem/ReplicatedImplicitTransactionalReadRepairTest.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.processors.cache.consistency;
+package org.apache.ignite.internal.processors.cache.consistency.inmem;
import org.apache.ignite.cache.CacheMode;
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/SingleBackupExplicitTransactionalReadRepairTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/inmem/SingleBackupExplicitTransactionalReadRepairTest.java
similarity index 93%
rename from modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/SingleBackupExplicitTransactionalReadRepairTest.java
rename to modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/inmem/SingleBackupExplicitTransactionalReadRepairTest.java
index 13ed4bd..6a0f494 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/SingleBackupExplicitTransactionalReadRepairTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/inmem/SingleBackupExplicitTransactionalReadRepairTest.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.processors.cache.consistency;
+package org.apache.ignite.internal.processors.cache.consistency.inmem;
/**
*
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/SingleBackupImplicitTransactionalReadRepairTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/inmem/SingleBackupImplicitTransactionalReadRepairTest.java
similarity index 93%
copy from modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/SingleBackupImplicitTransactionalReadRepairTest.java
copy to modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/inmem/SingleBackupImplicitTransactionalReadRepairTest.java
index 9d77fd4..07c5cd0 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/SingleBackupImplicitTransactionalReadRepairTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/inmem/SingleBackupImplicitTransactionalReadRepairTest.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.processors.cache.consistency;
+package org.apache.ignite.internal.processors.cache.consistency.inmem;
/**
*
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/SingleBackupImplicitTransactionalReadRepairTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/persistence/PdsAtomicReadRepairTest.java
similarity index 73%
copy from modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/SingleBackupImplicitTransactionalReadRepairTest.java
copy to modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/persistence/PdsAtomicReadRepairTest.java
index 9d77fd4..86a02b5 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/SingleBackupImplicitTransactionalReadRepairTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/persistence/PdsAtomicReadRepairTest.java
@@ -15,14 +15,16 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.processors.cache.consistency;
+package org.apache.ignite.internal.processors.cache.consistency.persistence;
+
+import org.apache.ignite.internal.processors.cache.consistency.inmem.AtomicReadRepairTest;
/**
*
*/
-public class SingleBackupImplicitTransactionalReadRepairTest extends ImplicitTransactionalReadRepairTest {
+public class PdsAtomicReadRepairTest extends AtomicReadRepairTest {
/** {@inheritDoc} */
- @Override protected Integer backupsCount() {
- return 1; // Single backup possible optimisations check.
+ @Override protected boolean persistenceEnabled() {
+ return true;
}
}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/SingleBackupImplicitTransactionalReadRepairTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/persistence/PdsExplicitTransactionalReadRepairTest.java
similarity index 70%
copy from modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/SingleBackupImplicitTransactionalReadRepairTest.java
copy to modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/persistence/PdsExplicitTransactionalReadRepairTest.java
index 9d77fd4..eb74690 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/SingleBackupImplicitTransactionalReadRepairTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/persistence/PdsExplicitTransactionalReadRepairTest.java
@@ -15,14 +15,16 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.processors.cache.consistency;
+package org.apache.ignite.internal.processors.cache.consistency.persistence;
+
+import org.apache.ignite.internal.processors.cache.consistency.inmem.ExplicitTransactionalReadRepairTest;
/**
*
*/
-public class SingleBackupImplicitTransactionalReadRepairTest extends ImplicitTransactionalReadRepairTest {
+public class PdsExplicitTransactionalReadRepairTest extends ExplicitTransactionalReadRepairTest {
/** {@inheritDoc} */
- @Override protected Integer backupsCount() {
- return 1; // Single backup possible optimisations check.
+ @Override protected boolean persistenceEnabled() {
+ return true;
}
}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/SingleBackupImplicitTransactionalReadRepairTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/persistence/PdsImplicitTransactionalReadRepairTest.java
similarity index 70%
rename from modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/SingleBackupImplicitTransactionalReadRepairTest.java
rename to modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/persistence/PdsImplicitTransactionalReadRepairTest.java
index 9d77fd4..f73cf79 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/SingleBackupImplicitTransactionalReadRepairTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/consistency/persistence/PdsImplicitTransactionalReadRepairTest.java
@@ -15,14 +15,16 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.processors.cache.consistency;
+package org.apache.ignite.internal.processors.cache.consistency.persistence;
+
+import org.apache.ignite.internal.processors.cache.consistency.inmem.ImplicitTransactionalReadRepairTest;
/**
*
*/
-public class SingleBackupImplicitTransactionalReadRepairTest extends ImplicitTransactionalReadRepairTest {
+public class PdsImplicitTransactionalReadRepairTest extends ImplicitTransactionalReadRepairTest {
/** {@inheritDoc} */
- @Override protected Integer backupsCount() {
- return 1; // Single backup possible optimisations check.
+ @Override protected boolean persistenceEnabled() {
+ return true;
}
}
diff --git a/modules/core/src/test/java/org/apache/ignite/testframework/junits/multijvm/IgniteCacheProcessProxy.java b/modules/core/src/test/java/org/apache/ignite/testframework/junits/multijvm/IgniteCacheProcessProxy.java
index b71d37d..68d8c4c 100644
--- a/modules/core/src/test/java/org/apache/ignite/testframework/junits/multijvm/IgniteCacheProcessProxy.java
+++ b/modules/core/src/test/java/org/apache/ignite/testframework/junits/multijvm/IgniteCacheProcessProxy.java
@@ -39,6 +39,7 @@ import org.apache.ignite.cache.CacheEntry;
import org.apache.ignite.cache.CacheEntryProcessor;
import org.apache.ignite.cache.CacheMetrics;
import org.apache.ignite.cache.CachePeekMode;
+import org.apache.ignite.cache.ReadRepairStrategy;
import org.apache.ignite.cache.query.FieldsQueryCursor;
import org.apache.ignite.cache.query.Query;
import org.apache.ignite.cache.query.QueryCursor;
@@ -140,7 +141,7 @@ public class IgniteCacheProcessProxy<K, V> implements IgniteCache<K, V> {
}
/** {@inheritDoc} */
- @Override public IgniteCache<K, V> withReadRepair() {
+ @Override public IgniteCache<K, V> withReadRepair(ReadRepairStrategy strategy) {
throw new UnsupportedOperationException("Method should be supported.");
}
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheConsistencySelfTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheConsistencySelfTestSuite.java
new file mode 100644
index 0000000..82bd4b8
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheConsistencySelfTestSuite.java
@@ -0,0 +1,55 @@
+/*
+ * 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.testsuites;
+
+import org.apache.ignite.internal.processors.cache.consistency.inmem.AtomicReadRepairTest;
+import org.apache.ignite.internal.processors.cache.consistency.inmem.ExplicitTransactionalReadRepairTest;
+import org.apache.ignite.internal.processors.cache.consistency.inmem.ImplicitTransactionalReadRepairTest;
+import org.apache.ignite.internal.processors.cache.consistency.inmem.ReplicatedExplicitTransactionalReadRepairTest;
+import org.apache.ignite.internal.processors.cache.consistency.inmem.ReplicatedImplicitTransactionalReadRepairTest;
+import org.apache.ignite.internal.processors.cache.consistency.inmem.SingleBackupExplicitTransactionalReadRepairTest;
+import org.apache.ignite.internal.processors.cache.consistency.inmem.SingleBackupImplicitTransactionalReadRepairTest;
+import org.apache.ignite.internal.processors.cache.consistency.persistence.PdsAtomicReadRepairTest;
+import org.apache.ignite.internal.processors.cache.consistency.persistence.PdsExplicitTransactionalReadRepairTest;
+import org.apache.ignite.internal.processors.cache.consistency.persistence.PdsImplicitTransactionalReadRepairTest;
+import org.junit.runner.RunWith;
+import org.junit.runners.Suite;
+
+/**
+ * Test suite for cache consistency checks.
+ */
+@RunWith(Suite.class)
+@Suite.SuiteClasses({
+ // Inmem
+ AtomicReadRepairTest.class,
+ ExplicitTransactionalReadRepairTest.class,
+ ImplicitTransactionalReadRepairTest.class,
+
+ // PDS
+ PdsAtomicReadRepairTest.class,
+ PdsExplicitTransactionalReadRepairTest.class,
+ PdsImplicitTransactionalReadRepairTest.class,
+
+ // Special (inmem)
+ ReplicatedExplicitTransactionalReadRepairTest.class,
+ ReplicatedImplicitTransactionalReadRepairTest.class,
+ SingleBackupExplicitTransactionalReadRepairTest.class,
+ SingleBackupImplicitTransactionalReadRepairTest.class,
+})
+public class IgniteCacheConsistencySelfTestSuite {
+}
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheDataStructuresSelfTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheDataStructuresSelfTestSuite.java
index dc0810d..895e841 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheDataStructuresSelfTestSuite.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheDataStructuresSelfTestSuite.java
@@ -18,13 +18,6 @@
package org.apache.ignite.testsuites;
import org.apache.ignite.internal.processors.cache.AtomicCacheAffinityConfigurationTest;
-import org.apache.ignite.internal.processors.cache.consistency.AtomicReadRepairTest;
-import org.apache.ignite.internal.processors.cache.consistency.ExplicitTransactionalReadRepairTest;
-import org.apache.ignite.internal.processors.cache.consistency.ImplicitTransactionalReadRepairTest;
-import org.apache.ignite.internal.processors.cache.consistency.ReplicatedExplicitTransactionalReadRepairTest;
-import org.apache.ignite.internal.processors.cache.consistency.ReplicatedImplicitTransactionalReadRepairTest;
-import org.apache.ignite.internal.processors.cache.consistency.SingleBackupExplicitTransactionalReadRepairTest;
-import org.apache.ignite.internal.processors.cache.consistency.SingleBackupImplicitTransactionalReadRepairTest;
import org.apache.ignite.internal.processors.cache.datastructures.GridCacheQueueCleanupSelfTest;
import org.apache.ignite.internal.processors.cache.datastructures.GridCacheQueueClientDisconnectTest;
import org.apache.ignite.internal.processors.cache.datastructures.GridCacheQueueMultiNodeConsistencySelfTest;
@@ -208,16 +201,6 @@ import org.junit.runners.Suite;
IgniteCacheDataStructuresBinarySelfTestSuite.class,
- ImplicitTransactionalReadRepairTest.class,
- SingleBackupImplicitTransactionalReadRepairTest.class,
- ReplicatedImplicitTransactionalReadRepairTest.class,
-
- AtomicReadRepairTest.class,
-
- ExplicitTransactionalReadRepairTest.class,
- SingleBackupExplicitTransactionalReadRepairTest.class,
- ReplicatedExplicitTransactionalReadRepairTest.class,
-
IgniteAtomicLongClusterReadOnlyTest.class,
IgniteAtomicReferenceClusterReadOnlyTest.class,
IgniteAtomicSequenceClusterReadOnlyTest.class,
diff --git a/modules/ducktests/tests/ignitetest/tests/control_utility/consistency_test.py b/modules/ducktests/tests/ignitetest/tests/control_utility/consistency_test.py
index 2caa273..fb3c466 100644
--- a/modules/ducktests/tests/ignitetest/tests/control_utility/consistency_test.py
+++ b/modules/ducktests/tests/ignitetest/tests/control_utility/consistency_test.py
@@ -101,7 +101,7 @@ class ConsistencyTest(IgniteTest):
except AssertionError:
pass
- control_utility.check_consistency(f"repair {self.CACHE_NAME} 0") # checking/repairing
+ control_utility.check_consistency(f"repair {self.CACHE_NAME} 0 LWW") # checking/repairing
message = "Cache consistency violations recorded."
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/dml/DmlUtils.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/dml/DmlUtils.java
index 9ef416e..2b6a89b 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/dml/DmlUtils.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/dml/DmlUtils.java
@@ -559,7 +559,7 @@ public class DmlUtils {
if (opCtx == null)
// Mimics behavior of GridCacheAdapter#keepBinary and GridCacheProxyImpl#keepBinary
- newOpCtx = new CacheOperationContext(false, true, null, false, null, false, false, true);
+ newOpCtx = new CacheOperationContext(false, true, null, false, null, false, null, true);
else if (!opCtx.isKeepBinary())
newOpCtx = opCtx.keepBinary();