You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by nt...@apache.org on 2017/12/08 10:36:36 UTC
ignite git commit: IGNITE-7086 - Backups are not updated when
ReadFromBackup=true and ReadThrough happens.
Repository: ignite
Updated Branches:
refs/heads/master a7893b630 -> c76983833
IGNITE-7086 - Backups are not updated when ReadFromBackup=true and ReadThrough happens.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/c7698383
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/c7698383
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/c7698383
Branch: refs/heads/master
Commit: c7698383331b9680d28fbd8eca1d8584d41c062e
Parents: a7893b6
Author: dkarachentsev <dk...@gridgain.com>
Authored: Fri Dec 8 13:36:28 2017 +0300
Committer: dkarachentsev <dk...@gridgain.com>
Committed: Fri Dec 8 13:36:28 2017 +0300
----------------------------------------------------------------------
.../processors/cache/GridCacheUtils.java | 105 ++++++++
.../distributed/dht/GridDhtCacheAdapter.java | 1 +
.../dht/GridPartitionedGetFuture.java | 30 ++-
.../dht/GridPartitionedSingleGetFuture.java | 27 ++-
.../distributed/near/GridNearGetFuture.java | 32 ++-
.../store/CacheStoreReadFromBackupTest.java | 238 +++++++++++++++++++
...cheTransactionalStoreReadFromBackupTest.java | 32 +++
.../ignite/testsuites/IgniteCacheTestSuite.java | 4 +
8 files changed, 460 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/c7698383/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUtils.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUtils.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUtils.java
index 248f2aa..4bf54bf 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUtils.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUtils.java
@@ -66,10 +66,14 @@ import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.cluster.ClusterTopologyServerNotFoundException;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.distributed.GridDistributedLockCancelledException;
+import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCacheAdapter;
+import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtInvalidPartitionException;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearCacheAdapter;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLocal;
import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxEntry;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
+import org.apache.ignite.internal.processors.dr.GridDrType;
import org.apache.ignite.internal.processors.igfs.IgfsUtils;
import org.apache.ignite.internal.processors.query.QueryUtils;
import org.apache.ignite.internal.processors.query.schema.SchemaOperationException;
@@ -84,6 +88,7 @@ import org.apache.ignite.internal.util.typedef.X;
import org.apache.ignite.internal.util.typedef.internal.A;
import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.lang.IgniteBiInClosure;
import org.apache.ignite.lang.IgniteClosure;
import org.apache.ignite.lang.IgniteInClosure;
import org.apache.ignite.lang.IgnitePredicate;
@@ -1678,6 +1683,99 @@ public class GridCacheUtils {
}
/**
+ * Creates closure that saves initial value to backup partition.
+ * <p>
+ * Useful only when store with readThrough is used. In situation when
+ * get() on backup node returns successful result, it's expected that
+ * localPeek() will be successful as well. But it doesn't true when
+ * primary node loaded value from local store, in this case backups
+ * will remain non-initialized.
+ * <br>
+ * To meet that requirement the value requested from primary should
+ * be saved on backup during get().
+ * </p>
+ *
+ * @param topVer Topology version.
+ * @param log Logger.
+ * @param cctx Cache context.
+ * @param key Key.
+ * @param readThrough Read through.
+ * @param skipVals Skip values.
+ */
+ @Nullable public static BackupPostProcessingClosure createBackupPostProcessingClosure(
+ final AffinityTopologyVersion topVer,
+ final IgniteLogger log,
+ final GridCacheContext cctx,
+ final @Nullable KeyCacheObject key,
+ boolean readThrough,
+ boolean skipVals
+ ) {
+ if (!readThrough || skipVals ||
+ (key != null && !cctx.affinity().backupsByKey(key, topVer).contains(cctx.localNode())))
+ return null;
+
+ return new BackupPostProcessingClosure() {
+ private void process(KeyCacheObject key, CacheObject val, GridCacheVersion ver, GridDhtCacheAdapter colocated) {
+ while (true) {
+ GridCacheEntryEx entry = null;
+
+ try {
+ entry = colocated.entryEx(key, topVer);
+
+ entry.initialValue(
+ val,
+ ver,
+ 0,
+ 0,
+ false,
+ topVer,
+ GridDrType.DR_BACKUP,
+ true);
+
+ break;
+ }
+ catch (GridCacheEntryRemovedException ignore) {
+ if (log.isDebugEnabled())
+ log.debug("Got removed entry during postprocessing (will retry): " +
+ entry);
+ }
+ catch (IgniteCheckedException e) {
+ U.error(log, "Error saving backup value: " + entry, e);
+ }
+ catch (GridDhtInvalidPartitionException ignored) {
+ break;
+ }
+ finally {
+ if (entry != null)
+ cctx.evicts().touch(entry, topVer);
+ }
+ }
+ }
+
+ @Override public void apply(CacheObject val, GridCacheVersion ver) {
+ process(key, val, ver, cctx.dht());
+ }
+
+ @Override public void apply(Collection<GridCacheEntryInfo> infos) {
+ if (!F.isEmpty(infos)) {
+ GridCacheAffinityManager aff = cctx.affinity();
+ ClusterNode locNode = cctx.localNode();
+
+ GridDhtCacheAdapter colocated = cctx.cache().isNear()
+ ? ((GridNearCacheAdapter)cctx.cache()).dht()
+ : cctx.dht();
+
+ for (GridCacheEntryInfo info : infos) {
+ // Save backup value.
+ if (aff.backupsByKey(info.key(), topVer).contains(locNode))
+ process(info.key(), info.value(), info.version(), colocated);
+ }
+ }
+ }
+ };
+ }
+
+ /**
* Checks if cache configuration belongs to persistent cache.
*
* @param ccfg Cache configuration.
@@ -1752,4 +1850,11 @@ public class GridCacheUtils {
return false;
}
+
+ /**
+ *
+ */
+ public interface BackupPostProcessingClosure extends IgniteInClosure<Collection<GridCacheEntryInfo>>,
+ IgniteBiInClosure<CacheObject, GridCacheVersion>{
+ }
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/c7698383/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java
----------------------------------------------------------------------
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 bbb2c5b..28f9c76 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
@@ -287,6 +287,7 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
}
}
+ /** {@inheritDoc} */
@Override public void onKernalStop() {
super.onKernalStop();
http://git-wip-us.apache.org/repos/asf/ignite/blob/c7698383/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java
index 015eb82..3954cf9 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java
@@ -41,9 +41,9 @@ import org.apache.ignite.internal.processors.cache.GridCacheEntryRemovedExceptio
import org.apache.ignite.internal.processors.cache.GridCacheMessage;
import org.apache.ignite.internal.processors.cache.IgniteCacheExpiryPolicy;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
-import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearGetRequest;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearGetResponse;
+import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.util.GridLeanMap;
import org.apache.ignite.internal.util.future.GridFinishedFuture;
@@ -57,6 +57,7 @@ import org.apache.ignite.internal.util.typedef.P1;
import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.lang.IgniteInClosure;
import org.apache.ignite.lang.IgniteUuid;
import org.jetbrains.annotations.Nullable;
@@ -224,7 +225,7 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
private void map(
Collection<KeyCacheObject> keys,
Map<ClusterNode, LinkedHashMap<KeyCacheObject, Boolean>> mapped,
- AffinityTopologyVersion topVer
+ final AffinityTopologyVersion topVer
) {
Collection<ClusterNode> cacheNodes = CU.affinityNodes(cctx, topVer);
@@ -331,7 +332,8 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
}));
}
else {
- MiniFuture fut = new MiniFuture(n, mappedKeys, topVer);
+ MiniFuture fut = new MiniFuture(n, mappedKeys, topVer,
+ CU.createBackupPostProcessingClosure(topVer, log, cctx, null, readThrough, skipVals));
GridCacheMessage req = new GridNearGetRequest(
cctx.cacheId(),
@@ -647,6 +649,9 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
/** Topology version on which this future was mapped. */
private final AffinityTopologyVersion topVer;
+ /** Post processing closure. */
+ private final IgniteInClosure<Collection<GridCacheEntryInfo>> postProcessingClos;
+
/** {@code True} if remapped after node left. */
private boolean remapped;
@@ -654,11 +659,14 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
* @param node Node.
* @param keys Keys.
* @param topVer Topology version.
+ * @param postProcessingClos Post processing closure.
*/
- MiniFuture(ClusterNode node, LinkedHashMap<KeyCacheObject, Boolean> keys, AffinityTopologyVersion topVer) {
+ MiniFuture(ClusterNode node, LinkedHashMap<KeyCacheObject, Boolean> keys, AffinityTopologyVersion topVer,
+ @Nullable IgniteInClosure<Collection<GridCacheEntryInfo>> postProcessingClos) {
this.node = node;
this.keys = keys;
this.topVer = topVer;
+ this.postProcessingClos = postProcessingClos;
}
/**
@@ -774,6 +782,8 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
}
}), F.t(node, keys), topVer);
+ postProcessResult(res);
+
onDone(createResultMap(res.entries()));
return;
@@ -795,12 +805,16 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
}
}), F.t(node, keys), topVer);
+ postProcessResult(res);
+
onDone(createResultMap(res.entries()));
}
});
}
else {
try {
+ postProcessResult(res);
+
onDone(createResultMap(res.entries()));
}
catch (Exception e) {
@@ -809,6 +823,14 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
}
}
+ /**
+ * @param res Response.
+ */
+ private void postProcessResult(final GridNearGetResponse res) {
+ if (postProcessingClos != null)
+ postProcessingClos.apply(res.entries());
+ }
+
/** {@inheritDoc} */
@Override public String toString() {
return S.toString(MiniFuture.class, this);
http://git-wip-us.apache.org/repos/asf/ignite/blob/c7698383/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
index 61489e5..f761b9c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
@@ -39,18 +39,20 @@ import org.apache.ignite.internal.processors.cache.GridCacheEntryRemovedExceptio
import org.apache.ignite.internal.processors.cache.GridCacheFuture;
import org.apache.ignite.internal.processors.cache.GridCacheFutureAdapter;
import org.apache.ignite.internal.processors.cache.GridCacheMessage;
+import org.apache.ignite.internal.processors.cache.GridCacheUtils.BackupPostProcessingClosure;
import org.apache.ignite.internal.processors.cache.IgniteCacheExpiryPolicy;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
-import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
import org.apache.ignite.internal.processors.cache.distributed.near.CacheVersionedValue;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearGetResponse;
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.persistence.CacheDataRow;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
import org.apache.ignite.internal.util.typedef.CI1;
import org.apache.ignite.internal.util.typedef.CIX1;
import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteUuid;
@@ -122,6 +124,9 @@ public class GridPartitionedSingleGetFuture extends GridCacheFutureAdapter<Objec
@GridToStringInclude
private ClusterNode node;
+ /** Post processing closure. */
+ private volatile BackupPostProcessingClosure postProcessingClos;
+
/**
* @param cctx Context.
* @param key Key.
@@ -275,6 +280,17 @@ public class GridPartitionedSingleGetFuture extends GridCacheFutureAdapter<Objec
cctx.mvcc().addFuture(this, futId);
}
+ boolean needVer = this.needVer;
+
+ final BackupPostProcessingClosure postClos = CU.createBackupPostProcessingClosure(topVer, log, cctx, key, readThrough, skipVals);
+
+ if (postClos != null) {
+ // Need version to correctly store value.
+ needVer = true;
+
+ postProcessingClos = postClos;
+ }
+
GridCacheMessage req = new GridNearSingleGetRequest(cctx.cacheId(),
futId.localId(),
key,
@@ -502,6 +518,12 @@ public class GridPartitionedSingleGetFuture extends GridCacheFutureAdapter<Objec
else {
if (skipVals)
setSkipValueResult(res.containsValue(), null);
+ else if (readThrough && res0 instanceof CacheVersionedValue) {
+ // Could be versioned value for store in backup.
+ CacheVersionedValue verVal = (CacheVersionedValue)res0;
+
+ setResult(verVal.value(), verVal.version());
+ }
else
setResult((CacheObject)res0, null);
}
@@ -640,6 +662,9 @@ public class GridPartitionedSingleGetFuture extends GridCacheFutureAdapter<Objec
assert !skipVals;
if (val != null) {
+ if (postProcessingClos != null)
+ postProcessingClos.apply(val, ver);
+
if (!keepCacheObjects) {
Object res = cctx.unwrapBinaryIfNeeded(val, !deserializeBinary);
http://git-wip-us.apache.org/repos/asf/ignite/blob/c7698383/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
index 2361507..75b9baf 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
@@ -39,6 +39,7 @@ import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
import org.apache.ignite.internal.processors.cache.GridCacheEntryInfo;
import org.apache.ignite.internal.processors.cache.GridCacheEntryRemovedException;
import org.apache.ignite.internal.processors.cache.GridCacheMessage;
+import org.apache.ignite.internal.processors.cache.GridCacheUtils.BackupPostProcessingClosure;
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.CacheDistributedGetFutureAdapter;
@@ -363,7 +364,8 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
cctx.mvcc().addFuture(this, futId);
}
- MiniFuture fut = new MiniFuture(n, mappedKeys, saved, topVer);
+ MiniFuture fut = new MiniFuture(n, mappedKeys, saved, topVer,
+ CU.createBackupPostProcessingClosure(topVer, log, cctx, null, readThrough, skipVals));
GridCacheMessage req = new GridNearGetRequest(
cctx.cacheId(),
@@ -398,6 +400,8 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
}
}
+
+
/**
* @param mappings Mappings.
* @param key Key to map.
@@ -834,6 +838,9 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
/** Topology version on which this future was mapped. */
private AffinityTopologyVersion topVer;
+ /** Post processing closure. */
+ private final BackupPostProcessingClosure postProcessingClos;
+
/** {@code True} if remapped after node left. */
private boolean remapped;
@@ -842,17 +849,19 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
* @param keys Keys.
* @param savedEntries Saved entries.
* @param topVer Topology version.
+ * @param postProcessingClos Post processing closure.
*/
MiniFuture(
ClusterNode node,
LinkedHashMap<KeyCacheObject, Boolean> keys,
Map<KeyCacheObject, GridNearCacheEntry> savedEntries,
- AffinityTopologyVersion topVer
- ) {
+ AffinityTopologyVersion topVer,
+ BackupPostProcessingClosure postProcessingClos) {
this.node = node;
this.keys = keys;
this.savedEntries = savedEntries;
this.topVer = topVer;
+ this.postProcessingClos = postProcessingClos;
}
/**
@@ -977,6 +986,8 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
}
}), F.t(node, keys), topVer);
+ postProcessResult(res);
+
// It is critical to call onDone after adding futures to compound list.
onDone(loadEntries(node.id(), keys.keySet(), res.entries(), savedEntries, topVer));
@@ -998,13 +1009,26 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
}
}), F.t(node, keys), readyTopVer);
+ postProcessResult(res);
+
// It is critical to call onDone after adding futures to compound list.
onDone(loadEntries(node.id(), keys.keySet(), res.entries(), savedEntries, topVer));
}
});
}
- else
+ else {
+ postProcessResult(res);
+
onDone(loadEntries(node.id(), keys.keySet(), res.entries(), savedEntries, topVer));
+ }
+ }
+
+ /**
+ * @param res Response.
+ */
+ private void postProcessResult(final GridNearGetResponse res) {
+ if (postProcessingClos != null)
+ postProcessingClos.apply(res.entries());
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/c7698383/modules/core/src/test/java/org/apache/ignite/cache/store/CacheStoreReadFromBackupTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/cache/store/CacheStoreReadFromBackupTest.java b/modules/core/src/test/java/org/apache/ignite/cache/store/CacheStoreReadFromBackupTest.java
new file mode 100644
index 0000000..d8913dc
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/cache/store/CacheStoreReadFromBackupTest.java
@@ -0,0 +1,238 @@
+/*
+ * 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.store;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import javax.cache.Cache;
+import javax.cache.configuration.FactoryBuilder;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.cache.CacheWriteSynchronizationMode;
+import org.apache.ignite.cache.affinity.Affinity;
+import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.configuration.NearCacheConfiguration;
+import org.apache.ignite.lang.IgniteBiInClosure;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+
+import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC;
+import static org.apache.ignite.cache.CacheMode.PARTITIONED;
+import static org.apache.ignite.cache.CacheMode.REPLICATED;
+import static org.apache.ignite.cache.CachePeekMode.BACKUP;
+import static org.apache.ignite.cache.CachePeekMode.PRIMARY;
+
+/**
+ * Checks that once value is read from store, it will be loaded in
+ * backups as well.
+ */
+public class CacheStoreReadFromBackupTest extends GridCommonAbstractTest {
+ /** */
+ public static final String CACHE_NAME = "cache";
+
+ /** */
+ private static final Map<Integer, String> storeMap = new ConcurrentHashMap<>();
+
+ /** */
+ private CacheMode cacheMode = REPLICATED;
+
+ /** */
+ private int backups;
+
+ /** Near. */
+ private boolean near;
+
+ /** */
+ @SuppressWarnings("unchecked")
+ private CacheConfiguration cacheConfig(String cacheName) {
+ CacheConfiguration ccfg = new CacheConfiguration<>(cacheName);
+
+ ccfg.setCacheMode(cacheMode);
+ ccfg.setBackups(backups);
+ ccfg.setAtomicityMode(atomicityMode());
+ ccfg.setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC);
+ ccfg.setAffinity(new RendezvousAffinityFunction(false, 1));
+ ccfg.setReadThrough(true);
+ ccfg.setReadFromBackup(true);
+ ccfg.setCacheStoreFactory(FactoryBuilder.factoryOf(TestStore.class));
+
+ if (near)
+ ccfg.setNearConfiguration(new NearCacheConfiguration());
+
+ return ccfg;
+ }
+
+ /** */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ return super.getConfiguration(gridName).setCacheConfiguration(cacheConfig(CACHE_NAME));
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ stopAllGrids();
+ }
+
+ /**
+ * @return Atomicity mode.
+ */
+ protected CacheAtomicityMode atomicityMode() {
+ return ATOMIC;
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testReplicated() throws Exception {
+ cacheMode = REPLICATED;
+ backups = 0;
+ near = false;
+
+ checkReadFromBackup();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testPartitioned() throws Exception {
+ cacheMode = PARTITIONED;
+ backups = 1;
+ near = false;
+
+ checkReadFromBackup();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNearReplicated() throws Exception {
+ cacheMode = REPLICATED;
+ backups = 0;
+ near = true;
+
+ checkReadFromBackup();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNearPartitioned() throws Exception {
+ cacheMode = PARTITIONED;
+ backups = 1;
+ near = true;
+
+ checkReadFromBackup();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ private void checkReadFromBackup() throws Exception {
+ startGridsMultiThreaded(2, true);
+
+ checkReadSingleFromBackup();
+ checkReadAllFromBackup();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ private void checkReadSingleFromBackup() throws Exception {
+ storeMap.put(1, "val-1");
+
+ IgniteCache<Integer, String> cache0 = grid(0).cache(CACHE_NAME);
+ IgniteCache<Integer, String> cache1 = grid(1).cache(CACHE_NAME);
+
+ // Load value on primary and backup.
+ assertNotNull(cache0.get(1));
+ assertNotNull(cache1.get(1));
+
+ if (cache0.localPeek(1, PRIMARY) != null)
+ assertNotNull(cache1.localPeek(1, BACKUP));
+ else {
+ assertNotNull(cache0.localPeek(1, BACKUP));
+ assertNotNull(cache1.localPeek(1, PRIMARY));
+ }
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ private void checkReadAllFromBackup() throws Exception {
+ for (int i = 0; i < 100; i++)
+ storeMap.put(i, String.valueOf(i));
+
+ IgniteCache<Integer, String> cache0 = grid(0).cache(CACHE_NAME);
+ IgniteCache<Integer, String> cache1 = grid(1).cache(CACHE_NAME);
+
+ assertEquals(storeMap.size(), cache0.getAll(storeMap.keySet()).size());
+ assertEquals(storeMap.size(), cache1.getAll(storeMap.keySet()).size());
+
+ Affinity<Integer> aff = grid(0).affinity(CACHE_NAME);
+ ClusterNode node0 = grid(0).cluster().localNode();
+
+ for (Integer key : storeMap.keySet()) {
+ if (aff.isPrimary(node0, key)) {
+ assertNotNull(cache0.localPeek(key, PRIMARY));
+ assertNotNull(cache1.localPeek(key, BACKUP));
+ }
+ else {
+ assertNotNull(cache0.localPeek(key, BACKUP));
+ assertNotNull(cache1.localPeek(key, PRIMARY));
+ }
+ }
+ }
+
+ /**
+ *
+ */
+ public static class TestStore extends CacheStoreAdapter<Integer, String> {
+ /** */
+ public TestStore() {
+ }
+
+ /** {@inheritDoc} */
+ @Override public void loadCache(IgniteBiInClosure<Integer, String> clo, Object... args) {
+ for (Map.Entry<Integer, String> e : storeMap.entrySet())
+ clo.apply(e.getKey(), e.getValue());
+ }
+
+ /** {@inheritDoc} */
+ @Override public String load(Integer key) {
+ return storeMap.get(key);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void write(Cache.Entry<? extends Integer, ? extends String> entry) {
+ storeMap.put(entry.getKey(), entry.getValue());
+ }
+
+ /** {@inheritDoc} */
+ @SuppressWarnings("SuspiciousMethodCalls")
+ @Override public void delete(Object key) {
+ storeMap.remove(key);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void sessionEnd(boolean commit) {
+ // No-op.
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/c7698383/modules/core/src/test/java/org/apache/ignite/cache/store/CacheTransactionalStoreReadFromBackupTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/cache/store/CacheTransactionalStoreReadFromBackupTest.java b/modules/core/src/test/java/org/apache/ignite/cache/store/CacheTransactionalStoreReadFromBackupTest.java
new file mode 100644
index 0000000..4837936
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/cache/store/CacheTransactionalStoreReadFromBackupTest.java
@@ -0,0 +1,32 @@
+/*
+ * 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.store;
+
+import org.apache.ignite.cache.CacheAtomicityMode;
+
+import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
+
+/**
+ *
+ */
+public class CacheTransactionalStoreReadFromBackupTest extends CacheStoreReadFromBackupTest {
+ /** {@inheritDoc} */
+ @Override protected CacheAtomicityMode atomicityMode() {
+ return TRANSACTIONAL;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/c7698383/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite.java
index e3ebbc1..b309b0a 100755
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite.java
@@ -20,6 +20,8 @@ package org.apache.ignite.testsuites;
import java.util.Set;
import junit.framework.TestSuite;
import org.apache.ignite.cache.IgniteWarmupClosureSelfTest;
+import org.apache.ignite.cache.store.CacheStoreReadFromBackupTest;
+import org.apache.ignite.cache.store.CacheTransactionalStoreReadFromBackupTest;
import org.apache.ignite.cache.store.GridCacheBalancingStoreSelfTest;
import org.apache.ignite.cache.store.GridCacheLoadOnlyStoreAdapterSelfTest;
import org.apache.ignite.cache.store.GridStoreLoadCacheTest;
@@ -318,6 +320,8 @@ public class IgniteCacheTestSuite extends TestSuite {
suite.addTestSuite(IgniteIncompleteCacheObjectSelfTest.class);
suite.addTestSuite(GridStoreLoadCacheTest.class);
+ suite.addTestSuite(CacheStoreReadFromBackupTest.class);
+ suite.addTestSuite(CacheTransactionalStoreReadFromBackupTest.class);
return suite;
}