ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From vkuliche...@apache.org
Subject [25/34] incubator-ignite git commit: IGNITE-891 - Cache store improvements
Date Tue, 02 Jun 2015 03:51:19 GMT
IGNITE-891 - Cache store improvements


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

Branch: refs/heads/ignite-sprint-5
Commit: ada1b2a7c4d82722cc5721bad50042af7216bfdc
Parents: 990bf9e
Author: Valentin Kulichenko <vkulichenko@gridgain.com>
Authored: Sun May 24 20:42:53 2015 -0700
Committer: Valentin Kulichenko <vkulichenko@gridgain.com>
Committed: Sun May 24 20:42:53 2015 -0700

----------------------------------------------------------------------
 .../hibernate/CacheHibernatePersonStore.java    |  27 +-
 .../store/jdbc/CacheJdbcPersonStore.java        |  69 ++--
 .../store/jdbc/CacheJdbcStoreExample.java       |   3 +-
 .../store/spring/CacheSpringPersonStore.java    | 128 ++++++
 .../store/spring/CacheSpringStoreExample.java   | 143 +++++++
 .../datagrid/store/spring/package-info.java     |  22 ++
 .../processors/cache/GridCacheProcessor.java    |   4 +-
 .../processors/cache/GridCacheUtils.java        |  25 +-
 .../store/GridCacheStoreManagerAdapter.java     |  61 ++-
 ...heStoreSessionListenerLifeCycleSelfTest.java | 395 +++++++++++++++++++
 .../IgniteCrossCacheTxStoreSelfTest.java        |  24 --
 .../junits/common/GridCommonAbstractTest.java   |  24 ++
 .../spring/CacheSpringStoreSessionListener.java |   2 +-
 13 files changed, 810 insertions(+), 117 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ada1b2a7/examples/src/main/java/org/apache/ignite/examples/datagrid/store/hibernate/CacheHibernatePersonStore.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/datagrid/store/hibernate/CacheHibernatePersonStore.java b/examples/src/main/java/org/apache/ignite/examples/datagrid/store/hibernate/CacheHibernatePersonStore.java
index 557ec6f..80a9f22 100644
--- a/examples/src/main/java/org/apache/ignite/examples/datagrid/store/hibernate/CacheHibernatePersonStore.java
+++ b/examples/src/main/java/org/apache/ignite/examples/datagrid/store/hibernate/CacheHibernatePersonStore.java
@@ -54,13 +54,7 @@ public class CacheHibernatePersonStore extends CacheStoreAdapter<Long, Person> {
         Long key = entry.getKey();
         Person val = entry.getValue();
 
-        System.out.println(">>> Store put [key=" + key + ", val=" + val + ']');
-
-        if (val == null) {
-            delete(key);
-
-            return;
-        }
+        System.out.println(">>> Store write [key=" + key + ", val=" + val + ']');
 
         Session hibSes = ses.attachment();
 
@@ -75,13 +69,14 @@ public class CacheHibernatePersonStore extends CacheStoreAdapter<Long, Person> {
     /** {@inheritDoc} */
     @SuppressWarnings({"JpaQueryApiInspection"})
     @Override public void delete(Object key) {
-        System.out.println(">>> Store remove [key=" + key + ']');
+        System.out.println(">>> Store delete [key=" + key + ']');
 
         Session hibSes = ses.attachment();
 
         try {
-            hibSes.createQuery("delete " + Person.class.getSimpleName() + " where key = :key")
-                .setParameter("key", key).setFlushMode(FlushMode.ALWAYS).executeUpdate();
+            hibSes.createQuery("delete " + Person.class.getSimpleName() + " where key = :key").
+                setParameter("key", key).
+                executeUpdate();
         }
         catch (HibernateException e) {
             throw new CacheWriterException("Failed to remove value from cache store [key=" + key + ']', e);
@@ -100,13 +95,13 @@ public class CacheHibernatePersonStore extends CacheStoreAdapter<Long, Person> {
         try {
             int cnt = 0;
 
-            List res = hibSes.createCriteria(Person.class).list();
-
-            if (res != null) {
-                Iterator iter = res.iterator();
+            List list = hibSes.createCriteria(Person.class).
+                setMaxResults(entryCnt).
+                list();
 
-                while (cnt < entryCnt && iter.hasNext()) {
-                    Person person = (Person)iter.next();
+            if (list != null) {
+                for (Object obj : list) {
+                    Person person = (Person)obj;
 
                     clo.apply(person.getId(), person);
 

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ada1b2a7/examples/src/main/java/org/apache/ignite/examples/datagrid/store/jdbc/CacheJdbcPersonStore.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/datagrid/store/jdbc/CacheJdbcPersonStore.java b/examples/src/main/java/org/apache/ignite/examples/datagrid/store/jdbc/CacheJdbcPersonStore.java
index 6eb0386..ed14a99 100644
--- a/examples/src/main/java/org/apache/ignite/examples/datagrid/store/jdbc/CacheJdbcPersonStore.java
+++ b/examples/src/main/java/org/apache/ignite/examples/datagrid/store/jdbc/CacheJdbcPersonStore.java
@@ -22,9 +22,11 @@ import org.apache.ignite.cache.store.*;
 import org.apache.ignite.examples.datagrid.store.*;
 import org.apache.ignite.lang.*;
 import org.apache.ignite.resources.*;
+import org.h2.jdbcx.*;
 
 import javax.cache.*;
 import javax.cache.integration.*;
+import javax.sql.*;
 import java.sql.*;
 
 /**
@@ -32,6 +34,10 @@ import java.sql.*;
  * transaction with cache transactions and maps {@link Long} to {@link Person}.
  */
 public class CacheJdbcPersonStore extends CacheStoreAdapter<Long, Person> {
+    /** Data source. */
+    public static final DataSource DATA_SRC =
+        JdbcConnectionPool.create("jdbc:h2:mem:example;DB_CLOSE_DELAY=-1", "", "");
+
     /** Store session. */
     @CacheStoreSessionResource
     private CacheStoreSession ses;
@@ -52,12 +58,10 @@ public class CacheJdbcPersonStore extends CacheStoreAdapter<Long, Person> {
      * @throws IgniteException If failed.
      */
     private void prepareDb() throws IgniteException {
-        try (
-            Connection conn = DriverManager.getConnection("jdbc:h2:mem:example;DB_CLOSE_DELAY=-1");
-            Statement st = conn.createStatement()
-        ) {
-            st.execute("create table if not exists PERSONS (id number unique, firstName varchar(255), " +
-                "lastName varchar(255))");
+        try (Connection conn = DATA_SRC.getConnection()) {
+            conn.createStatement().execute(
+                "create table if not exists PERSONS (" +
+                "id number unique, firstName varchar(255), lastName varchar(255))");
         }
         catch (SQLException e) {
             throw new IgniteException("Failed to create database table.", e);
@@ -66,34 +70,28 @@ public class CacheJdbcPersonStore extends CacheStoreAdapter<Long, Person> {
 
     /** {@inheritDoc} */
     @Override public Person load(Long key) {
-        System.out.println(">>> Loading key: " + key);
+        System.out.println(">>> Store load [key=" + key + ']');
 
-        try {
-            Connection conn = ses.attachment();
+        Connection conn = ses.attachment();
 
-            try (PreparedStatement st = conn.prepareStatement("select * from PERSONS where id=?")) {
-                st.setString(1, key.toString());
+        try (PreparedStatement st = conn.prepareStatement("select * from PERSONS where id = ?")) {
+            st.setString(1, key.toString());
 
-                ResultSet rs = st.executeQuery();
+            ResultSet rs = st.executeQuery();
 
-                if (rs.next())
-                    return new Person(rs.getLong(1), rs.getString(2), rs.getString(3));
-            }
+            return rs.next() ? new Person(rs.getLong(1), rs.getString(2), rs.getString(3)) : null;
         }
         catch (SQLException e) {
-            throw new CacheLoaderException("Failed to load object: " + key, e);
+            throw new CacheLoaderException("Failed to load object [key=" + key + ']', e);
         }
-
-        return null;
     }
 
     /** {@inheritDoc} */
     @Override public void write(Cache.Entry<? extends Long, ? extends Person> entry) {
         Long key = entry.getKey();
-
         Person val = entry.getValue();
 
-        System.out.println(">>> Putting [key=" + key + ", val=" + val +  ']');
+        System.out.println(">>> Store write [key=" + key + ", val=" + val + ']');
 
         try {
             Connection conn = ses.attachment();
@@ -103,7 +101,7 @@ public class CacheJdbcPersonStore extends CacheStoreAdapter<Long, Person> {
             // Try update first. If it does not work, then try insert.
             // Some databases would allow these to be done in one 'upsert' operation.
             try (PreparedStatement st = conn.prepareStatement(
-                "update PERSONS set firstName=?, lastName=? where id=?")) {
+                "update PERSONS set firstName = ?, lastName = ? where id = ?")) {
                 st.setString(1, val.getFirstName());
                 st.setString(2, val.getLastName());
                 st.setLong(3, val.getId());
@@ -114,7 +112,7 @@ public class CacheJdbcPersonStore extends CacheStoreAdapter<Long, Person> {
             // If update failed, try to insert.
             if (updated == 0) {
                 try (PreparedStatement st = conn.prepareStatement(
-                    "insert into PERSONS (id, firstName, lastName) values(?, ?, ?)")) {
+                    "insert into PERSONS (id, firstName, lastName) values (?, ?, ?)")) {
                     st.setLong(1, val.getId());
                     st.setString(2, val.getFirstName());
                     st.setString(3, val.getLastName());
@@ -124,25 +122,23 @@ public class CacheJdbcPersonStore extends CacheStoreAdapter<Long, Person> {
             }
         }
         catch (SQLException e) {
-            throw new CacheLoaderException("Failed to put object [key=" + key + ", val=" + val + ']', e);
+            throw new CacheWriterException("Failed to write object [key=" + key + ", val=" + val + ']', e);
         }
     }
 
     /** {@inheritDoc} */
     @Override public void delete(Object key) {
-        System.out.println(">>> Removing key: " + key);
+        System.out.println(">>> Store delete [key=" + key + ']');
 
-        try {
-            Connection conn = ses.attachment();
+        Connection conn = ses.attachment();
 
-            try (PreparedStatement st = conn.prepareStatement("delete from PERSONS where id=?")) {
-                st.setLong(1, (Long)key);
+        try (PreparedStatement st = conn.prepareStatement("delete from PERSONS where id=?")) {
+            st.setLong(1, (Long)key);
 
-                st.executeUpdate();
-            }
+            st.executeUpdate();
         }
         catch (SQLException e) {
-            throw new CacheWriterException("Failed to remove object: " + key, e);
+            throw new CacheWriterException("Failed to delete object [key=" + key + ']', e);
         }
     }
 
@@ -155,13 +151,14 @@ public class CacheJdbcPersonStore extends CacheStoreAdapter<Long, Person> {
 
         Connection conn = ses.attachment();
 
-        try (
-            PreparedStatement st = conn.prepareStatement("select * from PERSONS");
-            ResultSet rs = st.executeQuery()
-        ) {
+        try (PreparedStatement stmt = conn.prepareStatement("select * from PERSONS limit ?")) {
+            stmt.setInt(1, entryCnt);
+
+            ResultSet rs = stmt.executeQuery();
+
             int cnt = 0;
 
-            while (cnt < entryCnt && rs.next()) {
+            while (rs.next()) {
                 Person person = new Person(rs.getLong(1), rs.getString(2), rs.getString(3));
 
                 clo.apply(person.getId(), person);

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ada1b2a7/examples/src/main/java/org/apache/ignite/examples/datagrid/store/jdbc/CacheJdbcStoreExample.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/datagrid/store/jdbc/CacheJdbcStoreExample.java b/examples/src/main/java/org/apache/ignite/examples/datagrid/store/jdbc/CacheJdbcStoreExample.java
index 74e262c..637d6dc 100644
--- a/examples/src/main/java/org/apache/ignite/examples/datagrid/store/jdbc/CacheJdbcStoreExample.java
+++ b/examples/src/main/java/org/apache/ignite/examples/datagrid/store/jdbc/CacheJdbcStoreExample.java
@@ -24,7 +24,6 @@ import org.apache.ignite.configuration.*;
 import org.apache.ignite.examples.*;
 import org.apache.ignite.examples.datagrid.store.*;
 import org.apache.ignite.transactions.*;
-import org.h2.jdbcx.*;
 
 import javax.cache.configuration.*;
 import java.util.*;
@@ -79,7 +78,7 @@ public class CacheJdbcStoreExample {
                 @Override public CacheStoreSessionListener create() {
                     CacheJdbcStoreSessionListener lsnr = new CacheJdbcStoreSessionListener();
 
-                    lsnr.setDataSource(JdbcConnectionPool.create("jdbc:h2:mem:example;DB_CLOSE_DELAY=-1", "", ""));
+                    lsnr.setDataSource(CacheJdbcPersonStore.DATA_SRC);
 
                     return lsnr;
                 }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ada1b2a7/examples/src/main/java/org/apache/ignite/examples/datagrid/store/spring/CacheSpringPersonStore.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/datagrid/store/spring/CacheSpringPersonStore.java b/examples/src/main/java/org/apache/ignite/examples/datagrid/store/spring/CacheSpringPersonStore.java
new file mode 100644
index 0000000..50149ba
--- /dev/null
+++ b/examples/src/main/java/org/apache/ignite/examples/datagrid/store/spring/CacheSpringPersonStore.java
@@ -0,0 +1,128 @@
+/*
+ * 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.examples.datagrid.store.spring;
+
+import org.apache.ignite.*;
+import org.apache.ignite.cache.store.*;
+import org.apache.ignite.examples.datagrid.store.*;
+import org.apache.ignite.lang.*;
+import org.springframework.dao.*;
+import org.springframework.jdbc.core.*;
+import org.springframework.jdbc.datasource.*;
+
+import javax.cache.*;
+import javax.cache.integration.*;
+import javax.sql.*;
+import java.sql.*;
+import java.util.concurrent.atomic.*;
+
+/**
+ * Example of {@link CacheStore} implementation that uses JDBC
+ * transaction with cache transactions and maps {@link Long} to {@link Person}.
+ */
+public class CacheSpringPersonStore extends CacheStoreAdapter<Long, Person> {
+    /** Data source. */
+    public static final DataSource DATA_SRC = new DriverManagerDataSource("jdbc:h2:mem:example;DB_CLOSE_DELAY=-1");
+
+    /** Spring JDBC template. */
+    private JdbcTemplate jdbcTemplate;
+
+    /**
+     * Constructor.
+     *
+     * @throws IgniteException If failed.
+     */
+    public CacheSpringPersonStore() throws IgniteException {
+        jdbcTemplate = new JdbcTemplate(DATA_SRC);
+
+        prepareDb();
+    }
+
+    /**
+     * Prepares database for example execution. This method will create a
+     * table called "PERSONS" so it can be used by store implementation.
+     *
+     * @throws IgniteException If failed.
+     */
+    private void prepareDb() throws IgniteException {
+        jdbcTemplate.update(
+            "create table if not exists PERSONS (" +
+            "id number unique, firstName varchar(255), lastName varchar(255))");
+    }
+
+    /** {@inheritDoc} */
+    @Override public Person load(Long key) {
+        System.out.println(">>> Store load [key=" + key + ']');
+
+        try {
+            return jdbcTemplate.queryForObject("select * from PERSONS where id = ?", new RowMapper<Person>() {
+                @Override public Person mapRow(ResultSet rs, int rowNum) throws SQLException {
+                    return new Person(rs.getLong(1), rs.getString(2), rs.getString(3));
+                }
+            }, key);
+        }
+        catch (EmptyResultDataAccessException ignored) {
+            return null;
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public void write(Cache.Entry<? extends Long, ? extends Person> entry) {
+        Long key = entry.getKey();
+        Person val = entry.getValue();
+
+        System.out.println(">>> Store write [key=" + key + ", val=" + val + ']');
+
+        int updated = jdbcTemplate.update("update PERSONS set firstName = ?, lastName = ? where id = ?",
+            val.getFirstName(), val.getLastName(), val.getId());
+
+        if (updated == 0) {
+            jdbcTemplate.update("insert into PERSONS (id, firstName, lastName) values (?, ?, ?)",
+                val.getId(), val.getFirstName(), val.getLastName());
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public void delete(Object key) {
+        System.out.println(">>> Store delete [key=" + key + ']');
+
+        jdbcTemplate.update("delete from PERSONS where id = ?", key);
+    }
+
+    /** {@inheritDoc} */
+    @Override public void loadCache(final IgniteBiInClosure<Long, Person> clo, Object... args) {
+        if (args == null || args.length == 0 || args[0] == null)
+            throw new CacheLoaderException("Expected entry count parameter is not provided.");
+
+        int entryCnt = (Integer)args[0];
+
+        final AtomicInteger cnt = new AtomicInteger();
+
+        jdbcTemplate.query("select * from PERSONS limit ?", new RowCallbackHandler() {
+            @Override public void processRow(ResultSet rs) throws SQLException {
+                Person person = new Person(rs.getLong(1), rs.getString(2), rs.getString(3));
+
+                clo.apply(person.getId(), person);
+
+                cnt.incrementAndGet();
+            }
+        }, entryCnt);
+
+        System.out.println(">>> Loaded " + cnt + " values into cache.");
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ada1b2a7/examples/src/main/java/org/apache/ignite/examples/datagrid/store/spring/CacheSpringStoreExample.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/datagrid/store/spring/CacheSpringStoreExample.java b/examples/src/main/java/org/apache/ignite/examples/datagrid/store/spring/CacheSpringStoreExample.java
new file mode 100644
index 0000000..9be6672
--- /dev/null
+++ b/examples/src/main/java/org/apache/ignite/examples/datagrid/store/spring/CacheSpringStoreExample.java
@@ -0,0 +1,143 @@
+/*
+ * 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.examples.datagrid.store.spring;
+
+import org.apache.ignite.*;
+import org.apache.ignite.cache.store.*;
+import org.apache.ignite.cache.store.jdbc.*;
+import org.apache.ignite.configuration.*;
+import org.apache.ignite.examples.*;
+import org.apache.ignite.examples.datagrid.store.*;
+import org.apache.ignite.transactions.*;
+
+import javax.cache.configuration.*;
+import java.util.*;
+
+import static org.apache.ignite.cache.CacheAtomicityMode.*;
+
+/**
+ * Demonstrates usage of cache with underlying persistent store configured.
+ * <p>
+ * This example uses {@link CacheSpringPersonStore} as a persistent store.
+ * <p>
+ * Remote nodes can be started with {@link ExampleNodeStartup} in another JVM which will
+ * start node with {@code examples/config/example-ignite.xml} configuration.
+ */
+public class CacheSpringStoreExample {
+    /** Cache name. */
+    private static final String CACHE_NAME = CacheSpringStoreExample.class.getSimpleName();
+
+    /** Heap size required to run this example. */
+    public static final int MIN_MEMORY = 1024 * 1024 * 1024;
+
+    /** Number of entries to load. */
+    private static final int ENTRY_COUNT = 100_000;
+
+    /** Global person ID to use across entire example. */
+    private static final Long id = Math.abs(UUID.randomUUID().getLeastSignificantBits());
+
+    /**
+     * Executes example.
+     *
+     * @param args Command line arguments, none required.
+     * @throws IgniteException If example execution failed.
+     */
+    public static void main(String[] args) throws IgniteException {
+        ExamplesUtils.checkMinMemory(MIN_MEMORY);
+
+        // To start ignite with desired configuration uncomment the appropriate line.
+        try (Ignite ignite = Ignition.start("examples/config/example-ignite.xml")) {
+            System.out.println();
+            System.out.println(">>> Cache store example started.");
+
+            CacheConfiguration<Long, Person> cacheCfg = new CacheConfiguration<>(CACHE_NAME);
+
+            // Set atomicity as transaction, since we are showing transactions in example.
+            cacheCfg.setAtomicityMode(TRANSACTIONAL);
+
+            // Configure JDBC store.
+            cacheCfg.setCacheStoreFactory(FactoryBuilder.factoryOf(CacheSpringPersonStore.class));
+
+            // Configure JDBC session listener.
+            cacheCfg.setCacheStoreSessionListenerFactories(new Factory<CacheStoreSessionListener>() {
+                @Override public CacheStoreSessionListener create() {
+                    CacheJdbcStoreSessionListener lsnr = new CacheJdbcStoreSessionListener();
+
+                    lsnr.setDataSource(CacheSpringPersonStore.DATA_SRC);
+
+                    return lsnr;
+                }
+            });
+
+            cacheCfg.setReadThrough(true);
+            cacheCfg.setWriteThrough(true);
+
+            try (IgniteCache<Long, Person> cache = ignite.getOrCreateCache(cacheCfg)) {
+                // Make initial cache loading from persistent store. This is a
+                // distributed operation and will call CacheStore.loadCache(...)
+                // method on all nodes in topology.
+                loadCache(cache);
+
+                // Start transaction and execute several cache operations with
+                // read/write-through to persistent store.
+                executeTransaction(cache);
+            }
+        }
+    }
+
+    /**
+     * Makes initial cache loading.
+     *
+     * @param cache Cache to load.
+     */
+    private static void loadCache(IgniteCache<Long, Person> cache) {
+        long start = System.currentTimeMillis();
+
+        // Start loading cache from persistent store on all caching nodes.
+        cache.loadCache(null, ENTRY_COUNT);
+
+        long end = System.currentTimeMillis();
+
+        System.out.println(">>> Loaded " + cache.size() + " keys with backups in " + (end - start) + "ms.");
+    }
+
+    /**
+     * Executes transaction with read/write-through to persistent store.
+     *
+     * @param cache Cache to execute transaction on.
+     */
+    private static void executeTransaction(IgniteCache<Long, Person> cache) {
+        try (Transaction tx = Ignition.ignite().transactions().txStart()) {
+            Person val = cache.get(id);
+
+            System.out.println("Read value: " + val);
+
+            val = cache.getAndPut(id, new Person(id, "Isaac", "Newton"));
+
+            System.out.println("Overwrote old value: " + val);
+
+            val = cache.get(id);
+
+            System.out.println("Read value: " + val);
+
+            tx.commit();
+        }
+
+        System.out.println("Read value after commit: " + cache.get(id));
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ada1b2a7/examples/src/main/java/org/apache/ignite/examples/datagrid/store/spring/package-info.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/datagrid/store/spring/package-info.java b/examples/src/main/java/org/apache/ignite/examples/datagrid/store/spring/package-info.java
new file mode 100644
index 0000000..211239f
--- /dev/null
+++ b/examples/src/main/java/org/apache/ignite/examples/datagrid/store/spring/package-info.java
@@ -0,0 +1,22 @@
+/*
+ * 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 description. -->
+ * Contains Spring-based cache store implementation.
+ */
+package org.apache.ignite.examples.datagrid.store.spring;

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ada1b2a7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
index 5b57817..4457f98 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
@@ -567,7 +567,7 @@ public class GridCacheProcessor extends GridProcessorAdapter {
 
         CacheConfiguration[] cfgs = ctx.config().getCacheConfiguration();
 
-        sharedCtx = createSharedContext(ctx, CU.createStoreSessionListeners(ctx,
+        sharedCtx = createSharedContext(ctx, CU.startStoreSessionListeners(ctx,
             ctx.config().getCacheStoreSessionListenerFactories()));
 
         ctx.performance().add("Disable serializable transactions (set 'txSerializableEnabled' to false)",
@@ -813,6 +813,8 @@ public class GridCacheProcessor extends GridProcessorAdapter {
             mgr.stop(cancel);
         }
 
+        CU.stopStoreSessionListeners(ctx, sharedCtx.storeSessionListeners());
+
         sharedCtx.cleanup();
 
         if (log.isDebugEnabled())

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ada1b2a7/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 6968fcb..7096da5 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
@@ -1795,13 +1795,14 @@ public class GridCacheUtils {
     }
 
     /**
-     * Creates store session listeners.
+     * Creates and starts store session listeners.
      *
      * @param ctx Kernal context.
      * @param factories Factories.
      * @return Listeners.
+     * @throws IgniteCheckedException In case of error.
      */
-    public static Collection<CacheStoreSessionListener> createStoreSessionListeners(GridKernalContext ctx,
+    public static Collection<CacheStoreSessionListener> startStoreSessionListeners(GridKernalContext ctx,
         Factory<CacheStoreSessionListener>[] factories) throws IgniteCheckedException {
         if (factories == null)
             return null;
@@ -1823,4 +1824,24 @@ public class GridCacheUtils {
 
         return lsnrs;
     }
+
+    /**
+     * Stops store session listeners.
+     *
+     * @param ctx Kernal context.
+     * @param sesLsnrs Session listeners.
+     * @throws IgniteCheckedException In case of error.
+     */
+    public static void stopStoreSessionListeners(GridKernalContext ctx, Collection<CacheStoreSessionListener> sesLsnrs)
+        throws IgniteCheckedException {
+        if (sesLsnrs == null)
+            return;
+
+        for (CacheStoreSessionListener lsnr : sesLsnrs) {
+            if (lsnr instanceof LifecycleAware)
+                ((LifecycleAware)lsnr).stop();
+
+            ctx.resource().cleanupGeneric(lsnr);
+        }
+    }
 }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ada1b2a7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/store/GridCacheStoreManagerAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/store/GridCacheStoreManagerAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/store/GridCacheStoreManagerAdapter.java
index 11d711c..bc5a0a2 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/store/GridCacheStoreManagerAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/store/GridCacheStoreManagerAdapter.java
@@ -70,6 +70,9 @@ public abstract class GridCacheStoreManagerAdapter extends GridCacheManagerAdapt
     /** */
     private Collection<CacheStoreSessionListener> sesLsnrs;
 
+    /** */
+    private boolean globalSesLsnrs;
+
     /** {@inheritDoc} */
     @SuppressWarnings("unchecked")
     @Override public void initialize(@Nullable CacheStore cfgStore, Map sesHolders) throws IgniteCheckedException {
@@ -166,10 +169,13 @@ public abstract class GridCacheStoreManagerAdapter extends GridCacheManagerAdapt
                 "Persistence store is configured, but both read-through and write-through are disabled.");
         }
 
-        sesLsnrs = CU.createStoreSessionListeners(cctx.kernalContext(), cfg.getCacheStoreSessionListenerFactories());
+        sesLsnrs = CU.startStoreSessionListeners(cctx.kernalContext(), cfg.getCacheStoreSessionListenerFactories());
 
-        if (sesLsnrs == null)
+        if (sesLsnrs == null) {
             sesLsnrs = cctx.shared().storeSessionListeners();
+
+            globalSesLsnrs = true;
+        }
     }
 
     /** {@inheritDoc} */
@@ -187,18 +193,12 @@ public abstract class GridCacheStoreManagerAdapter extends GridCacheManagerAdapt
             }
         }
 
-        if (sesLsnrs != null) {
-            for (CacheStoreSessionListener lsnr : sesLsnrs) {
-                if (lsnr instanceof LifecycleAware)
-                    ((LifecycleAware)lsnr).stop();
-
-                try {
-                    cctx.kernalContext().resource().cleanupGeneric(lsnr);
-                }
-                catch (IgniteCheckedException e) {
-                    U.error(log, "Failed to remove injected resources from store session listener (ignoring): " +
-                        lsnr, e);
-                }
+        if (!globalSesLsnrs) {
+            try {
+                CU.stopStoreSessionListeners(cctx.kernalContext(), sesLsnrs);
+            }
+            catch (IgniteCheckedException e) {
+                U.error(log, "Failed to stop store session listeners for cache: " + cctx.name(), e);
             }
         }
     }
@@ -721,7 +721,7 @@ public abstract class GridCacheStoreManagerAdapter extends GridCacheManagerAdapt
                     lsnr.onSessionEnd(locSes, commit);
             }
 
-            if (!sesHolder.get().storeEnded(store))
+            if (!sesHolder.get().ended(store))
                 store.sessionEnd(commit);
         }
         catch (Throwable e) {
@@ -788,13 +788,9 @@ public abstract class GridCacheStoreManagerAdapter extends GridCacheManagerAdapt
 
         sesHolder.set(ses);
 
-        if (!ses.started()) {
-            if (sesLsnrs != null) {
-                for (CacheStoreSessionListener lsnr : sesLsnrs)
-                    lsnr.onSessionStart(locSes);
-            }
-
-            ses.onStarted();
+        if (sesLsnrs != null && !ses.started(this)) {
+            for (CacheStoreSessionListener lsnr : sesLsnrs)
+                lsnr.onSessionStart(locSes);
         }
     }
 
@@ -809,7 +805,7 @@ public abstract class GridCacheStoreManagerAdapter extends GridCacheManagerAdapt
                         lsnr.onSessionEnd(locSes, !threwEx);
                 }
 
-                assert !sesHolder.get().storeEnded(store);
+                assert !sesHolder.get().ended(store);
 
                 store.sessionEnd(!threwEx);
             }
@@ -858,10 +854,11 @@ public abstract class GridCacheStoreManagerAdapter extends GridCacheManagerAdapt
         private Object attachment;
 
         /** */
-        private boolean started;
+        private final Set<CacheStoreManager> started =
+            new GridSetWrapper<>(new IdentityHashMap<CacheStoreManager, Object>());
 
         /** */
-        private final Set<CacheStore> endedStores = new GridSetWrapper<>(new IdentityHashMap<CacheStore, Object>());
+        private final Set<CacheStore> ended = new GridSetWrapper<>(new IdentityHashMap<CacheStore, Object>());
 
         /**
          * @param tx Current transaction.
@@ -918,24 +915,18 @@ public abstract class GridCacheStoreManagerAdapter extends GridCacheManagerAdapt
         }
 
         /**
-         */
-        private void onStarted() {
-            started = true;
-        }
-
-        /**
          * @return If session is started.
          */
-        private boolean started() {
-            return started;
+        private boolean started(CacheStoreManager mgr) {
+            return !started.add(mgr);
         }
 
         /**
          * @param store Cache store.
          * @return Whether session already ended on this store instance.
          */
-        private boolean storeEnded(CacheStore store) {
-            return !endedStores.add(store);
+        private boolean ended(CacheStore store) {
+            return !ended.add(store);
         }
 
         /** {@inheritDoc} */

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ada1b2a7/modules/core/src/test/java/org/apache/ignite/cache/store/CacheStoreSessionListenerLifeCycleSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/cache/store/CacheStoreSessionListenerLifeCycleSelfTest.java b/modules/core/src/test/java/org/apache/ignite/cache/store/CacheStoreSessionListenerLifeCycleSelfTest.java
new file mode 100644
index 0000000..814c8a5
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/cache/store/CacheStoreSessionListenerLifeCycleSelfTest.java
@@ -0,0 +1,395 @@
+/*
+ * 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.*;
+import org.apache.ignite.configuration.*;
+import org.apache.ignite.lifecycle.*;
+import org.apache.ignite.resources.*;
+import org.apache.ignite.spi.discovery.tcp.*;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.*;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.*;
+import org.apache.ignite.testframework.junits.common.*;
+import org.apache.ignite.transactions.*;
+
+import javax.cache.*;
+import javax.cache.configuration.*;
+import javax.cache.integration.*;
+import java.util.*;
+import java.util.concurrent.*;
+
+import static org.apache.ignite.cache.CacheAtomicityMode.*;
+
+/**
+ * Store session listeners test.
+ */
+public class CacheStoreSessionListenerLifecycleSelfTest extends GridCommonAbstractTest {
+    /** */
+    private static final TcpDiscoveryIpFinder IP_FINDER = new TcpDiscoveryVmIpFinder(true);
+
+    /** */
+    private static final Queue<String> evts = new ConcurrentLinkedDeque<>();
+
+    /** {@inheritDoc} */
+    @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+        IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+        cfg.setCacheStoreSessionListenerFactories(
+            new SessionListenerFactory("Shared 1"),
+            new SessionListenerFactory("Shared 2")
+        );
+
+        TcpDiscoverySpi disco = new TcpDiscoverySpi();
+
+        disco.setIpFinder(IP_FINDER);
+
+        cfg.setDiscoverySpi(disco);
+
+        return cfg;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void beforeTest() throws Exception {
+        evts.clear();
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testNoCaches() throws Exception {
+        try {
+            startGrid();
+        }
+        finally {
+            stopGrid();
+        }
+
+        assertEqualsCollections(Arrays.asList("Shared 1 START", "Shared 2 START", "Shared 1 STOP", "Shared 2 STOP"),
+            evts);
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testNoOverride() throws Exception {
+        try {
+            Ignite ignite = startGrid();
+
+            for (int i = 0; i < 2; i++) {
+                CacheConfiguration<Integer, Integer> cacheCfg = cacheConfiguration("cache-" + i);
+
+                cacheCfg.setAtomicityMode(TRANSACTIONAL);
+
+                ignite.createCache(cacheCfg);
+            }
+
+            ignite.cache("cache-0").put(1, 1);
+            ignite.cache("cache-1").put(1, 1);
+
+            try (Transaction tx = ignite.transactions().txStart()) {
+                ignite.cache("cache-0").put(2, 2);
+                ignite.cache("cache-0").put(3, 3);
+                ignite.cache("cache-1").put(2, 2);
+                ignite.cache("cache-1").put(3, 3);
+
+                tx.commit();
+            }
+        }
+        finally {
+            stopGrid();
+        }
+
+        assertEqualsCollections(Arrays.asList(
+            "Shared 1 START",
+            "Shared 2 START",
+
+            // Put to cache-0.
+            "Shared 1 SESSION START cache-0",
+            "Shared 2 SESSION START cache-0",
+            "Shared 1 SESSION END cache-0",
+            "Shared 2 SESSION END cache-0",
+
+            // Put to cache-1.
+            "Shared 1 SESSION START cache-1",
+            "Shared 2 SESSION START cache-1",
+            "Shared 1 SESSION END cache-1",
+            "Shared 2 SESSION END cache-1",
+
+            // Transaction.
+            "Shared 1 SESSION START cache-0",
+            "Shared 2 SESSION START cache-0",
+            "Shared 1 SESSION START cache-1",
+            "Shared 2 SESSION START cache-1",
+            "Shared 1 SESSION END cache-0",
+            "Shared 2 SESSION END cache-0",
+            "Shared 1 SESSION END cache-1",
+            "Shared 2 SESSION END cache-1",
+
+            "Shared 1 STOP",
+            "Shared 2 STOP"
+        ), evts);
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testPartialOverride() throws Exception {
+        try {
+            Ignite ignite = startGrid();
+
+            for (int i = 0; i < 2; i++) {
+                String name = "cache-" + i;
+
+                CacheConfiguration cacheCfg = cacheConfiguration(name);
+
+                cacheCfg.setAtomicityMode(TRANSACTIONAL);
+
+                if (i == 0) {
+                    cacheCfg.setCacheStoreSessionListenerFactories(
+                        new SessionListenerFactory(name + " 1"),
+                        new SessionListenerFactory(name + " 2")
+                    );
+                }
+
+                ignite.createCache(cacheCfg);
+            }
+
+            ignite.cache("cache-0").put(1, 1);
+            ignite.cache("cache-1").put(1, 1);
+
+            try (Transaction tx = ignite.transactions().txStart()) {
+                ignite.cache("cache-0").put(2, 2);
+                ignite.cache("cache-0").put(3, 3);
+                ignite.cache("cache-1").put(2, 2);
+                ignite.cache("cache-1").put(3, 3);
+
+                tx.commit();
+            }
+        }
+        finally {
+            stopGrid();
+        }
+
+        assertEqualsCollections(Arrays.asList(
+            "Shared 1 START",
+            "Shared 2 START",
+            "cache-0 1 START",
+            "cache-0 2 START",
+
+            // Put to cache-0.
+            "cache-0 1 SESSION START cache-0",
+            "cache-0 2 SESSION START cache-0",
+            "cache-0 1 SESSION END cache-0",
+            "cache-0 2 SESSION END cache-0",
+
+            // Put to cache-1.
+            "Shared 1 SESSION START cache-1",
+            "Shared 2 SESSION START cache-1",
+            "Shared 1 SESSION END cache-1",
+            "Shared 2 SESSION END cache-1",
+
+            // Transaction.
+            "cache-0 1 SESSION START cache-0",
+            "cache-0 2 SESSION START cache-0",
+            "Shared 1 SESSION START cache-1",
+            "Shared 2 SESSION START cache-1",
+            "cache-0 1 SESSION END cache-0",
+            "cache-0 2 SESSION END cache-0",
+            "Shared 1 SESSION END cache-1",
+            "Shared 2 SESSION END cache-1",
+
+            "cache-0 1 STOP",
+            "cache-0 2 STOP",
+            "Shared 1 STOP",
+            "Shared 2 STOP"
+        ), evts);
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testOverride() throws Exception {
+        try {
+            Ignite ignite = startGrid();
+
+            for (int i = 0; i < 2; i++) {
+                String name = "cache-" + i;
+
+                CacheConfiguration cacheCfg = cacheConfiguration(name);
+
+                cacheCfg.setCacheStoreSessionListenerFactories(new SessionListenerFactory(name + " 1"), new SessionListenerFactory(name + " 2"));
+
+                ignite.createCache(cacheCfg);
+            }
+
+            ignite.cache("cache-0").put(1, 1);
+            ignite.cache("cache-1").put(1, 1);
+
+            try (Transaction tx = ignite.transactions().txStart()) {
+                ignite.cache("cache-0").put(2, 2);
+                ignite.cache("cache-0").put(3, 3);
+                ignite.cache("cache-1").put(2, 2);
+                ignite.cache("cache-1").put(3, 3);
+
+                tx.commit();
+            }
+        }
+        finally {
+            stopGrid();
+        }
+
+        assertEqualsCollections(Arrays.asList(
+            "Shared 1 START",
+            "Shared 2 START",
+            "cache-0 1 START",
+            "cache-0 2 START",
+            "cache-1 1 START",
+            "cache-1 2 START",
+
+            // Put to cache-0.
+            "cache-0 1 SESSION START cache-0",
+            "cache-0 2 SESSION START cache-0",
+            "cache-0 1 SESSION END cache-0",
+            "cache-0 2 SESSION END cache-0",
+
+            // Put to cache-1.
+            "cache-1 1 SESSION START cache-1",
+            "cache-1 2 SESSION START cache-1",
+            "cache-1 1 SESSION END cache-1",
+            "cache-1 2 SESSION END cache-1",
+
+            // Transaction.
+            "cache-0 1 SESSION START cache-0",
+            "cache-0 2 SESSION START cache-0",
+            "cache-1 1 SESSION START cache-1",
+            "cache-1 2 SESSION START cache-1",
+            "cache-0 1 SESSION END cache-0",
+            "cache-0 2 SESSION END cache-0",
+            "cache-1 1 SESSION END cache-1",
+            "cache-1 2 SESSION END cache-1",
+
+            "cache-0 1 STOP",
+            "cache-0 2 STOP",
+            "cache-1 1 STOP",
+            "cache-1 2 STOP",
+            "Shared 1 STOP",
+            "Shared 2 STOP"
+        ), evts);
+    }
+
+    /**
+     * @param name Cache name.
+     * @return Cache configuration.
+     */
+    private CacheConfiguration<Integer, Integer> cacheConfiguration(String name) {
+        CacheConfiguration<Integer, Integer> cacheCfg = new CacheConfiguration<>(name);
+
+        cacheCfg.setAtomicityMode(TRANSACTIONAL);
+        cacheCfg.setCacheStoreFactory(FactoryBuilder.factoryOf(Store.class));
+        cacheCfg.setWriteThrough(true);
+
+        return cacheCfg;
+    }
+
+    /**
+     */
+    private static class SessionListener implements CacheStoreSessionListener, LifecycleAware {
+        /** */
+        private final String name;
+
+        /** */
+        @IgniteInstanceResource
+        private Ignite ignite;
+
+        /**
+         * @param name Name.
+         */
+        private SessionListener(String name) {
+            this.name = name;
+        }
+
+        /** {@inheritDoc} */
+        @Override public void start() throws IgniteException {
+            assertNotNull(ignite);
+
+            evts.add(name + " START");
+        }
+
+        /** {@inheritDoc} */
+        @Override public void stop() throws IgniteException {
+            assertNotNull(ignite);
+
+            evts.add(name + " STOP");
+        }
+
+        /** {@inheritDoc} */
+        @Override public void onSessionStart(CacheStoreSession ses) {
+            assertNotNull(ignite);
+
+            evts.add(name + " SESSION START " + ses.cacheName());
+        }
+
+        /** {@inheritDoc} */
+        @Override public void onSessionEnd(CacheStoreSession ses, boolean commit) {
+            assertNotNull(ignite);
+
+            evts.add(name + " SESSION END " + ses.cacheName());
+        }
+    }
+
+    /**
+     */
+    private static class SessionListenerFactory implements Factory<CacheStoreSessionListener> {
+        /** */
+        private String name;
+
+        /**
+         * @param name Name.
+         */
+        private SessionListenerFactory(String name) {
+            this.name = name;
+        }
+
+        @Override public CacheStoreSessionListener create() {
+            return new SessionListener(name);
+        }
+    }
+
+    /**
+     */
+    public static class Store extends CacheStoreAdapter<Integer, Integer> {
+        public Store() {
+        }
+
+        /** {@inheritDoc} */
+        @Override public Integer load(Integer key) throws CacheLoaderException {
+            return null;
+        }
+
+        /** {@inheritDoc} */
+        @Override public void write(Cache.Entry<? extends Integer, ? extends Integer> entry)
+            throws CacheWriterException {
+            // No-op.
+        }
+
+        /** {@inheritDoc} */
+        @Override public void delete(Object key) throws CacheWriterException {
+            // No-op.
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ada1b2a7/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCrossCacheTxStoreSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCrossCacheTxStoreSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCrossCacheTxStoreSelfTest.java
index f72ea47..f2de8ce 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCrossCacheTxStoreSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCrossCacheTxStoreSelfTest.java
@@ -266,30 +266,6 @@ public class IgniteCrossCacheTxStoreSelfTest extends GridCommonAbstractTest {
     }
 
     /**
-     * @param col1 Collection 1.
-     * @param col2 Collection 2.
-     */
-    private static void assertEqualsCollections(Collection<?> col1, Collection<?> col2) {
-        if (col1.size() != col2.size())
-            fail("Collections are not equal:\nExpected:\t" + col1 + "\nActual:\t" + col2);
-
-        Iterator<?> it1 = col1.iterator();
-        Iterator<?> it2 = col2.iterator();
-
-        int idx = 0;
-
-        while (it1.hasNext()) {
-            Object item1 = it1.next();
-            Object item2 = it2.next();
-
-            if (!F.eq(item1, item2))
-                fail("Collections are not equal (position " + idx + "):\nExpected: " + col1 + "\nActual:   " + col2);
-
-            idx++;
-        }
-    }
-
-    /**
      *
      */
     private static class TestStore implements CacheStore<Object, Object> {

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ada1b2a7/modules/core/src/test/java/org/apache/ignite/testframework/junits/common/GridCommonAbstractTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testframework/junits/common/GridCommonAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/testframework/junits/common/GridCommonAbstractTest.java
index 5533897..a19ea23 100644
--- a/modules/core/src/test/java/org/apache/ignite/testframework/junits/common/GridCommonAbstractTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/testframework/junits/common/GridCommonAbstractTest.java
@@ -858,4 +858,28 @@ public abstract class GridCommonAbstractTest extends GridAbstractTest {
             ccfg.getAtomicWriteOrderMode() == CacheAtomicWriteOrderMode.CLOCK)
             U.sleep(50);
     }
+
+    /**
+     * @param exp Expected.
+     * @param act Actual.
+     */
+    protected void assertEqualsCollections(Collection<?> exp, Collection<?> act) {
+        if (exp.size() != act.size())
+            fail("Collections are not equal:\nExpected:\t" + exp + "\nActual:\t" + act);
+
+        Iterator<?> it1 = exp.iterator();
+        Iterator<?> it2 = act.iterator();
+
+        int idx = 0;
+
+        while (it1.hasNext()) {
+            Object item1 = it1.next();
+            Object item2 = it2.next();
+
+            if (!F.eq(item1, item2))
+                fail("Collections are not equal (position " + idx + "):\nExpected: " + exp + "\nActual:   " + act);
+
+            idx++;
+        }
+    }
 }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ada1b2a7/modules/spring/src/main/java/org/apache/ignite/cache/store/spring/CacheSpringStoreSessionListener.java
----------------------------------------------------------------------
diff --git a/modules/spring/src/main/java/org/apache/ignite/cache/store/spring/CacheSpringStoreSessionListener.java b/modules/spring/src/main/java/org/apache/ignite/cache/store/spring/CacheSpringStoreSessionListener.java
index 81736cd..90431d7 100644
--- a/modules/spring/src/main/java/org/apache/ignite/cache/store/spring/CacheSpringStoreSessionListener.java
+++ b/modules/spring/src/main/java/org/apache/ignite/cache/store/spring/CacheSpringStoreSessionListener.java
@@ -154,7 +154,7 @@ public class CacheSpringStoreSessionListener implements CacheStoreSessionListene
 
     /** {@inheritDoc} */
     @Override public void onSessionStart(CacheStoreSession ses) {
-        if (ses.isWithinTransaction()) {
+        if (ses.isWithinTransaction() && ses.attachment() == null) {
             try {
                 TransactionDefinition def = definition(ses.transaction(), ses.cacheName());
 


Mime
View raw message