ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [2/2] incubator-ignite git commit: # ignite-42
Date Wed, 21 Jan 2015 09:40:04 GMT
# ignite-42


Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/25f7984a
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/25f7984a
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/25f7984a

Branch: refs/heads/ignite-42
Commit: 25f7984a2a581d94fac6bc282cc9f8fb8097c3b7
Parents: ee7bb3f
Author: sboikov <sboikov@gridgain.com>
Authored: Wed Jan 21 10:28:40 2015 +0300
Committer: sboikov <sboikov@gridgain.com>
Committed: Wed Jan 21 12:36:45 2015 +0300

----------------------------------------------------------------------
 .../integration/GridClientAbstractSelfTest.java |   1 +
 .../apache/ignite/cache/CacheConfiguration.java |  18 +
 .../apache/ignite/cache/store/CacheStore.java   |  12 +
 .../processors/cache/GridCacheAttributes.java   |  44 +-
 .../processors/cache/GridCacheContext.java      |   7 +
 .../processors/cache/GridCacheMapEntry.java     |   4 +-
 .../processors/cache/GridCacheProcessor.java    |  17 +-
 .../processors/cache/GridCacheStoreManager.java |  78 ++--
 .../dht/atomic/GridDhtAtomicCache.java          |   6 +-
 .../local/atomic/GridLocalAtomicCache.java      |   4 +-
 .../transactions/IgniteTxLocalAdapter.java      |   5 +-
 ...CacheJdbcBlobStoreMultithreadedSelfTest.java |   1 +
 .../cache/IgniteCacheAbstractTest.java          |   1 +
 .../IgniteCacheAtomicStoreSessionTest.java      |  55 +++
 .../IgniteCacheLoaderWriterAbstractTest.java    | 157 +++----
 .../IgniteCacheStoreSessionAbstractTest.java    | 303 +++++++++++++
 .../IgniteCacheStoreSessionTest.java            | 435 -------------------
 .../IgniteCacheTxStoreSessionTest.java          | 285 ++++++++++++
 .../GridCacheLoadOnlyStoreAdapterSelfTest.java  |   1 +
 .../cache/GridCacheAbstractSelfTest.java        |   1 +
 .../cache/GridCacheBasicStoreAbstractTest.java  |   1 +
 ...acheBasicStoreMultithreadedAbstractTest.java |   1 +
 ...idCacheConfigurationConsistencySelfTest.java |   5 +
 ...idCacheGetAndTransformStoreAbstractTest.java |   1 +
 .../GridCacheGroupLockAbstractSelfTest.java     |   1 +
 .../cache/GridCacheLifecycleAwareSelfTest.java  |   1 +
 .../cache/GridCachePartitionedWritesTest.java   |   1 +
 .../cache/GridCacheReloadSelfTest.java          |   1 +
 .../cache/GridCacheStorePutxSelfTest.java       |   1 +
 .../cache/GridCacheSwapReloadSelfTest.java      |   2 +
 .../GridCacheWriteBehindStoreAbstractTest.java  |   1 +
 ...BehindStorePartitionedMultiNodeSelfTest.java |   1 +
 .../IgniteTxExceptionAbstractSelfTest.java      |   1 +
 .../IgniteTxStoreExceptionAbstractSelfTest.java |   1 +
 ...actQueueFailoverDataConsistencySelfTest.java |   1 +
 ...CacheAtomicReferenceApiSelfAbstractTest.java |   2 +
 .../GridCacheClientModesAbstractSelfTest.java   |   2 +
 ...chePartitionedReloadAllAbstractSelfTest.java |   1 +
 .../IgniteTxPreloadAbstractTest.java            |   6 +-
 ...heAbstractTransformWriteThroughSelfTest.java |   1 +
 .../dht/GridCacheColocatedDebugTest.java        |   1 +
 .../near/GridCacheGetStoreErrorSelfTest.java    |   1 +
 .../near/GridCacheNearOneNodeSelfTest.java      |   1 +
 .../GridCacheNearPartitionedClearSelfTest.java  |   1 +
 ...ePartitionedBasicStoreMultiNodeSelfTest.java |   1 +
 .../GridCachePartitionedLoadCacheSelfTest.java  |   1 +
 .../GridCachePartitionedStorePutSelfTest.java   |   1 +
 .../near/GridPartitionedBackupLoadSelfTest.java |   1 +
 .../GridCacheBatchEvictUnswapSelfTest.java      |   1 +
 .../GridCacheEmptyEntriesAbstractSelfTest.java  |   1 +
 .../GridCacheEvictionTouchSelfTest.java         |   1 +
 .../local/GridCacheLocalLoadAllSelfTest.java    |   1 +
 ...ridCacheContinuousQueryAbstractSelfTest.java |   1 +
 .../GridCacheWriteBehindStoreLoadTest.java      |   1 +
 .../loadtests/hashmap/GridCacheTestContext.java |   8 +-
 .../swap/GridSwapEvictAllBenchmark.java         |   1 +
 .../bamboo/GridDataGridTestSuite.java           |   5 +-
 .../cache/GridCacheAbstractQuerySelfTest.java   |   1 +
 .../cache/GridCacheQueryLoadSelfTest.java       |   1 +
 59 files changed, 921 insertions(+), 576 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/25f7984a/modules/clients/src/test/java/org/gridgain/client/integration/GridClientAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/clients/src/test/java/org/gridgain/client/integration/GridClientAbstractSelfTest.java b/modules/clients/src/test/java/org/gridgain/client/integration/GridClientAbstractSelfTest.java
index 6eb6184..9cc1fc2 100644
--- a/modules/clients/src/test/java/org/gridgain/client/integration/GridClientAbstractSelfTest.java
+++ b/modules/clients/src/test/java/org/gridgain/client/integration/GridClientAbstractSelfTest.java
@@ -266,6 +266,7 @@ public abstract class GridClientAbstractSelfTest extends GridCommonAbstractTest
         cfg.setCacheStoreFactory(new FactoryBuilder.SingletonFactory(cacheStore));
         cfg.setWriteThrough(true);
         cfg.setReadThrough(true);
+        cfg.setLoadPreviousValue(true);
 
         cfg.setSwapEnabled(true);
 

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/25f7984a/modules/core/src/main/java/org/apache/ignite/cache/CacheConfiguration.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/cache/CacheConfiguration.java b/modules/core/src/main/java/org/apache/ignite/cache/CacheConfiguration.java
index 54c3444..5a91df8 100644
--- a/modules/core/src/main/java/org/apache/ignite/cache/CacheConfiguration.java
+++ b/modules/core/src/main/java/org/apache/ignite/cache/CacheConfiguration.java
@@ -239,6 +239,9 @@ public class CacheConfiguration extends MutableConfiguration {
     /** */
     private Factory storeFactory;
 
+    /** */
+    private boolean loadPrevVal;
+
     /** Node group resolver. */
     private GridCacheAffinityFunction aff;
 
@@ -383,6 +386,7 @@ public class CacheConfiguration extends MutableConfiguration {
         isWriteThrough = cc.isWriteThrough();
         keepPortableInStore = cc.isKeepPortableInStore();
         listenerConfigurations = cc.listenerConfigurations;
+        loadPrevVal = cc.isLoadPreviousValue();
         offHeapMaxMem = cc.getOffHeapMaxMemory();
         maxConcurrentAsyncOps = cc.getMaxConcurrentAsyncOperations();
         maxQryIterCnt = cc.getMaximumQueryIteratorCount();
@@ -790,6 +794,20 @@ public class CacheConfiguration extends MutableConfiguration {
     }
 
     /**
+     * @return
+     */
+    public boolean isLoadPreviousValue() {
+        return loadPrevVal;
+    }
+
+    /**
+     * @param loadPrevVal
+     */
+    public void setLoadPreviousValue(boolean loadPrevVal) {
+        this.loadPrevVal = loadPrevVal;
+    }
+
+    /**
      * Gets factory for underlying persistent storage for read-through and write-through operations.
      *
      * @return Cache store factory.

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/25f7984a/modules/core/src/main/java/org/apache/ignite/cache/store/CacheStore.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/cache/store/CacheStore.java b/modules/core/src/main/java/org/apache/ignite/cache/store/CacheStore.java
index 0b0832e..d2593ae 100644
--- a/modules/core/src/main/java/org/apache/ignite/cache/store/CacheStore.java
+++ b/modules/core/src/main/java/org/apache/ignite/cache/store/CacheStore.java
@@ -22,6 +22,7 @@ import org.apache.ignite.cache.*;
 import org.apache.ignite.cache.store.jdbc.*;
 import org.apache.ignite.lang.*;
 import org.apache.ignite.portables.*;
+import org.apache.ignite.resources.*;
 import org.apache.ignite.transactions.*;
 import org.gridgain.grid.*;
 import org.gridgain.grid.cache.*;
@@ -125,6 +126,10 @@ public abstract class CacheStore<K, V> implements CacheLoader<K, V>, CacheWriter
     /** */
     private CacheStoreSession ses;
 
+    /** */
+    @IgniteInstanceResource
+    private Ignite ignite;
+
     /**
      * Loads all values from underlying persistent storage. Note that keys are not
      * passed, so it is up to implementation to figure out what to load. This method
@@ -165,4 +170,11 @@ public abstract class CacheStore<K, V> implements CacheLoader<K, V>, CacheWriter
     @Nullable public CacheStoreSession session() {
         return ses;
     }
+
+    /**
+     * @return {@link Ignite} instance.
+     */
+    public Ignite ignite() {
+        return ignite;
+    }
 }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/25f7984a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheAttributes.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheAttributes.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheAttributes.java
index da76ecc..711d375 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheAttributes.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheAttributes.java
@@ -88,9 +88,6 @@ public class GridCacheAttributes implements Externalizable {
     /** Flag indicating whether  query indexing is enabled. */
     private boolean qryIdxEnabled;
 
-    /** Flag indicating whether GridGain should activate read-through/write-through behaviour by default. */
-    private boolean storeEnabled;
-
     /** Flag indicating whether GridGain should use write-behind behaviour for the cache store. */
     private boolean writeBehindEnabled;
 
@@ -154,6 +151,15 @@ public class GridCacheAttributes implements Externalizable {
     /** Transaction Manager lookup class name. */
     private String tmLookupClsName;
 
+    /** Store read-through flag. */
+    private boolean readThrough;
+
+    /** Store write-through flag. */
+    private boolean writeThrough;
+
+    /** Store load previous value flag. */
+    private boolean loadPrevVal;
+
     /**
      * @param cfg Cache configuration.
      * @param store Cache store.
@@ -167,13 +173,14 @@ public class GridCacheAttributes implements Externalizable {
         evictNearSync = cfg.isEvictNearSynchronized();
         evictSync = cfg.isEvictSynchronized();
         indexingSpiName = cfg.getIndexingSpiName();
+        loadPrevVal = cfg.isLoadPreviousValue();
         name = cfg.getName();
         partDistro = GridCacheUtils.distributionMode(cfg);
         preloadBatchSize = cfg.getPreloadBatchSize();
         preloadMode = cfg.getPreloadMode();
         qryIdxEnabled = cfg.isQueryIndexEnabled();
+        readThrough = cfg.isReadThrough();
         seqReserveSize = cfg.getAtomicSequenceReserveSize();
-        storeEnabled = store != null;
         storeValBytes = cfg.isStoreValueBytes();
         swapEnabled = cfg.isSwapEnabled();
         ttl = cfg.getDefaultTimeToLive();
@@ -183,6 +190,7 @@ public class GridCacheAttributes implements Externalizable {
         writeBehindFlushSize = cfg.getWriteBehindFlushSize();
         writeBehindFlushThreadCnt = cfg.getWriteBehindFlushThreadCount();
         writeSyncMode = cfg.getWriteSynchronizationMode();
+        writeThrough = cfg.isWriteThrough();
 
         affMapperClsName = className(cfg.getAffinityMapper());
 
@@ -451,10 +459,24 @@ public class GridCacheAttributes implements Externalizable {
     }
 
     /**
-     * @return Flag indicating whether GridGain should activate read-through/write-through behaviour by default.
+     * @return Flag indicating whether read-through behaviour is enabled.
+     */
+    public boolean readThrough() {
+        return readThrough;
+    }
+
+    /**
+     * @return Flag indicating whether read-through behaviour is enabled.
+     */
+    public boolean writeThrough() {
+        return writeThrough;
+    }
+
+    /**
+     * @return Flag indicating whether old value is loaded from store for cache operation.
      */
-    public boolean storeEnabled() {
-        return storeEnabled;
+    public boolean loadPreviousValue() {
+        return loadPrevVal;
     }
 
     /**
@@ -516,13 +538,14 @@ public class GridCacheAttributes implements Externalizable {
         out.writeBoolean(evictNearSync);
         out.writeBoolean(evictSync);
         U.writeString(out, indexingSpiName);
+        out.writeBoolean(loadPrevVal);
         U.writeString(out, name);
         U.writeEnum0(out, partDistro);
         out.writeInt(preloadBatchSize);
         U.writeEnum0(out, preloadMode);
         out.writeBoolean(qryIdxEnabled);
+        out.writeBoolean(readThrough);
         out.writeInt(seqReserveSize);
-        out.writeBoolean(storeEnabled);
         out.writeBoolean(storeValBytes);
         out.writeBoolean(swapEnabled);
         out.writeLong(ttl);
@@ -532,6 +555,7 @@ public class GridCacheAttributes implements Externalizable {
         out.writeInt(writeBehindFlushSize);
         out.writeInt(writeBehindFlushThreadCnt);
         U.writeEnum0(out, writeSyncMode);
+        out.writeBoolean(writeThrough);
 
         U.writeString(out, affClsName);
         U.writeString(out, affMapperClsName);
@@ -560,13 +584,14 @@ public class GridCacheAttributes implements Externalizable {
         evictNearSync = in.readBoolean();
         evictSync  = in.readBoolean();
         indexingSpiName = U.readString(in);
+        loadPrevVal = in.readBoolean();
         name = U.readString(in);
         partDistro = GridCacheDistributionMode.fromOrdinal(U.readEnumOrdinal0(in));
         preloadBatchSize = in.readInt();
         preloadMode = GridCachePreloadMode.fromOrdinal(U.readEnumOrdinal0(in));
         qryIdxEnabled = in.readBoolean();
+        readThrough = in.readBoolean();
         seqReserveSize = in.readInt();
-        storeEnabled = in.readBoolean();
         storeValBytes = in.readBoolean();
         swapEnabled = in.readBoolean();
         ttl = in.readLong();
@@ -576,6 +601,7 @@ public class GridCacheAttributes implements Externalizable {
         writeBehindFlushSize = in.readInt();
         writeBehindFlushThreadCnt = in.readInt();
         writeSyncMode = GridCacheWriteSynchronizationMode.fromOrdinal(U.readEnumOrdinal0(in));
+        writeThrough = in.readBoolean();
 
         affClsName = U.readString(in);
         affMapperClsName = U.readString(in);

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/25f7984a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheContext.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheContext.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheContext.java
index 7508ac4..b61b4cf 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheContext.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheContext.java
@@ -1406,6 +1406,13 @@ public class GridCacheContext<K, V> implements Externalizable {
     }
 
     /**
+     * @return {@code True} if store read-through mode is enabled.
+     */
+    public boolean loadPreviousValue() {
+        return cacheCfg.isLoadPreviousValue();
+    }
+
+    /**
      * @return {@code True} if store write-through is enabled.
      */
     public boolean writeThrough() {

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/25f7984a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheMapEntry.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheMapEntry.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheMapEntry.java
index d9c84ab..12e7b82 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheMapEntry.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheMapEntry.java
@@ -1426,7 +1426,7 @@ public abstract class GridCacheMapEntry<K, V> implements GridCacheEntryEx<K, V>
 
             GridCacheValueBytes oldBytes = valueBytesUnlocked();
 
-            if (needVal && old == null && cctx.readThrough()) {
+            if (needVal && old == null && (cctx.readThrough() && (op == TRANSFORM || cctx.loadPreviousValue()))) {
                 old = readThrough(null, key, false, CU.<K, V>empty(), subjId, taskName);
 
                 // Detach value before index update.
@@ -1766,7 +1766,7 @@ public abstract class GridCacheMapEntry<K, V> implements GridCacheEntryEx<K, V>
 
             GridCacheValueBytes oldBytes = valueBytesUnlocked();
 
-            if (needVal && old == null && cctx.readThrough()) {
+            if (needVal && old == null && (cctx.readThrough() && (op == TRANSFORM || cctx.loadPreviousValue()))) {
                 old = readThrough(null, key, false, CU.<K, V>empty(), subjId, taskName);
 
                 // Detach value before index update.

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/25f7984a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheProcessor.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheProcessor.java
index e013216..793db4e 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheProcessor.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheProcessor.java
@@ -598,6 +598,8 @@ public class GridCacheProcessor extends GridProcessorAdapter {
 
         Collection<GridCacheAdapter<?, ?>> startSeq = new ArrayList<>(cfgs.length);
 
+        IdentityHashMap<CacheStore, ThreadLocal> sesHolders = new IdentityHashMap<>();
+
         for (int i = 0; i < cfgs.length; i++) {
             CacheConfiguration cfg = new CacheConfiguration(cfgs[i]);
 
@@ -632,7 +634,7 @@ public class GridCacheProcessor extends GridProcessorAdapter {
             GridCacheTtlManager ttlMgr = new GridCacheTtlManager();
             GridCacheDrManager drMgr = ctx.createComponent(GridCacheDrManager.class);
 
-            GridCacheStoreManager storeMgr = new GridCacheStoreManager(ctx, cfgStore, cfg);
+            GridCacheStoreManager storeMgr = new GridCacheStoreManager(ctx, sesHolders, cfgStore, cfg);
 
             GridCacheContext<?, ?> cacheCtx = new GridCacheContext(
                 ctx,
@@ -1116,10 +1118,21 @@ public class GridCacheProcessor extends GridProcessorAdapter {
                         CU.checkAttributeMismatch(log, rmtAttr.cacheName(), rmt, "evictionPolicy", "Eviction policy",
                             locAttr.evictionPolicyClassName(), rmtAttr.evictionPolicyClassName(), true);
 
-                        if (!skipStoreConsistencyCheck(locAttr, rmtAttr))
+                        if (!skipStoreConsistencyCheck(locAttr, rmtAttr)) {
                             CU.checkAttributeMismatch(log, rmtAttr.cacheName(), rmt, "store", "Cache store",
                                 locAttr.storeClassName(), rmtAttr.storeClassName(), true);
 
+                            CU.checkAttributeMismatch(log, rmtAttr.cacheName(), rmt, "readThrough",
+                                "Read through enabled", locAttr.readThrough(), locAttr.readThrough(), true);
+
+                            CU.checkAttributeMismatch(log, rmtAttr.cacheName(), rmt, "writeThrough",
+                                "Write through enabled", locAttr.writeThrough(), locAttr.writeThrough(), true);
+
+                            CU.checkAttributeMismatch(log, rmtAttr.cacheName(), rmt, "loadPreviousValue",
+                                "Load previous value enabled", locAttr.loadPreviousValue(),
+                                locAttr.loadPreviousValue(), true);
+                        }
+
                         CU.checkAttributeMismatch(log, rmtAttr.cacheName(), rmt, "cloner", "Cache cloner",
                             locAttr.clonerClassName(), rmtAttr.clonerClassName(), false);
 

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/25f7984a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheStoreManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheStoreManager.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheStoreManager.java
index aeb9110..7d92f83 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheStoreManager.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheStoreManager.java
@@ -56,10 +56,7 @@ public class GridCacheStoreManager<K, V> extends GridCacheManagerAdapter<K, V> {
     private final CacheStoreBalancingWrapper<K, Object> singleThreadGate;
 
     /** */
-    private final ThreadLocal<SessionData> sesHolder = new ThreadLocal<>();
-
-    /** */
-    private final boolean sesEnabled;
+    private final ThreadLocal<SessionData> sesHolder;
 
     /** */
     private final boolean locStore;
@@ -69,12 +66,15 @@ public class GridCacheStoreManager<K, V> extends GridCacheManagerAdapter<K, V> {
 
     /**
      * @param ctx Kernal context.
+     * @param sesHolders Session holders map to use the same session holder for different managers if they use
+     *        the same store instance.
      * @param cfgStore Store provided in configuration.
      * @param cfg Cache configuration.
      * @throws IgniteCheckedException In case of error.
      */
     @SuppressWarnings("unchecked")
     public GridCacheStoreManager(GridKernalContext ctx,
+        IdentityHashMap<CacheStore, ThreadLocal> sesHolders,
         @Nullable CacheStore<K, Object> cfgStore,
         CacheConfiguration cfg) throws IgniteCheckedException {
         this.cfgStore = cfgStore;
@@ -83,24 +83,34 @@ public class GridCacheStoreManager<K, V> extends GridCacheManagerAdapter<K, V> {
 
         singleThreadGate = store == null ? null : new CacheStoreBalancingWrapper<>(store);
 
-        if (cfgStore != null && cfg.getAtomicityMode() == TRANSACTIONAL) {
+        ThreadLocal<SessionData> sesHolder0 = null;
+
+        if (cfgStore != null) {
             try {
-                Field sesField = CacheStore.class.getDeclaredField("ses");
+                if (!sesHolders.containsKey(cfgStore)) {
+                    sesHolder0 = new ThreadLocal<>();
+
+                    Field sesField = CacheStore.class.getDeclaredField("ses");
 
-                sesField.setAccessible(true);
+                    sesField.setAccessible(true);
 
-                sesField.set(cfgStore, new ThreadLocalSession(sesHolder));
+                    sesField.set(cfgStore, new ThreadLocalSession(sesHolder0));
 
-                sesEnabled = true;
+                    sesHolders.put(cfgStore, sesHolder0);
+                }
+                else
+                    sesHolder0 = sesHolders.get(cfgStore);
             }
             catch (IllegalAccessException | NoSuchFieldException e) {
                 throw new IgniteCheckedException(e);
             }
         }
-        else
-            sesEnabled = true;
+
+        sesHolder = sesHolder0;
 
         locStore = U.hasAnnotation(cfgStore, CacheLocalStore.class);
+
+        assert sesHolder != null || cfgStore == null;
     }
 
     /**
@@ -236,7 +246,7 @@ public class GridCacheStoreManager<K, V> extends GridCacheManagerAdapter<K, V> {
 
             Object val = null;
 
-            boolean ses = initSession(tx);
+            initSession(tx);
 
             try {
                 val = singleThreadGate.load(key);
@@ -251,8 +261,7 @@ public class GridCacheStoreManager<K, V> extends GridCacheManagerAdapter<K, V> {
                 throw new IgniteCheckedException(new CacheLoaderException(e));
             }
             finally {
-                if (ses)
-                    sesHolder.set(null);
+                sesHolder.set(null);
             }
 
             if (log.isDebugEnabled())
@@ -374,7 +383,7 @@ public class GridCacheStoreManager<K, V> extends GridCacheManagerAdapter<K, V> {
             if (log.isDebugEnabled())
                 log.debug("Loading values from store for keys: " + keys0);
 
-            boolean ses = initSession(tx);
+            initSession(tx);
 
             try {
                 CI2<K, Object> c = new CI2<K, Object>() {
@@ -419,8 +428,7 @@ public class GridCacheStoreManager<K, V> extends GridCacheManagerAdapter<K, V> {
                 throw new IgniteCheckedException(new CacheLoaderException(e));
             }
             finally {
-                if (ses)
-                    sesHolder.set(null);
+                sesHolder.set(null);
             }
 
             if (log.isDebugEnabled())
@@ -506,7 +514,7 @@ public class GridCacheStoreManager<K, V> extends GridCacheManagerAdapter<K, V> {
             if (log.isDebugEnabled())
                 log.debug("Storing value in cache store [key=" + key + ", val=" + val + ']');
 
-            boolean ses = initSession(tx);
+            initSession(tx);
 
             try {
                 store.write(new CacheEntryImpl<>(key, locStore ? F.t(val, ver) : val));
@@ -518,8 +526,7 @@ public class GridCacheStoreManager<K, V> extends GridCacheManagerAdapter<K, V> {
                 throw new IgniteCheckedException(e);
             }
             finally {
-                if (ses)
-                    sesHolder.set(null);
+                sesHolder.set(null);
             }
 
             if (log.isDebugEnabled())
@@ -575,7 +582,7 @@ public class GridCacheStoreManager<K, V> extends GridCacheManagerAdapter<K, V> {
                 for (Map.Entry<K, IgniteBiTuple<V, GridCacheVersion>> e : map.entrySet())
                     entries.add(new CacheEntryImpl<>(e.getKey(), locStore ? e.getValue() : e.getValue().get1()));
 
-                boolean ses = initSession(tx);
+                initSession(tx);
 
                 try {
                     store.writeAll(entries);
@@ -596,8 +603,7 @@ public class GridCacheStoreManager<K, V> extends GridCacheManagerAdapter<K, V> {
                     throw new IgniteCheckedException(e);
                 }
                 finally {
-                    if (ses)
-                        sesHolder.set(null);
+                    sesHolder.set(null);
                 }
 
                 if (log.isDebugEnabled())
@@ -628,7 +634,7 @@ public class GridCacheStoreManager<K, V> extends GridCacheManagerAdapter<K, V> {
             if (log.isDebugEnabled())
                 log.debug("Removing value from cache store [key=" + key + ']');
 
-            boolean ses = initSession(tx);
+            initSession(tx);
 
             try {
                 store.delete(key);
@@ -640,8 +646,7 @@ public class GridCacheStoreManager<K, V> extends GridCacheManagerAdapter<K, V> {
                 throw new IgniteCheckedException(e);
             }
             finally {
-                if (ses)
-                    sesHolder.set(null);
+                sesHolder.set(null);
             }
 
             if (log.isDebugEnabled())
@@ -677,7 +682,7 @@ public class GridCacheStoreManager<K, V> extends GridCacheManagerAdapter<K, V> {
             if (log.isDebugEnabled())
                 log.debug("Removing values from cache store [keys=" + keys0 + ']');
 
-            boolean ses = initSession(tx);
+            initSession(tx);
 
             try {
                 store.deleteAll(keys0);
@@ -692,8 +697,7 @@ public class GridCacheStoreManager<K, V> extends GridCacheManagerAdapter<K, V> {
                 throw new IgniteCheckedException(e);
             }
             finally {
-                if (ses)
-                    sesHolder.set(null);
+                sesHolder.set(null);
             }
 
             if (log.isDebugEnabled())
@@ -728,17 +732,15 @@ public class GridCacheStoreManager<K, V> extends GridCacheManagerAdapter<K, V> {
     public void txEnd(IgniteTx tx, boolean commit) throws IgniteCheckedException {
         assert store != null;
 
-        boolean ses = initSession(tx);
+        initSession(tx);
 
         try {
             store.txEnd(commit);
         }
         finally {
-            if (ses) {
-                sesHolder.set(null);
+            sesHolder.set(null);
 
-                ((GridMetadataAware)tx).removeMeta(SES_ATTR);
-            }
+            ((GridMetadataAware)tx).removeMeta(SES_ATTR);
         }
     }
 
@@ -760,12 +762,8 @@ public class GridCacheStoreManager<K, V> extends GridCacheManagerAdapter<K, V> {
 
     /**
      * @param tx Current transaction.
-     * @return {@code True} if thread local session was initialized.
      */
-    private boolean initSession(@Nullable IgniteTx tx) {
-        if (!sesEnabled)
-            return false;
-
+    private void initSession(@Nullable IgniteTx tx) {
         SessionData ses;
 
         if (tx != null) {
@@ -781,8 +779,6 @@ public class GridCacheStoreManager<K, V> extends GridCacheManagerAdapter<K, V> {
             ses = new SessionData(null, cctx.name());
 
         sesHolder.set(ses);
-
-        return true;
     }
 
     /**

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/25f7984a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
index 23f5ce3..186f2a0 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
@@ -1214,7 +1214,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
         assert !ctx.dr().receiveEnabled(); // Cannot update in batches during DR due to possible conflicts.
         assert !req.returnValue() || req.operation() == TRANSFORM; // Should not request return values for putAll.
 
-        if (!F.isEmpty(req.filter())) {
+        if (!F.isEmpty(req.filter()) && ctx.loadPreviousValue()) {
             try {
                 reloadIfNeeded(locked);
             }
@@ -1416,7 +1416,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
                         V old = entry.innerGet(
                              null,
                             /*read swap*/true,
-                            /*read through*/true,
+                            /*read through*/ctx.loadPreviousValue(),
                             /*fail fast*/false,
                             /*unmarshal*/true,
                             /*metrics*/true,
@@ -1450,7 +1450,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
                         V old = entry.innerGet(
                             null,
                             /*read swap*/true,
-                            /*read through*/true,
+                            /*read through*/ctx.loadPreviousValue(),
                             /*fail fast*/false,
                             /*unmarshal*/true,
                             /*metrics*/true,

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/25f7984a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/local/atomic/GridLocalAtomicCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/local/atomic/GridLocalAtomicCache.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/local/atomic/GridLocalAtomicCache.java
index 978072e..c4080d0 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/local/atomic/GridLocalAtomicCache.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/local/atomic/GridLocalAtomicCache.java
@@ -1148,7 +1148,7 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
                         if (intercept) {
                             V old = entry.innerGet(null,
                                 /*swap*/true,
-                                /*read-through*/true,
+                                /*read-through*/ctx.loadPreviousValue(),
                                 /*fail-fast*/false,
                                 /*unmarshal*/true,
                                 /**update-metrics*/true,
@@ -1179,7 +1179,7 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
                         if (intercept) {
                             V old = entry.innerGet(null,
                                 /*swap*/true,
-                                /*read-through*/true,
+                                /*read-through*/ctx.loadPreviousValue(),
                                 /*fail-fast*/false,
                                 /*unmarshal*/true,
                                 /**update-metrics*/true,

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/25f7984a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/transactions/IgniteTxLocalAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/transactions/IgniteTxLocalAdapter.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/transactions/IgniteTxLocalAdapter.java
index 123072e..11cd1af 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/transactions/IgniteTxLocalAdapter.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/transactions/IgniteTxLocalAdapter.java
@@ -1942,7 +1942,8 @@ public abstract class IgniteTxLocalAdapter<K, V> extends IgniteTxAdapter<K, V>
 
                             V old = null;
 
-                            boolean readThrough = !F.isEmptyOrNulls(filter) && !F.isAlwaysTrue(filter);
+                            boolean readThrough =
+                                cacheCtx.loadPreviousValue() && !F.isEmptyOrNulls(filter) && !F.isAlwaysTrue(filter);
 
                             if (optimistic()) {
                                 try {
@@ -2275,7 +2276,7 @@ public abstract class IgniteTxLocalAdapter<K, V> extends IgniteTxAdapter<K, V>
                                 if (!hasPrevVal)
                                     v = cached.innerGet(this,
                                         /*swap*/true,
-                                        /*read-through*/true,
+                                        /*read-through*/invoke || cacheCtx.loadPreviousValue(),
                                         /*failFast*/false,
                                         /*unmarshal*/true,
                                         /*metrics*/!invoke,

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/25f7984a/modules/core/src/test/java/org/apache/ignite/cache/store/jdbc/GridCacheJdbcBlobStoreMultithreadedSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/cache/store/jdbc/GridCacheJdbcBlobStoreMultithreadedSelfTest.java b/modules/core/src/test/java/org/apache/ignite/cache/store/jdbc/GridCacheJdbcBlobStoreMultithreadedSelfTest.java
index 8075172..01ae3a6 100644
--- a/modules/core/src/test/java/org/apache/ignite/cache/store/jdbc/GridCacheJdbcBlobStoreMultithreadedSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/cache/store/jdbc/GridCacheJdbcBlobStoreMultithreadedSelfTest.java
@@ -103,6 +103,7 @@ public class GridCacheJdbcBlobStoreMultithreadedSelfTest extends GridCommonAbstr
         cc.setCacheStoreFactory(new FactoryBuilder.SingletonFactory(store()));
         cc.setReadThrough(true);
         cc.setWriteThrough(true);
+        cc.setLoadPreviousValue(true);
 
         c.setCacheConfiguration(cc);
 

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/25f7984a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheAbstractTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheAbstractTest.java
index bf8206c..ac51490 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheAbstractTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheAbstractTest.java
@@ -132,6 +132,7 @@ public abstract class IgniteCacheAbstractTest extends GridCommonAbstractTest {
             cfg.setCacheStoreFactory(new FactoryBuilder.SingletonFactory(store));
             cfg.setReadThrough(true);
             cfg.setWriteThrough(true);
+            cfg.setLoadPreviousValue(true);
         }
 
         if (cacheMode() == PARTITIONED)

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/25f7984a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/integration/IgniteCacheAtomicStoreSessionTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/integration/IgniteCacheAtomicStoreSessionTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/integration/IgniteCacheAtomicStoreSessionTest.java
new file mode 100644
index 0000000..4488e04
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/integration/IgniteCacheAtomicStoreSessionTest.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.internal.processors.cache.integration;
+
+import org.gridgain.grid.cache.*;
+
+import static org.gridgain.grid.cache.GridCacheAtomicWriteOrderMode.*;
+import static org.gridgain.grid.cache.GridCacheAtomicityMode.*;
+import static org.gridgain.grid.cache.GridCacheDistributionMode.*;
+import static org.gridgain.grid.cache.GridCacheMode.*;
+
+/**
+ *
+ */
+public class IgniteCacheAtomicStoreSessionTest extends IgniteCacheStoreSessionAbstractTest {
+    /** {@inheritDoc} */
+    @Override protected int gridCount() {
+        return 3;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected GridCacheMode cacheMode() {
+        return PARTITIONED;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected GridCacheAtomicityMode atomicityMode() {
+        return ATOMIC;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected GridCacheDistributionMode distributionMode() {
+        return PARTITIONED_ONLY;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected GridCacheAtomicWriteOrderMode atomicWriteOrderMode() {
+        return PRIMARY;
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/25f7984a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/integration/IgniteCacheLoaderWriterAbstractTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/integration/IgniteCacheLoaderWriterAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/integration/IgniteCacheLoaderWriterAbstractTest.java
index 7564f06..8bff838 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/integration/IgniteCacheLoaderWriterAbstractTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/integration/IgniteCacheLoaderWriterAbstractTest.java
@@ -28,8 +28,6 @@ import javax.cache.processor.*;
 import java.util.*;
 import java.util.concurrent.atomic.*;
 
-import static org.gridgain.grid.cache.GridCacheAtomicityMode.*;
-
 /**
  *
  */
@@ -72,88 +70,75 @@ public abstract class IgniteCacheLoaderWriterAbstractTest extends IgniteCacheAbs
         storeMap.clear();
     }
 
-    protected boolean putFromPrimary() {
-        return atomicityMode() == ATOMIC;
-    }
-
     /**
      * @throws Exception If failed.
      */
     public void testLoaderWriter() throws Exception {
-        final Object key = Integer.MAX_VALUE;
-
-        for (int i = 0; i < gridCount(); i++) {
-            log.info("Test with grid: " + i);
+        IgniteCache<Object, Object> cache = jcache(0);
 
-            storeMap.clear();
+        Object key = primaryKey(cache);
 
-            ldrCallCnt.set(0);
-            writerCallCnt.set(0);
+        assertNull(cache.get(key));
 
-            IgniteCache<Object, Object> cache = jcache(i);
+        checkCalls(1, 0);
 
-            assertNull(cache.get(key));
+        storeMap.put(key, "test");
 
-            checkCalls(1, 0);
+        assertEquals("test", cache.get(key));
 
-            storeMap.put(key, "test");
+        checkCalls(1, 0);
 
-            assertEquals("test", cache.get(key));
+        assertTrue(storeMap.containsKey(key));
 
-            checkCalls(2, 0);
+        cache.remove(key);
 
-            assertTrue(storeMap.containsKey(key));
+        checkCalls(0, 1);
 
-            cache.remove(key);
-
-            checkCalls(2, 1);
-
-            assertFalse(storeMap.containsKey(key));
+        assertFalse(storeMap.containsKey(key));
 
-            assertNull(cache.get(key));
+        assertNull(cache.get(key));
 
-            checkCalls(3, 1);
+        checkCalls(1, 0);
 
-            cache.put(key, "test1");
+        cache.put(key, "test1");
 
-            checkCalls(3, 2);
+        checkCalls(0, 1);
 
-            assertEquals("test1", storeMap.get(key));
+        assertEquals("test1", storeMap.get(key));
 
-            assertEquals("test1", cache.get(key));
+        assertEquals("test1", cache.get(key));
 
-            checkCalls(3, 2);
+        checkCalls(0, 0);
 
-            cache.invoke(key, new EntryProcessor<Object, Object, Object>() {
-                @Override public Object process(MutableEntry<Object, Object> e, Object... args) {
-                    e.setValue("test2");
+        cache.invoke(key, new EntryProcessor<Object, Object, Object>() {
+            @Override public Object process(MutableEntry<Object, Object> e, Object... args) {
+                e.setValue("test2");
 
-                    return null;
-                }
-            });
+                return null;
+            }
+        });
 
-            checkCalls(3, 3);
+        checkCalls(0, 1);
 
-            assertEquals("test2", storeMap.get(key));
+        assertEquals("test2", storeMap.get(key));
 
-            assertEquals("test2", cache.get(key));
+        assertEquals("test2", cache.get(key));
 
-            checkCalls(3, 3);
+        checkCalls(0, 0);
 
-            cache.invoke(key, new EntryProcessor<Object, Object, Object>() {
-                @Override public Object process(MutableEntry<Object, Object> e, Object... args) {
-                    e.remove();
+        cache.invoke(key, new EntryProcessor<Object, Object, Object>() {
+            @Override public Object process(MutableEntry<Object, Object> e, Object... args) {
+                e.remove();
 
-                    return null;
-                }
-            });
+                return null;
+            }
+        });
 
-            checkCalls(3, 4);
+        checkCalls(0, 1);
 
-            assertFalse(storeMap.containsKey(key));
+        assertFalse(storeMap.containsKey(key));
 
-            assertNull(cache.get(key));
-        }
+        assertNull(cache.get(key));
     }
 
     /**
@@ -162,43 +147,63 @@ public abstract class IgniteCacheLoaderWriterAbstractTest extends IgniteCacheAbs
     public void testLoaderWriterBulk() throws Exception {
         Map<Object, Object> vals = new HashMap<>();
 
-        for (int i = 0; i < 100; i++)
-            vals.put(i, i);
+        IgniteCache<Object, Object> cache = jcache(0);
 
-        for (int i = 0; i < gridCount(); i++) {
-            log.info("Test with grid: " + i);
+        for (Object key : primaryKeys(cache, 100, 0))
+            vals.put(key, key);
 
-            storeMap.clear();
+        assertTrue(cache.getAll(vals.keySet()).isEmpty());
 
-            ldrCallCnt.set(0);
-            writerCallCnt.set(0);
+        checkCalls(1, 0);
 
-            IgniteCache<Object, Object> cache = jcache(i);
+        storeMap.putAll(vals);
 
-            assertTrue(cache.getAll(vals.keySet()).isEmpty());
+        assertEquals(vals, cache.getAll(vals.keySet()));
 
-            int expLoads = gridCount();
+        checkCalls(1, 0);
 
-            checkCalls(expLoads, 0);
+        for (Object key : vals.keySet())
+            assertEquals(key, storeMap.get(key));
 
-            storeMap.putAll(vals);
+        cache.removeAll(vals.keySet());
 
-            assertEquals(vals, cache.getAll(vals.keySet()));
+        checkCalls(0, 1);
 
-            expLoads += gridCount();
+        for (Object key : vals.keySet())
+            assertFalse(storeMap.containsKey(key));
 
-            checkCalls(expLoads, 0);
+        cache.putAll(vals);
 
-            for (Object key : vals.keySet())
-                assertTrue(storeMap.contains(key));
+        checkCalls(0, 1);
 
-            cache.removeAll(vals.keySet());
+        for (Object key : vals.keySet())
+            assertEquals(key, storeMap.get(key));
 
-            checkCalls(expLoads, gridCount());
+        cache.invokeAll(vals.keySet(), new EntryProcessor<Object, Object, Object>() {
+            @Override public Object process(MutableEntry<Object, Object> entry, Object... args) {
+                entry.setValue("test1");
 
-            for (Object key : vals.keySet())
-                assertFalse(storeMap.containsKey(key));
-        }
+                return null;
+            }
+        });
+
+        checkCalls(0, 1);
+
+        for (Object key : vals.keySet())
+            assertEquals("test1", storeMap.get(key));
+
+        cache.invokeAll(vals.keySet(), new EntryProcessor<Object, Object, Object>() {
+            @Override public Object process(MutableEntry<Object, Object> entry, Object... args) {
+                entry.remove();
+
+                return null;
+            }
+        });
+
+        checkCalls(0, 1);
+
+        for (Object key : vals.keySet())
+            assertFalse(storeMap.containsKey(key));
     }
 
     /**
@@ -207,8 +212,10 @@ public abstract class IgniteCacheLoaderWriterAbstractTest extends IgniteCacheAbs
      */
     private void checkCalls(int expLdr, int expWriter) {
         assertEquals(expLdr, ldrCallCnt.get());
-
         assertEquals(expWriter, writerCallCnt.get());
+
+        ldrCallCnt.set(0);
+        writerCallCnt.set(0);
     }
 
     /**

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/25f7984a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/integration/IgniteCacheStoreSessionAbstractTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/integration/IgniteCacheStoreSessionAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/integration/IgniteCacheStoreSessionAbstractTest.java
new file mode 100644
index 0000000..87de0bf
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/integration/IgniteCacheStoreSessionAbstractTest.java
@@ -0,0 +1,303 @@
+package org.apache.ignite.internal.processors.cache.integration;
+
+import org.apache.ignite.*;
+import org.apache.ignite.cache.*;
+import org.apache.ignite.cache.store.*;
+import org.apache.ignite.configuration.*;
+import org.apache.ignite.internal.processors.cache.*;
+import org.apache.ignite.lang.*;
+import org.gridgain.grid.util.typedef.*;
+import org.jetbrains.annotations.*;
+
+import javax.cache.*;
+import javax.cache.configuration.*;
+import javax.cache.integration.*;
+import javax.cache.processor.*;
+import java.util.*;
+
+import static org.gridgain.grid.cache.GridCacheAtomicityMode.*;
+
+/**
+ *
+ */
+public abstract class IgniteCacheStoreSessionAbstractTest extends IgniteCacheAbstractTest {
+    /** */
+    protected static volatile List<ExpectedData> expData;
+
+    /** */
+    protected static final String CACHE_NAME1 = "cache1";
+
+    /** {@inheritDoc} */
+    @SuppressWarnings("unchecked")
+    @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+        IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+        TestStore store = new TestStore(); // Use the same store instance for both caches.
+
+        assert cfg.getCacheConfiguration().length == 1;
+
+        CacheConfiguration ccfg0 = cfg.getCacheConfiguration()[0];
+
+        ccfg0.setReadThrough(true);
+        ccfg0.setWriteThrough(true);
+
+        ccfg0.setCacheStoreFactory(new FactoryBuilder.SingletonFactory(store));
+
+        CacheConfiguration ccfg1 = cacheConfiguration(gridName);
+
+        ccfg1.setReadThrough(true);
+        ccfg1.setWriteThrough(true);
+
+        ccfg1.setName(CACHE_NAME1);
+
+        ccfg1.setCacheStoreFactory(new FactoryBuilder.SingletonFactory(store));
+
+        cfg.setCacheConfiguration(ccfg0, ccfg1);
+
+        return cfg;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void afterTestsStopped() throws Exception {
+        super.afterTestsStopped();
+
+        expData = null;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void beforeTestsStarted() throws Exception {
+        expData = Collections.synchronizedList(new ArrayList<ExpectedData>());
+
+        super.beforeTestsStarted();
+    }
+
+    /**
+     * @param cache Cache.
+     * @param cnt Keys count.
+     * @return Keys.
+     * @throws Exception If failed.
+     */
+    protected List<Integer> testKeys(IgniteCache cache, int cnt) throws Exception {
+        return primaryKeys(cache, cnt, 0);
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testStoreSession() throws Exception {
+        assertNull(jcache(0).getName());
+
+        assertEquals(CACHE_NAME1, ignite(0).jcache(CACHE_NAME1).getName());
+
+        testStoreSession(jcache(0));
+
+        testStoreSession(ignite(0).jcache(CACHE_NAME1));
+    }
+
+    /**
+     * @param cache Cache.
+     * @throws Exception If failed.
+     */
+    private void testStoreSession(IgniteCache<Object, Object> cache) throws Exception {
+        Set<Integer> keys = new HashSet<>(primaryKeys(cache, 3, 100_000));
+
+        Integer key = keys.iterator().next();
+
+        boolean tx = atomicityMode() == TRANSACTIONAL;
+
+        expData.add(new ExpectedData(false, "load", new HashMap<>(), cache.getName()));
+
+        assertEquals(key, cache.get(key));
+
+        assertTrue(expData.isEmpty());
+
+        expData.add(new ExpectedData(false, "loadAll", new HashMap<>(), cache.getName()));
+
+        assertEquals(3, cache.getAll(keys).size());
+
+        assertTrue(expData.isEmpty());
+
+        expectedData(tx, "write", cache.getName());
+
+        cache.put(key, key);
+
+        assertTrue(expData.isEmpty());
+
+        expectedData(tx, "write", cache.getName());
+
+        cache.invoke(key, new EntryProcessor<Object, Object, Object>() {
+            @Override public Object process(MutableEntry<Object, Object> e, Object... args) {
+                e.setValue("val1");
+
+                return null;
+            }
+        });
+
+        assertTrue(expData.isEmpty());
+
+        expectedData(tx, "delete", cache.getName());
+
+        cache.remove(key);
+
+        assertTrue(expData.isEmpty());
+
+        Map<Object, Object> vals = new HashMap<>();
+
+        for (Object key0 : keys)
+            vals.put(key0, key0);
+
+        expectedData(tx, "writeAll", cache.getName());
+
+        cache.putAll(vals);
+
+        assertTrue(expData.isEmpty());
+
+        expectedData(tx, "deleteAll", cache.getName());
+
+        cache.removeAll(keys);
+
+        assertTrue(expData.isEmpty());
+    }
+
+    /**
+     * @param tx {@code True} is transaction is expected.
+     * @param expMtd Expected method.
+     * @param expCacheName Expected cache name.
+     */
+    private void expectedData(boolean tx, String expMtd, String expCacheName) {
+        expData.add(new ExpectedData(tx, expMtd, new HashMap<>(), expCacheName));
+
+        if (tx)
+            expData.add(new ExpectedData(true, "txEnd", F.<Object, Object>asMap(0, expMtd), expCacheName));
+    }
+
+    /**
+     *
+     */
+    static class ExpectedData {
+        /** */
+        private final boolean tx;
+
+        /** */
+        private final String expMtd;
+
+        /** */
+        private  final Map<Object, Object> expProps;
+
+        /** */
+        private final String expCacheName;
+
+        /**
+         * @param tx {@code True} if transaction is enabled.
+         * @param expMtd Expected method.
+         * @param expProps Expected properties.
+         * @param expCacheName Expected cache name.
+         */
+        public ExpectedData(boolean tx, String expMtd, Map<Object, Object> expProps, String expCacheName) {
+            this.tx = tx;
+            this.expMtd = expMtd;
+            this.expProps = expProps;
+            this.expCacheName = expCacheName;
+        }
+    }
+
+    /**
+     *
+     */
+    private class TestStore extends CacheStore<Object, Object> {
+        /** {@inheritDoc} */
+        @Override public void loadCache(IgniteBiInClosure<Object, Object> clo, @Nullable Object... args) {
+            fail();
+        }
+
+        /** {@inheritDoc} */
+        @Override public void txEnd(boolean commit) throws CacheWriterException {
+            log.info("Tx end [commit=" + commit + ", tx=" + session().transaction() + ']');
+
+            checkSession("txEnd");
+        }
+
+        /** {@inheritDoc} */
+        @Override public Object load(Object key) throws CacheLoaderException {
+            log.info("Load [key=" + key + ", tx=" + session().transaction() + ']');
+
+            checkSession("load");
+
+            return key;
+        }
+
+        /** {@inheritDoc} */
+        @Override public Map<Object, Object> loadAll(Iterable<?> keys) throws CacheLoaderException {
+            log.info("LoadAll [keys=" + keys + ", tx=" + session().transaction() + ']');
+
+            checkSession("loadAll");
+
+            Map<Object, Object> loaded = new HashMap<>();
+
+            for (Object key : keys)
+                loaded.put(key, key);
+
+            return loaded;
+        }
+
+        /** {@inheritDoc} */
+        @Override public void write(Cache.Entry<?, ?> entry) throws CacheWriterException {
+            log.info("Write [write=" + entry + ", tx=" + session().transaction() + ']');
+
+            checkSession("write");
+        }
+
+        /** {@inheritDoc} */
+        @Override public void writeAll(Collection<Cache.Entry<?, ?>> entries) throws CacheWriterException {
+            log.info("WriteAll: [writeAll=" + entries + ", tx=" + session().transaction() + ']');
+
+            checkSession("writeAll");
+        }
+
+        /** {@inheritDoc} */
+        @Override public void delete(Object key) throws CacheWriterException {
+            log.info("Delete [key=" + key + ", tx=" + session().transaction() + ']');
+
+            checkSession("delete");
+        }
+
+        /** {@inheritDoc} */
+        @Override public void deleteAll(Collection<?> keys) throws CacheWriterException {
+            log.info("DeleteAll [keys=" + keys + ", tx=" + session().transaction() + ']');
+
+            checkSession("deleteAll");
+        }
+
+        /**
+         * @param mtd Called stored method.
+         */
+        private void checkSession(String mtd) {
+            assertNotNull(ignite());
+
+            assertFalse(expData.isEmpty());
+
+            ExpectedData exp = expData.remove(0);
+
+            assertEquals(exp.expMtd, mtd);
+
+            CacheStoreSession ses = session();
+
+            assertNotNull(ses);
+
+            if (exp.tx)
+                assertNotNull(ses.transaction());
+            else
+                assertNull(ses.transaction());
+
+            Map<Object, Object> props = ses.properties();
+
+            assertNotNull(props);
+
+            assertEquals(exp.expProps, props);
+
+            props.put(props.size(), mtd);
+
+            assertEquals(exp.expCacheName, ses.cacheName());
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/25f7984a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/integration/IgniteCacheStoreSessionTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/integration/IgniteCacheStoreSessionTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/integration/IgniteCacheStoreSessionTest.java
deleted file mode 100644
index d333456..0000000
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/integration/IgniteCacheStoreSessionTest.java
+++ /dev/null
@@ -1,435 +0,0 @@
-package org.apache.ignite.internal.processors.cache.integration;
-
-import org.apache.ignite.*;
-import org.apache.ignite.cache.*;
-import org.apache.ignite.cache.store.*;
-import org.apache.ignite.configuration.*;
-import org.apache.ignite.internal.processors.cache.*;
-import org.apache.ignite.lang.*;
-import org.apache.ignite.transactions.*;
-import org.gridgain.grid.cache.*;
-import org.gridgain.grid.util.typedef.*;
-import org.jetbrains.annotations.*;
-
-import javax.cache.*;
-import javax.cache.integration.*;
-import java.util.*;
-
-import static org.apache.ignite.transactions.IgniteTxConcurrency.*;
-import static org.apache.ignite.transactions.IgniteTxIsolation.*;
-import static org.gridgain.grid.cache.GridCacheAtomicityMode.*;
-import static org.gridgain.grid.cache.GridCacheDistributionMode.*;
-import static org.gridgain.grid.cache.GridCacheMode.*;
-
-/**
- *
- */
-public class IgniteCacheStoreSessionTest extends IgniteCacheAbstractTest {
-    /** */
-    private static volatile List<ExpectedData> expData;
-
-    /** */
-    private static final String CACHE_NAME1 = "cache1";
-
-    private TestStore store = new TestStore();
-
-    /** {@inheritDoc} */
-    @Override protected int gridCount() {
-        return 3;
-    }
-
-    /** {@inheritDoc} */
-    @Override protected GridCacheMode cacheMode() {
-        return PARTITIONED;
-    }
-
-    /** {@inheritDoc} */
-    @Override protected GridCacheAtomicityMode atomicityMode() {
-        return TRANSACTIONAL;
-    }
-
-    /** {@inheritDoc} */
-    @Override protected GridCacheDistributionMode distributionMode() {
-        return PARTITIONED_ONLY;
-    }
-
-    /** {@inheritDoc} */
-    @Override protected CacheStore<?, ?> cacheStore() {
-        return store;
-    }
-
-    /** {@inheritDoc} */
-    @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
-        IgniteConfiguration cfg = super.getConfiguration(gridName);
-
-        assert cfg.getCacheConfiguration().length == 1;
-
-        CacheConfiguration ccfg0 = cfg.getCacheConfiguration()[0];
-
-        CacheConfiguration ccfg1 = cacheConfiguration(gridName);
-
-        ccfg1.setName(CACHE_NAME1);
-
-        cfg.setCacheConfiguration(ccfg0, ccfg1);
-
-        return cfg;
-    }
-
-    /** {@inheritDoc} */
-    @Override protected void afterTestsStopped() throws Exception {
-        super.afterTestsStopped();
-
-        expData = null;
-    }
-
-    /** {@inheritDoc} */
-    @Override protected void beforeTestsStarted() throws Exception {
-        expData = Collections.synchronizedList(new ArrayList<ExpectedData>());
-
-        super.beforeTestsStarted();
-    }
-
-    /**
-     * @throws Exception If failed.
-     */
-    public void testStoreSession() throws Exception {
-        testTxPut(jcache(0), null, null);
-
-        testTxPut(ignite(0).jcache(CACHE_NAME1), null, null);
-
-        testTxRemove(null, null);
-
-        testTxPutRemove(null, null);
-
-        for (IgniteTxConcurrency concurrency : F.asList(PESSIMISTIC)) {
-            for (IgniteTxIsolation isolation : F.asList(REPEATABLE_READ)) {
-                testTxPut(jcache(0), concurrency, isolation);
-
-                testTxRemove(concurrency, isolation);
-
-                testTxPutRemove(concurrency, isolation);
-            }
-        }
-    }
-
-    /**
-     * @param cnt Keys count.
-     * @return Keys.
-     * @throws Exception If failed.
-     */
-    private List<Integer> testKeys(IgniteCache cache, int cnt) throws Exception {
-        return primaryKeys(cache, cnt, 0);
-    }
-
-    /**
-     * @param concurrency Concurrency mode.
-     * @param isolation Isolation mode.
-     * @throws Exception If failed.
-     */
-    private void testTxPutRemove(IgniteTxConcurrency concurrency, IgniteTxIsolation isolation) throws Exception {
-        log.info("Test tx put/remove [concurrency=" + concurrency + ", isolation=" + isolation + ']');
-
-        IgniteCache<Integer, Integer> cache = jcache(0);
-
-        List<Integer> keys = testKeys(cache, 3);
-
-        Integer key1 = keys.get(0);
-        Integer key2 = keys.get(1);
-        Integer key3 = keys.get(2);
-
-        try (IgniteTx tx = startTx(concurrency, isolation)) {
-            log.info("Do tx put1.");
-
-            cache.put(key1, key1);
-
-            log.info("Do tx put2.");
-
-            cache.put(key2, key2);
-
-            log.info("Do tx remove.");
-
-            cache.remove(key3);
-
-            expData.add(new ExpectedData("writeAll", new HashMap<>(), null));
-            expData.add(new ExpectedData("delete", F.<Object, Object>asMap(0, "writeAll"), null));
-            expData.add(new ExpectedData("txEnd", F.<Object, Object>asMap(0, "writeAll", 1, "delete"), null));
-
-            log.info("Do tx commit.");
-
-            tx.commit();
-        }
-
-        assertEquals(0, expData.size());
-    }
-
-    /**
-     * @param concurrency Concurrency mode.
-     * @param isolation Isolation mode.
-     * @throws Exception If failed.
-     */
-    private void testTxPut(IgniteCache<Object, Object> cache,
-        IgniteTxConcurrency concurrency,
-        IgniteTxIsolation isolation) throws Exception {
-        log.info("Test tx put [concurrency=" + concurrency + ", isolation=" + isolation + ']');
-
-        List<Integer> keys = testKeys(cache, 3);
-
-        Integer key1 = keys.get(0);
-
-        try (IgniteTx tx = startTx(concurrency, isolation)) {
-            log.info("Do tx get.");
-
-            cache.get(key1);
-
-            log.info("Do tx put.");
-
-            cache.put(key1, key1);
-
-            expData.add(new ExpectedData("write", new HashMap<>(), cache.getName()));
-            expData.add(new ExpectedData("txEnd", F.<Object, Object>asMap(0, "write"), cache.getName()));
-
-            log.info("Do tx commit.");
-
-            tx.commit();
-        }
-
-        assertEquals(0, expData.size());
-
-        Integer key2 = keys.get(1);
-        Integer key3 = keys.get(2);
-
-        try (IgniteTx tx = startTx(concurrency, isolation);) {
-            log.info("Do tx put1.");
-
-            cache.put(key2, key2);
-
-            log.info("Do tx put2.");
-
-            cache.put(key3, key3);
-
-            expData.add(new ExpectedData("writeAll", new HashMap<>(), cache.getName()));
-            expData.add(new ExpectedData("txEnd", F.<Object, Object>asMap(0, "writeAll"), cache.getName()));
-
-            log.info("Do tx commit.");
-
-            tx.commit();
-        }
-
-        assertEquals(0, expData.size());
-    }
-
-    /**
-     * @param concurrency Concurrency mode.
-     * @param isolation Isolation mode.
-     * @throws Exception If failed.
-     */
-    private void testTxRemove(IgniteTxConcurrency concurrency, IgniteTxIsolation isolation) throws Exception {
-        log.info("Test tx remove [concurrency=" + concurrency + ", isolation=" + isolation + ']');
-
-        IgniteCache<Integer, Integer> cache = jcache(0);
-
-        List<Integer> keys = testKeys(cache, 3);
-
-        Integer key1 = keys.get(0);
-
-        try (IgniteTx tx = startTx(concurrency, isolation)) {
-            log.info("Do tx get.");
-
-            cache.get(key1);
-
-            log.info("Do tx remove.");
-
-            cache.remove(key1, key1);
-
-            expData.add(new ExpectedData("delete", new HashMap<>(), null));
-            expData.add(new ExpectedData("txEnd", F.<Object, Object>asMap(0, "delete"), null));
-
-            log.info("Do tx commit.");
-
-            tx.commit();
-        }
-
-        assertEquals(0, expData.size());
-
-        Integer key2 = keys.get(1);
-        Integer key3 = keys.get(2);
-
-        try (IgniteTx tx = startTx(concurrency, isolation);) {
-            log.info("Do tx remove1.");
-
-            cache.remove(key2, key2);
-
-            log.info("Do tx remove2.");
-
-            cache.remove(key3, key3);
-
-            expData.add(new ExpectedData("deleteAll", new HashMap<>(), null));
-            expData.add(new ExpectedData("txEnd", F.<Object, Object>asMap(0, "deleteAll"), null));
-
-            log.info("Do tx commit.");
-
-            tx.commit();
-        }
-
-        assertEquals(0, expData.size());
-    }
-
-    /**
-     * @param concurrency Concurrency mode.
-     * @param isolation Isolation mode.
-     * @return Transaction.
-     */
-    private IgniteTx startTx(IgniteTxConcurrency concurrency, IgniteTxIsolation isolation) {
-        IgniteTransactions txs = ignite(0).transactions();
-
-        if (concurrency == null)
-            return txs.txStart();
-
-        return txs.txStart(concurrency, isolation);
-    }
-
-    /**
-     * @throws Exception If failed.
-     */
-    public void testSessionCrossCacheTx() throws Exception {
-        IgniteCache<Object, Object> cache0 = ignite(0).jcache(null);
-
-        IgniteCache<Object, Object> cache1 = ignite(0).jcache(CACHE_NAME1);
-
-        Integer key1 = primaryKey(cache0);
-        Integer key2 = primaryKeys(cache1, 1, key1 + 1).get(0);
-
-        try (IgniteTx tx = startTx(null, null)) {
-            cache1.put(key2, key2);
-
-            cache0.put(key1, key1);
-
-            expData.add(new ExpectedData("writeAll", new HashMap<>(), null));
-            expData.add(new ExpectedData("txEnd", F.<Object, Object>asMap(0, "writeAll"), null));
-
-            tx.commit();
-        }
-
-        assertEquals(0, expData.size());
-    }
-
-    /**
-     *
-     */
-    static class ExpectedData {
-        /** */
-        private final String expMtd;
-
-        /** */
-        private  final Map<Object, Object> expProps;
-
-        /** */
-        private final String expCacheName;
-
-        /**
-         * @param expMtd Expected method.
-         * @param expProps Expected properties.
-         * @param expCacheName Expected cache name.
-         */
-        public ExpectedData(String expMtd, Map<Object, Object> expProps, String expCacheName) {
-            this.expMtd = expMtd;
-            this.expProps = expProps;
-            this.expCacheName = expCacheName;
-        }
-    }
-
-    /**
-     *
-     */
-    private class TestStore extends CacheStore<Object, Object> {
-        /** {@inheritDoc} */
-        @Override public void loadCache(IgniteBiInClosure<Object, Object> clo, @Nullable Object... args) {
-            fail();
-        }
-
-        /** {@inheritDoc} */
-        @Override public void txEnd(boolean commit) throws CacheWriterException {
-            log.info("Tx end [commit=" + commit + ", tx=" + session().transaction() + ']');
-
-            checkSession("txEnd");
-        }
-
-        /** {@inheritDoc} */
-        @Override public Object load(Object key) throws CacheLoaderException {
-            log.info("Load [key=" + key + ", tx=" + session().transaction() + ']');
-
-            checkSession("load");
-
-            return key;
-        }
-
-        /** {@inheritDoc} */
-        @Override public Map<Object, Object> loadAll(Iterable<?> keys) throws CacheLoaderException {
-            log.info("LoadAll [keys=" + keys + ", tx=" + session().transaction() + ']');
-
-            checkSession("loadAll");
-
-            Map<Object, Object> loaded = new HashMap<>();
-
-            for (Object key : keys)
-                loaded.put(key, key);
-
-            return loaded;
-        }
-
-        /** {@inheritDoc} */
-        @Override public void write(Cache.Entry<?, ?> entry) throws CacheWriterException {
-            log.info("Write [write=" + entry + ", tx=" + session().transaction() + ']');
-
-            checkSession("write");
-        }
-
-        /** {@inheritDoc} */
-        @Override public void writeAll(Collection<Cache.Entry<?, ?>> entries) throws CacheWriterException {
-            log.info("WriteAll: [writeAll=" + entries + ", tx=" + session().transaction() + ']');
-
-            checkSession("writeAll");
-        }
-
-        /** {@inheritDoc} */
-        @Override public void delete(Object key) throws CacheWriterException {
-            log.info("Delete [key=" + key + ", tx=" + session().transaction() + ']');
-
-            checkSession("delete");
-        }
-
-        /** {@inheritDoc} */
-        @Override public void deleteAll(Collection<?> keys) throws CacheWriterException {
-            log.info("DeleteAll [keys=" + keys + ", tx=" + session().transaction() + ']');
-
-            checkSession("deleteAll");
-        }
-
-        /**
-         * @param mtd Called stored method.
-         */
-        private void checkSession(String mtd) {
-            assertFalse(expData.isEmpty());
-
-            ExpectedData exp = expData.remove(0);
-
-            assertEquals(exp.expMtd, mtd);
-
-            CacheStoreSession ses = session();
-
-            assertNotNull(ses);
-
-            assertNotNull(ses.transaction());
-
-            Map<Object, Object> props = ses.properties();
-
-            assertNotNull(props);
-
-            assertEquals(exp.expProps, props);
-
-            props.put(props.size(), mtd);
-
-            assertEquals(exp.expCacheName, ses.cacheName());
-        }
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/25f7984a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/integration/IgniteCacheTxStoreSessionTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/integration/IgniteCacheTxStoreSessionTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/integration/IgniteCacheTxStoreSessionTest.java
new file mode 100644
index 0000000..99900cb
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/integration/IgniteCacheTxStoreSessionTest.java
@@ -0,0 +1,285 @@
+/*
+ * 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.integration;
+
+import org.apache.ignite.*;
+import org.apache.ignite.transactions.*;
+import org.gridgain.grid.cache.*;
+import org.gridgain.grid.util.typedef.*;
+
+import java.util.*;
+
+import static org.apache.ignite.transactions.IgniteTxConcurrency.*;
+import static org.apache.ignite.transactions.IgniteTxIsolation.*;
+import static org.gridgain.grid.cache.GridCacheAtomicityMode.*;
+import static org.gridgain.grid.cache.GridCacheDistributionMode.*;
+import static org.gridgain.grid.cache.GridCacheMode.*;
+
+/**
+ *
+ */
+public class IgniteCacheTxStoreSessionTest extends IgniteCacheStoreSessionAbstractTest {
+    /** {@inheritDoc} */
+    @Override protected int gridCount() {
+        return 3;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected GridCacheMode cacheMode() {
+        return PARTITIONED;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected GridCacheAtomicityMode atomicityMode() {
+        return TRANSACTIONAL;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected GridCacheDistributionMode distributionMode() {
+        return PARTITIONED_ONLY;
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testStoreSessionTx() throws Exception {
+        testTxPut(jcache(0), null, null);
+
+        testTxPut(ignite(0).jcache(CACHE_NAME1), null, null);
+
+        testTxRemove(null, null);
+
+        testTxPutRemove(null, null);
+
+        for (IgniteTxConcurrency concurrency : F.asList(PESSIMISTIC)) {
+            for (IgniteTxIsolation isolation : F.asList(REPEATABLE_READ)) {
+                testTxPut(jcache(0), concurrency, isolation);
+
+                testTxRemove(concurrency, isolation);
+
+                testTxPutRemove(concurrency, isolation);
+            }
+        }
+    }
+
+    /**
+     * @param concurrency Concurrency mode.
+     * @param isolation Isolation mode.
+     * @throws Exception If failed.
+     */
+    private void testTxPutRemove(IgniteTxConcurrency concurrency, IgniteTxIsolation isolation) throws Exception {
+        log.info("Test tx put/remove [concurrency=" + concurrency + ", isolation=" + isolation + ']');
+
+        IgniteCache<Integer, Integer> cache = jcache(0);
+
+        List<Integer> keys = testKeys(cache, 3);
+
+        Integer key1 = keys.get(0);
+        Integer key2 = keys.get(1);
+        Integer key3 = keys.get(2);
+
+        try (IgniteTx tx = startTx(concurrency, isolation)) {
+            log.info("Do tx put1.");
+
+            cache.put(key1, key1);
+
+            log.info("Do tx put2.");
+
+            cache.put(key2, key2);
+
+            log.info("Do tx remove.");
+
+            cache.remove(key3);
+
+            expData.add(new ExpectedData(true, "writeAll", new HashMap<>(), null));
+            expData.add(new ExpectedData(true, "delete", F.<Object, Object>asMap(0, "writeAll"), null));
+            expData.add(new ExpectedData(true, "txEnd", F.<Object, Object>asMap(0, "writeAll", 1, "delete"), null));
+
+            log.info("Do tx commit.");
+
+            tx.commit();
+        }
+
+        assertEquals(0, expData.size());
+    }
+
+    /**
+     * @param cache Cache.
+     * @param concurrency Concurrency mode.
+     * @param isolation Isolation mode.
+     * @throws Exception If failed.
+     */
+    private void testTxPut(IgniteCache<Object, Object> cache,
+        IgniteTxConcurrency concurrency,
+        IgniteTxIsolation isolation) throws Exception {
+        log.info("Test tx put [concurrency=" + concurrency + ", isolation=" + isolation + ']');
+
+        List<Integer> keys = testKeys(cache, 3);
+
+        Integer key1 = keys.get(0);
+
+        try (IgniteTx tx = startTx(concurrency, isolation)) {
+            log.info("Do tx get.");
+
+            cache.get(key1);
+
+            log.info("Do tx put.");
+
+            cache.put(key1, key1);
+
+            expData.add(new ExpectedData(true, "write", new HashMap<>(), cache.getName()));
+            expData.add(new ExpectedData(true, "txEnd", F.<Object, Object>asMap(0, "write"), cache.getName()));
+
+            log.info("Do tx commit.");
+
+            tx.commit();
+        }
+
+        assertEquals(0, expData.size());
+
+        Integer key2 = keys.get(1);
+        Integer key3 = keys.get(2);
+
+        try (IgniteTx tx = startTx(concurrency, isolation)) {
+            log.info("Do tx put1.");
+
+            cache.put(key2, key2);
+
+            log.info("Do tx put2.");
+
+            cache.put(key3, key3);
+
+            expData.add(new ExpectedData(true, "writeAll", new HashMap<>(), cache.getName()));
+            expData.add(new ExpectedData(true, "txEnd", F.<Object, Object>asMap(0, "writeAll"), cache.getName()));
+
+            log.info("Do tx commit.");
+
+            tx.commit();
+        }
+
+        assertEquals(0, expData.size());
+    }
+
+    /**
+     * @param concurrency Concurrency mode.
+     * @param isolation Isolation mode.
+     * @throws Exception If failed.
+     */
+    private void testTxRemove(IgniteTxConcurrency concurrency, IgniteTxIsolation isolation) throws Exception {
+        log.info("Test tx remove [concurrency=" + concurrency + ", isolation=" + isolation + ']');
+
+        IgniteCache<Integer, Integer> cache = jcache(0);
+
+        List<Integer> keys = testKeys(cache, 3);
+
+        Integer key1 = keys.get(0);
+
+        try (IgniteTx tx = startTx(concurrency, isolation)) {
+            log.info("Do tx get.");
+
+            cache.get(key1);
+
+            log.info("Do tx remove.");
+
+            cache.remove(key1, key1);
+
+            expData.add(new ExpectedData(true, "delete", new HashMap<>(), null));
+            expData.add(new ExpectedData(true, "txEnd", F.<Object, Object>asMap(0, "delete"), null));
+
+            log.info("Do tx commit.");
+
+            tx.commit();
+        }
+
+        assertEquals(0, expData.size());
+
+        Integer key2 = keys.get(1);
+        Integer key3 = keys.get(2);
+
+        try (IgniteTx tx = startTx(concurrency, isolation)) {
+            log.info("Do tx remove1.");
+
+            cache.remove(key2, key2);
+
+            log.info("Do tx remove2.");
+
+            cache.remove(key3, key3);
+
+            expData.add(new ExpectedData(true, "deleteAll", new HashMap<>(), null));
+            expData.add(new ExpectedData(true, "txEnd", F.<Object, Object>asMap(0, "deleteAll"), null));
+
+            log.info("Do tx commit.");
+
+            tx.commit();
+        }
+
+        assertEquals(0, expData.size());
+    }
+
+    /**
+     * @param concurrency Concurrency mode.
+     * @param isolation Isolation mode.
+     * @return Transaction.
+     */
+    private IgniteTx startTx(IgniteTxConcurrency concurrency, IgniteTxIsolation isolation) {
+        IgniteTransactions txs = ignite(0).transactions();
+
+        if (concurrency == null)
+            return txs.txStart();
+
+        return txs.txStart(concurrency, isolation);
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testSessionCrossCacheTx() throws Exception {
+        IgniteCache<Object, Object> cache0 = ignite(0).jcache(null);
+
+        IgniteCache<Object, Object> cache1 = ignite(0).jcache(CACHE_NAME1);
+
+        Integer key1 = primaryKey(cache0);
+        Integer key2 = primaryKeys(cache1, 1, key1 + 1).get(0);
+
+        try (IgniteTx tx = startTx(null, null)) {
+            cache0.put(key1, 1);
+
+            cache1.put(key2, 0);
+
+            expData.add(new ExpectedData(true, "writeAll", new HashMap<>(), null));
+            expData.add(new ExpectedData(true, "txEnd", F.<Object, Object>asMap(0, "writeAll"), null));
+
+            tx.commit();
+        }
+
+        assertEquals(0, expData.size());
+
+        try (IgniteTx tx = startTx(null, null)) {
+            cache1.put(key1, 1);
+
+            cache0.put(key2, 0);
+
+            expData.add(new ExpectedData(true, "writeAll", new HashMap<>(), null));
+            expData.add(new ExpectedData(true, "txEnd", F.<Object, Object>asMap(0, "writeAll"), null));
+
+            tx.commit();
+        }
+
+        assertEquals(0, expData.size());
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/25f7984a/modules/core/src/test/java/org/gridgain/grid/cache/store/GridCacheLoadOnlyStoreAdapterSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/gridgain/grid/cache/store/GridCacheLoadOnlyStoreAdapterSelfTest.java b/modules/core/src/test/java/org/gridgain/grid/cache/store/GridCacheLoadOnlyStoreAdapterSelfTest.java
index 3a761f8..931f428 100644
--- a/modules/core/src/test/java/org/gridgain/grid/cache/store/GridCacheLoadOnlyStoreAdapterSelfTest.java
+++ b/modules/core/src/test/java/org/gridgain/grid/cache/store/GridCacheLoadOnlyStoreAdapterSelfTest.java
@@ -50,6 +50,7 @@ public class GridCacheLoadOnlyStoreAdapterSelfTest extends GridCacheAbstractSelf
         cfg.setCacheStoreFactory(new FactoryBuilder.SingletonFactory(new TestStore()));
         cfg.setReadThrough(true);
         cfg.setWriteThrough(true);
+        cfg.setLoadPreviousValue(true);
 
         return cfg;
     }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/25f7984a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheAbstractSelfTest.java b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheAbstractSelfTest.java
index 4860f4d..e70ca7a 100644
--- a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheAbstractSelfTest.java
+++ b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheAbstractSelfTest.java
@@ -239,6 +239,7 @@ public abstract class GridCacheAbstractSelfTest extends GridCommonAbstractTest {
             cfg.setCacheStoreFactory(new FactoryBuilder.SingletonFactory(store));
             cfg.setReadThrough(true);
             cfg.setWriteThrough(true);
+            cfg.setLoadPreviousValue(true);
         }
 
         cfg.setSwapEnabled(swapEnabled());

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/25f7984a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheBasicStoreAbstractTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheBasicStoreAbstractTest.java b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheBasicStoreAbstractTest.java
index 0bb03b8..3c9c1b2 100644
--- a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheBasicStoreAbstractTest.java
+++ b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheBasicStoreAbstractTest.java
@@ -95,6 +95,7 @@ public abstract class GridCacheBasicStoreAbstractTest extends GridCommonAbstract
         cc.setCacheStoreFactory(new FactoryBuilder.SingletonFactory(store));
         cc.setReadThrough(true);
         cc.setWriteThrough(true);
+        cc.setLoadPreviousValue(true);
 
         c.setCacheConfiguration(cc);
 

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/25f7984a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheBasicStoreMultithreadedAbstractTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheBasicStoreMultithreadedAbstractTest.java b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheBasicStoreMultithreadedAbstractTest.java
index bd8e8ed..88d8638 100644
--- a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheBasicStoreMultithreadedAbstractTest.java
+++ b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheBasicStoreMultithreadedAbstractTest.java
@@ -81,6 +81,7 @@ public abstract class GridCacheBasicStoreMultithreadedAbstractTest extends GridC
         cc.setCacheStoreFactory(new FactoryBuilder.SingletonFactory(store));
         cc.setReadThrough(true);
         cc.setWriteThrough(true);
+        cc.setLoadPreviousValue(true);
 
         c.setCacheConfiguration(cc);
 

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/25f7984a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheConfigurationConsistencySelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheConfigurationConsistencySelfTest.java b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheConfigurationConsistencySelfTest.java
index d54cd41..db6f67c 100644
--- a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheConfigurationConsistencySelfTest.java
+++ b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheConfigurationConsistencySelfTest.java
@@ -746,6 +746,7 @@ public class GridCacheConfigurationConsistencySelfTest extends GridCommonAbstrac
                 cfg.setCacheStoreFactory(new FactoryBuilder.SingletonFactory(new TestStore()));
                 cfg.setReadThrough(true);
                 cfg.setWriteThrough(true);
+                cfg.setLoadPreviousValue(true);
 
                 return null;
             }
@@ -787,6 +788,7 @@ public class GridCacheConfigurationConsistencySelfTest extends GridCommonAbstrac
                 cc.setCacheStoreFactory(new FactoryBuilder.SingletonFactory(new TestStore()));
                 cc.setReadThrough(true);
                 cc.setWriteThrough(true);
+                cc.setLoadPreviousValue(true);
 
                 return null;
             }
@@ -825,6 +827,7 @@ public class GridCacheConfigurationConsistencySelfTest extends GridCommonAbstrac
                 cc.setCacheStoreFactory(new FactoryBuilder.SingletonFactory(new TestStore()));
                 cc.setReadThrough(true);
                 cc.setWriteThrough(true);
+                cc.setLoadPreviousValue(true);
 
                 return null;
             }
@@ -869,6 +872,7 @@ public class GridCacheConfigurationConsistencySelfTest extends GridCommonAbstrac
                 cc.setCacheStoreFactory(new FactoryBuilder.SingletonFactory(new TestStore()));
                 cc.setReadThrough(true);
                 cc.setWriteThrough(true);
+                cc.setLoadPreviousValue(true);
 
                 return null;
             }
@@ -915,6 +919,7 @@ public class GridCacheConfigurationConsistencySelfTest extends GridCommonAbstrac
                 cc.setCacheStoreFactory(new FactoryBuilder.SingletonFactory(new TestStore()));
                 cc.setReadThrough(true);
                 cc.setWriteThrough(true);
+                cc.setLoadPreviousValue(true);
 
                 return null;
             }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/25f7984a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheGetAndTransformStoreAbstractTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheGetAndTransformStoreAbstractTest.java b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheGetAndTransformStoreAbstractTest.java
index 0f2de67..2ee15b4 100644
--- a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheGetAndTransformStoreAbstractTest.java
+++ b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheGetAndTransformStoreAbstractTest.java
@@ -94,6 +94,7 @@ public abstract class GridCacheGetAndTransformStoreAbstractTest extends GridComm
         cc.setCacheStoreFactory(new FactoryBuilder.SingletonFactory(store));
         cc.setReadThrough(true);
         cc.setWriteThrough(true);
+        cc.setLoadPreviousValue(true);
 
         c.setCacheConfiguration(cc);
 


Mime
View raw message