You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by av...@apache.org on 2016/05/13 15:01:28 UTC
[1/2] ignite git commit: IGNITE-2864 Need update local store from
primary and backups
Repository: ignite
Updated Branches:
refs/heads/ignite-1.6 3319d2a7c -> ea909bc57
http://git-wip-us.apache.org/repos/asf/ignite/blob/ea909bc5/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractLocalStoreSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractLocalStoreSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractLocalStoreSelfTest.java
index 3e12ebf..7426f48 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractLocalStoreSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractLocalStoreSelfTest.java
@@ -22,6 +22,8 @@ import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
+import java.util.Random;
+import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@@ -32,6 +34,7 @@ import javax.cache.integration.CacheLoaderException;
import javax.cache.integration.CacheWriterException;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteSystemProperties;
import org.apache.ignite.Ignition;
import org.apache.ignite.cache.CacheAtomicityMode;
import org.apache.ignite.cache.CacheMemoryMode;
@@ -46,6 +49,7 @@ import org.apache.ignite.events.Event;
import org.apache.ignite.events.EventType;
import org.apache.ignite.internal.processors.cache.store.CacheLocalStore;
import org.apache.ignite.internal.util.lang.GridAbsPredicate;
+import org.apache.ignite.internal.util.typedef.G;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteBiInClosure;
import org.apache.ignite.lang.IgniteBiTuple;
@@ -55,8 +59,11 @@ import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
import org.apache.ignite.testframework.GridTestUtils;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.apache.ignite.transactions.Transaction;
import org.jetbrains.annotations.Nullable;
+import static org.apache.ignite.IgniteSystemProperties.IGNITE_LOCAL_STORE_KEEPS_PRIMARY_ONLY;
+import static org.apache.ignite.cache.CacheAtomicWriteOrderMode.PRIMARY;
import static org.apache.ignite.cache.CacheMemoryMode.OFFHEAP_TIERED;
import static org.apache.ignite.cache.CacheMode.REPLICATED;
import static org.apache.ignite.cache.CacheRebalanceMode.SYNC;
@@ -79,10 +86,28 @@ public abstract class GridCacheAbstractLocalStoreSelfTest extends GridCommonAbst
public static final TestLocalStore<Integer, Integer> LOCAL_STORE_3 = new TestLocalStore<>();
/** */
- public static final int KEYS = 1000;
+ public static final TestLocalStore<Integer, Integer> LOCAL_STORE_4 = new TestLocalStore<>();
/** */
- public static final String BACKUP_CACHE = "backup";
+ public static final TestLocalStore<Integer, Integer> LOCAL_STORE_5 = new TestLocalStore<>();
+
+ /** */
+ public static final TestLocalStore<Integer, Integer> LOCAL_STORE_6 = new TestLocalStore<>();
+
+ /** */
+ public static final int KEYS = 1025;
+
+ /** */
+ public static final String BACKUP_CACHE_1 = "backup_1";
+
+ /** */
+ public static final String BACKUP_CACHE_2 = "backup_2";
+
+ /** */
+ public static volatile boolean primaryWriteOrderMode = false;
+
+ /** */
+ public static volatile boolean near = false;
/**
*
@@ -99,11 +124,15 @@ public abstract class GridCacheAbstractLocalStoreSelfTest extends GridCommonAbst
cacheCfg.setAffinity(new RendezvousAffinityFunction());
- CacheConfiguration cacheBackupCfg = cache(gridName, BACKUP_CACHE, 2);
+ CacheConfiguration cacheBackup1Cfg = cache(gridName, BACKUP_CACHE_1, 1);
+
+ cacheBackup1Cfg.setAffinity(new RendezvousAffinityFunction());
- cacheBackupCfg.setAffinity(new RendezvousAffinityFunction());
+ CacheConfiguration cacheBackup2Cfg = cache(gridName, BACKUP_CACHE_2, 2);
- cfg.setCacheConfiguration(cacheCfg, cacheBackupCfg);
+ cacheBackup2Cfg.setAffinity(new RendezvousAffinityFunction());
+
+ cfg.setCacheConfiguration(cacheCfg, cacheBackup1Cfg, cacheBackup2Cfg);
TcpDiscoverySpi spi = new TcpDiscoverySpi();
@@ -123,6 +152,9 @@ public abstract class GridCacheAbstractLocalStoreSelfTest extends GridCommonAbst
LOCAL_STORE_1.clear();
LOCAL_STORE_2.clear();
LOCAL_STORE_3.clear();
+ LOCAL_STORE_4.clear();
+ LOCAL_STORE_5.clear();
+ LOCAL_STORE_6.clear();
}
/**
@@ -140,14 +172,24 @@ public abstract class GridCacheAbstractLocalStoreSelfTest extends GridCommonAbst
cacheCfg.setAtomicityMode(getAtomicMode());
cacheCfg.setNearConfiguration(nearConfiguration());
cacheCfg.setWriteSynchronizationMode(FULL_SYNC);
+
+ if (primaryWriteOrderMode)
+ cacheCfg.setAtomicWriteOrderMode(PRIMARY);
+
cacheCfg.setRebalanceMode(SYNC);
if (gridName.endsWith("1"))
cacheCfg.setCacheStoreFactory(singletonFactory(LOCAL_STORE_1));
else if (gridName.endsWith("2"))
cacheCfg.setCacheStoreFactory(singletonFactory(LOCAL_STORE_2));
- else
+ else if (gridName.endsWith("3"))
cacheCfg.setCacheStoreFactory(singletonFactory(LOCAL_STORE_3));
+ else if (gridName.endsWith("4"))
+ cacheCfg.setCacheStoreFactory(singletonFactory(LOCAL_STORE_4));
+ else if (gridName.endsWith("5"))
+ cacheCfg.setCacheStoreFactory(singletonFactory(LOCAL_STORE_5));
+ else
+ cacheCfg.setCacheStoreFactory(singletonFactory(LOCAL_STORE_6));
cacheCfg.setWriteThrough(true);
cacheCfg.setReadThrough(true);
@@ -164,7 +206,9 @@ public abstract class GridCacheAbstractLocalStoreSelfTest extends GridCommonAbst
/**
* @return Distribution mode.
*/
- protected abstract NearCacheConfiguration nearConfiguration();
+ protected NearCacheConfiguration nearConfiguration() {
+ return near ? new NearCacheConfiguration() : null;
+ }
/**
* @return Cache atomicity mode.
@@ -228,7 +272,7 @@ public abstract class GridCacheAbstractLocalStoreSelfTest extends GridCommonAbst
for (int i = 0; i < KEYS; i++)
cache.put(i, i);
- checkLocalStore(ignite1, LOCAL_STORE_1);
+ checkLocalStore(ignite1, LOCAL_STORE_1, null);
final AtomicInteger evtCnt = new AtomicInteger(0);
@@ -259,8 +303,322 @@ public abstract class GridCacheAbstractLocalStoreSelfTest extends GridCommonAbst
assertEquals(Ignition.allGrids().size(), 2);
- checkLocalStore(ignite1, LOCAL_STORE_1);
- checkLocalStore(ignite2, LOCAL_STORE_2);
+ checkLocalStore(ignite1, LOCAL_STORE_1, null);
+ checkLocalStore(ignite2, LOCAL_STORE_2, null);
+ }
+
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testBackupRestorePrimary() throws Exception {
+ try {
+ primaryWriteOrderMode = true;
+
+ testBackupRestore();
+ }
+ finally {
+ primaryWriteOrderMode = false;
+ }
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testBackupRestore() throws Exception {
+ Ignite ignite1 = startGrid(1);
+ Ignite ignite2 = startGrid(2);
+
+ String name = BACKUP_CACHE_2;
+
+ int key1 = -1;
+ int key2 = -1;
+
+ for (int i = 0; i < KEYS; i++) {
+ if (ignite1.affinity(name).isPrimary(ignite1.cluster().localNode(), i)) {
+ key1 = i;
+
+ break;
+ }
+ }
+
+ for (int i = 0; i < KEYS; i++) {
+ if (!ignite1.affinity(name).isPrimary(ignite1.cluster().localNode(), i)) {
+ key2 = i;
+
+ break;
+ }
+ }
+
+ assertTrue(key1 >= 0);
+ assertTrue(key2 >= 0);
+ assertNotSame(key1, key2);
+
+ assertEquals(0, ignite1.cache(name).size());
+
+ assertEquals(0, LOCAL_STORE_1.map.size());
+ assertEquals(0, LOCAL_STORE_2.map.size());
+
+ try (Transaction tx = ignite1.transactions().txStart()) {
+ ignite1.cache(name).put(key1, key1);
+ ignite1.cache(name).put(key2, key2);
+
+ Map<Integer, Integer> m = new HashMap<>();
+
+ for (int i = KEYS; i < KEYS + 100; i++)
+ m.put(i, i);
+
+ ignite1.cache(name).putAll(m);
+
+ tx.commit();
+ }
+
+ assertEquals(102, LOCAL_STORE_1.map.size());
+ assertEquals(102, LOCAL_STORE_2.map.size());
+
+ stopGrid(1);
+
+ assertEquals(1, G.allGrids().size());
+
+ assertEquals(key1, ignite2.cache(name).get(key1));
+ assertEquals(key2, ignite2.cache(name).get(key2));
+
+ for (int i = KEYS; i < KEYS + 100; i++)
+ assertEquals(i, ignite2.cache(name).get(i));
+
+ assertEquals(102, ignite2.cache(name).size());
+
+ assertEquals(102, LOCAL_STORE_1.map.size());
+ assertEquals(102, LOCAL_STORE_2.map.size());
+
+ stopGrid(2);
+
+ assertEquals(0, G.allGrids().size());
+
+ assertEquals(102, LOCAL_STORE_1.map.size());
+ assertEquals(102, LOCAL_STORE_2.map.size());
+
+ // Node restart. One node should have everything at local store.
+ ignite2 = startGrid(2);
+
+ assertEquals(1, G.allGrids().size());
+
+ assertEquals(key1, ignite2.cache(name).get(key1));
+ assertEquals(key2, ignite2.cache(name).get(key2));
+
+ for (int i = KEYS; i < KEYS + 100; i++)
+ assertEquals(i, ignite2.cache(name).get(i));
+
+ assertEquals(102, ignite2.cache(name).size());
+
+ assertEquals(102, LOCAL_STORE_1.map.size());
+ assertEquals(102, LOCAL_STORE_2.map.size());
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testLocalStoreCorrespondsAffinityWithBackups() throws Exception {
+ testLocalStoreCorrespondsAffinity(BACKUP_CACHE_2);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testLocalStoreCorrespondsAffinityWithBackup() throws Exception {
+ testLocalStoreCorrespondsAffinity(BACKUP_CACHE_1);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testLocalStoreCorrespondsAffinityNoBackups() throws Exception {
+ testLocalStoreCorrespondsAffinity(null);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ private void testLocalStoreCorrespondsAffinity(String name) throws Exception {
+ near = true;
+
+ try {
+
+ for (int i = 1; i <= 6; i++)
+ startGrid(i);
+
+ assertTrue(((IgniteCacheProxy)grid(1).cache(name)).context().isNear() ||
+ getCacheMode() == REPLICATED);
+
+ awaitPartitionMapExchange();
+
+ Random rn = new Random();
+
+ for (int i = 1; i <= 6; i++)
+ assertEquals(0, grid(i).cache(name).localSize(CachePeekMode.NEAR));
+
+ for (int i = 0; i < KEYS; i++) {
+ Ignite ignite = grid(rn.nextInt(6) + 1);
+
+ try (Transaction tx = ignite.transactions().txStart()) {
+ ignite.cache(name).put(i, i);
+
+ for (int j = 0; j < 5; j++)
+ ignite.cache(name).get(rn.nextInt(KEYS));
+
+ Map<Integer, Integer> m = new HashMap<>(5);
+
+ for (int j = 0; j < 5; j++) {
+ Integer key = rn.nextInt(KEYS);
+
+ m.put(key, key);
+ }
+
+ ignite.cache(name).putAll(m);
+
+ tx.commit();
+ }
+ }
+
+ for (int i = 1; i <= 6; i++) {
+ assertTrue(grid(i).cache(name).localSize(CachePeekMode.NEAR) > 0 ||
+ getCacheMode() == REPLICATED);
+ }
+
+ checkLocalStore(grid(1), LOCAL_STORE_1, name);
+ checkLocalStore(grid(2), LOCAL_STORE_2, name);
+ checkLocalStore(grid(3), LOCAL_STORE_3, name);
+ checkLocalStore(grid(4), LOCAL_STORE_4, name);
+ checkLocalStore(grid(5), LOCAL_STORE_5, name);
+ checkLocalStore(grid(6), LOCAL_STORE_6, name);
+
+ int fullStoreSize = LOCAL_STORE_1.map.size() +
+ LOCAL_STORE_2.map.size() +
+ LOCAL_STORE_3.map.size() +
+ LOCAL_STORE_4.map.size() +
+ LOCAL_STORE_5.map.size() +
+ LOCAL_STORE_6.map.size();
+
+ CacheConfiguration ccfg = grid(1).cache(name).getConfiguration(CacheConfiguration.class);
+
+ assertEquals(
+ getCacheMode() == REPLICATED ?
+ KEYS * 6 :
+ ccfg.getBackups() * KEYS + KEYS,
+ fullStoreSize);
+
+ }
+ finally {
+ near = false;
+ }
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testLocalStoreWithNearKeysPrimary() throws Exception {
+ try {
+ System.setProperty(IGNITE_LOCAL_STORE_KEEPS_PRIMARY_ONLY, "true");
+
+ testLocalStoreWithNearKeys();
+ }
+ finally {
+ System.setProperty(IGNITE_LOCAL_STORE_KEEPS_PRIMARY_ONLY, "false");
+ }
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testLocalStoreWithNearKeysPrimaryAndBackups() throws Exception {
+ testLocalStoreWithNearKeys();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testLocalStoreWithNearKeys() throws Exception {
+ if (getCacheMode() == REPLICATED)
+ return;
+
+ near = true;
+
+ try {
+ for (int i = 1; i <= 3; i++)
+ startGrid(i);
+
+ awaitPartitionMapExchange();
+
+ Ignite ignite = grid(1);
+
+ int k = 0;
+
+ for (int i = 1; i <= 3; i++) {
+ int kP = -1;
+ int kB = -1;
+ int kN = -1;
+
+ while (kP == -1 || kB == -1 || kN == -1) {
+ if (ignite.affinity(BACKUP_CACHE_1).isPrimary(grid(1).cluster().localNode(), k))
+ kP = k;
+
+ if (ignite.affinity(BACKUP_CACHE_1).isBackup(grid(1).cluster().localNode(), k) &&
+ ignite.affinity(BACKUP_CACHE_1).isPrimary(grid(2).cluster().localNode(), k))
+ kB = k;
+
+ if (!ignite.affinity(BACKUP_CACHE_1).isPrimaryOrBackup(grid(1).cluster().localNode(), k) &&
+ ignite.affinity(BACKUP_CACHE_1).isPrimary(grid(3).cluster().localNode(), k))
+ kN = k;
+
+ k++;
+ }
+
+ assertTrue(kP != kB && kB != kN && kN != kP);
+
+ Map<Integer, Integer> m = new HashMap<>(3);
+
+ m.put(kP, kP);
+ m.put(kB, kB);
+ m.put(kN, kN);
+
+ try (Transaction tx = grid(i).transactions().txStart()) {
+ grid(i).cache(BACKUP_CACHE_1).putAll(m);
+
+ tx.commit();
+ }
+
+ boolean locStoreBackups = !IgniteSystemProperties.getBoolean(IGNITE_LOCAL_STORE_KEEPS_PRIMARY_ONLY);
+
+ checkLocalStore(grid(1), LOCAL_STORE_1, BACKUP_CACHE_1, m.keySet(), locStoreBackups);
+ checkLocalStore(grid(2), LOCAL_STORE_2, BACKUP_CACHE_1, m.keySet(), locStoreBackups);
+ checkLocalStore(grid(3), LOCAL_STORE_3, BACKUP_CACHE_1, m.keySet(), locStoreBackups);
+ }
+
+ grid(1).cache(BACKUP_CACHE_1).removeAll();
+
+ Random rn = new Random();
+
+ for (int i = 1; i <= 3; i++) {
+ try (Transaction tx = grid(i).transactions().txStart()) {
+ Map<Integer, Integer> m = new HashMap<>(3);
+
+ for (int j = 0; j < 50; j++) {
+ m.put(rn.nextInt(1000), 1000);
+ }
+
+ grid(i).cache(BACKUP_CACHE_1).withSkipStore().putAll(m);
+
+ tx.commit();
+ }
+
+ assertTrue(LOCAL_STORE_1.map.isEmpty());
+ assertTrue(LOCAL_STORE_2.map.isEmpty());
+ assertTrue(LOCAL_STORE_3.map.isEmpty());
+ }
+ }
+ finally {
+ near = false;
+ }
}
/**
@@ -269,38 +627,63 @@ public abstract class GridCacheAbstractLocalStoreSelfTest extends GridCommonAbst
public void testBackupNode() throws Exception {
Ignite ignite1 = startGrid(1);
- IgniteCache<Object, Object> cache = ignite1.cache(BACKUP_CACHE);
+ IgniteCache<Object, Object> cache = ignite1.cache(BACKUP_CACHE_2);
for (int i = 0; i < KEYS; i++)
cache.put(i, i);
- for (int i = 0; i < KEYS; i++)
+ for (int i = 0; i < KEYS; i++) {
assertEquals(LOCAL_STORE_1.load(i).get1().intValue(), i);
+ assertNull(LOCAL_STORE_2.load(i));
+ assertNull(LOCAL_STORE_3.load(i));
+ }
- // Start 2'nd node.
- Ignite ignite2 = startGrid(2);
+ startGrid(2);
assertEquals(2, Ignition.allGrids().size());
- checkLocalStoreForBackup(ignite2, LOCAL_STORE_2);
+ for (int i = 0; i < KEYS; i++) {
+ assertEquals(LOCAL_STORE_1.load(i).get1().intValue(), i);
+ assertEquals(LOCAL_STORE_2.load(i).get1().intValue(), i);
+ assertNull(LOCAL_STORE_3.load(i));
+ }
- // Start 3'nd node.
- Ignite ignite3 = startGrid(3);
+ startGrid(3);
assertEquals(Ignition.allGrids().size(), 3);
for (int i = 0; i < KEYS; i++)
cache.put(i, i * 3);
- checkLocalStoreForBackup(ignite2, LOCAL_STORE_2);
- checkLocalStoreForBackup(ignite3, LOCAL_STORE_3);
+ for (int i = 0; i < KEYS; i++) {
+ assertEquals(LOCAL_STORE_1.load(i).get1().intValue(), i * 3);
+ assertEquals(LOCAL_STORE_2.load(i).get1().intValue(), i * 3);
+ assertEquals(LOCAL_STORE_3.load(i).get1().intValue(), i * 3);
+ }
// Stop 3'nd node.
stopGrid(3, true);
+ for (int i = 0; i < KEYS; i++)
+ cache.put(i, i * 7);
+
assertEquals(Ignition.allGrids().size(), 2);
- checkLocalStoreForBackup(ignite2, LOCAL_STORE_2);
+ for (int i = 0; i < KEYS; i++) {
+ assertEquals(LOCAL_STORE_1.load(i).get1().intValue(), i * 7);
+ assertEquals(LOCAL_STORE_2.load(i).get1().intValue(), i * 7);
+ assertEquals(LOCAL_STORE_3.load(i).get1().intValue(), i * 3);
+ }
+
+ startGrid(3);
+
+ assertEquals(Ignition.allGrids().size(), 3);
+
+ for (int i = 0; i < KEYS; i++) {
+ assertEquals(LOCAL_STORE_1.load(i).get1().intValue(), i * 7);
+ assertEquals(LOCAL_STORE_2.load(i).get1().intValue(), i * 7);
+ assertEquals(LOCAL_STORE_3.load(i).get1().intValue(), i * 7);
+ }
}
/**
@@ -315,7 +698,7 @@ public abstract class GridCacheAbstractLocalStoreSelfTest extends GridCommonAbst
for (int i = 0; i < KEYS; i++)
cache.put(i, i);
- checkLocalStore(ignite1, LOCAL_STORE_1);
+ checkLocalStore(ignite1, LOCAL_STORE_1, null);
// Push entry to swap.
for (int i = 0; i < KEYS; i++)
@@ -353,37 +736,60 @@ public abstract class GridCacheAbstractLocalStoreSelfTest extends GridCommonAbst
assertEquals(Ignition.allGrids().size(), 2);
- checkLocalStore(ignite1, LOCAL_STORE_1);
- checkLocalStore(ignite2, LOCAL_STORE_2);
+ checkLocalStore(ignite1, LOCAL_STORE_1, null);
+ checkLocalStore(ignite2, LOCAL_STORE_2, null);
}
/**
* Checks that local stores contains only primary entry.
- *
- * @param ignite Ignite.
+ * @param ignite Ignite.
* @param store Store.
+ * @param name Cache name.
*/
- private void checkLocalStore(Ignite ignite, CacheStore<Integer, IgniteBiTuple<Integer, ?>> store) {
+ private void checkLocalStore(Ignite ignite, CacheStore<Integer, IgniteBiTuple<Integer, ?>> store, String name) {
for (int i = 0; i < KEYS; i++) {
- if (ignite.affinity(null).isPrimary(ignite.cluster().localNode(), i))
+ if (ignite.affinity(name).isPrimaryOrBackup(ignite.cluster().localNode(), i))
assertEquals(store.load(i).get1().intValue(), i);
- else if (!ignite.affinity(null).isPrimaryOrBackup(ignite.cluster().localNode(), i))
+ else
assertNull(store.load(i));
}
}
/**
- * Checks that local stores contains only primary entry.
+ * Checks that local stores contains primary and backup entries.
+ * @param ignite Ignite.
+ * @param store Store.
+ * @param name Cache name.
+ * @param keys keys.
+ */
+ private void checkLocalStore(Ignite ignite, CacheStore<Integer, IgniteBiTuple<Integer, ?>> store, String name,
+ Set<Integer> keys) {
+ checkLocalStore(ignite, store, name, keys, true);
+ }
+
+ /**
+ * Checks that local stores contains primary and backup or only primary entries.
*
* @param ignite Ignite.
* @param store Store.
+ * @param name Cache name.
+ * @param keys keys.
*/
- private void checkLocalStoreForBackup(Ignite ignite, CacheStore<Integer, IgniteBiTuple<Integer, ?>> store) {
- for (int i = 0; i < KEYS; i++) {
- if (ignite.affinity(BACKUP_CACHE).isBackup(ignite.cluster().localNode(), i))
- assertEquals(store.load(i).get1().intValue(), i);
- else if (!ignite.affinity(BACKUP_CACHE).isPrimaryOrBackup(ignite.cluster().localNode(), i))
- assertNull(store.load(i).get1());
+ private void checkLocalStore(Ignite ignite, CacheStore<Integer, IgniteBiTuple<Integer, ?>> store, String name,
+ Set<Integer> keys, boolean withBackups) {
+ for (int key : keys) {
+ if (withBackups) {
+ if (ignite.affinity(name).isPrimaryOrBackup(ignite.cluster().localNode(), key))
+ assertEquals(store.load(key).get1().intValue(), key);
+ else
+ assertNull(store.load(key));
+ }
+ else {
+ if (ignite.affinity(name).isPrimary(ignite.cluster().localNode(), key))
+ assertEquals(store.load(key).get1().intValue(), key);
+ else
+ assertNull(store.load(key));
+ }
}
}
@@ -393,7 +799,7 @@ public abstract class GridCacheAbstractLocalStoreSelfTest extends GridCommonAbst
@CacheLocalStore
public static class TestLocalStore<K, V> implements CacheStore<K, IgniteBiTuple<V, ?>> {
/** */
- private Map<K, IgniteBiTuple<V, ?>> map = new ConcurrentHashMap<>();
+ private final Map<K, IgniteBiTuple<V, ?>> map = new ConcurrentHashMap<>();
/** {@inheritDoc} */
@Override public void loadCache(IgniteBiInClosure<K, IgniteBiTuple<V, ?>> clo, @Nullable Object... args)
http://git-wip-us.apache.org/repos/asf/ignite/blob/ea909bc5/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePartitionedLocalStoreSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePartitionedLocalStoreSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePartitionedLocalStoreSelfTest.java
index 854626e..d3431f7 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePartitionedLocalStoreSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePartitionedLocalStoreSelfTest.java
@@ -19,7 +19,6 @@ package org.apache.ignite.internal.processors.cache;
import org.apache.ignite.cache.CacheAtomicityMode;
import org.apache.ignite.cache.CacheMode;
-import org.apache.ignite.configuration.NearCacheConfiguration;
import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC;
import static org.apache.ignite.cache.CacheMode.PARTITIONED;
@@ -29,11 +28,6 @@ import static org.apache.ignite.cache.CacheMode.PARTITIONED;
*/
public class GridCachePartitionedLocalStoreSelfTest extends GridCacheAbstractLocalStoreSelfTest {
/** {@inheritDoc} */
- @Override protected NearCacheConfiguration nearConfiguration() {
- return null;
- }
-
- /** {@inheritDoc} */
@Override protected CacheAtomicityMode getAtomicMode() {
return ATOMIC;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/ea909bc5/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePartitionedOffHeapLocalStoreSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePartitionedOffHeapLocalStoreSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePartitionedOffHeapLocalStoreSelfTest.java
index 433ca67..8f29158 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePartitionedOffHeapLocalStoreSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePartitionedOffHeapLocalStoreSelfTest.java
@@ -19,7 +19,6 @@ package org.apache.ignite.internal.processors.cache;
import org.apache.ignite.cache.CacheAtomicityMode;
import org.apache.ignite.cache.CacheMode;
-import org.apache.ignite.configuration.NearCacheConfiguration;
import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC;
import static org.apache.ignite.cache.CacheMode.PARTITIONED;
@@ -29,11 +28,6 @@ import static org.apache.ignite.cache.CacheMode.PARTITIONED;
*/
public class GridCachePartitionedOffHeapLocalStoreSelfTest extends GridCacheAbstractLocalStoreSelfTest {
/** {@inheritDoc} */
- @Override protected NearCacheConfiguration nearConfiguration() {
- return null;
- }
-
- /** {@inheritDoc} */
@Override protected CacheAtomicityMode getAtomicMode() {
return ATOMIC;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/ea909bc5/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheReplicatedLocalStoreSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheReplicatedLocalStoreSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheReplicatedLocalStoreSelfTest.java
index 6465f89..2658e44 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheReplicatedLocalStoreSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheReplicatedLocalStoreSelfTest.java
@@ -19,7 +19,6 @@ package org.apache.ignite.internal.processors.cache;
import org.apache.ignite.cache.CacheAtomicityMode;
import org.apache.ignite.cache.CacheMode;
-import org.apache.ignite.configuration.NearCacheConfiguration;
import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC;
import static org.apache.ignite.cache.CacheMode.REPLICATED;
@@ -29,11 +28,6 @@ import static org.apache.ignite.cache.CacheMode.REPLICATED;
*/
public class GridCacheReplicatedLocalStoreSelfTest extends GridCacheAbstractLocalStoreSelfTest {
/** {@inheritDoc} */
- @Override protected NearCacheConfiguration nearConfiguration() {
- return null;
- }
-
- /** {@inheritDoc} */
@Override protected CacheAtomicityMode getAtomicMode() {
return ATOMIC;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/ea909bc5/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTxPartitionedLocalStoreSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTxPartitionedLocalStoreSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTxPartitionedLocalStoreSelfTest.java
index 4810715..272665b 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTxPartitionedLocalStoreSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTxPartitionedLocalStoreSelfTest.java
@@ -19,7 +19,6 @@ package org.apache.ignite.internal.processors.cache;
import org.apache.ignite.cache.CacheAtomicityMode;
import org.apache.ignite.cache.CacheMode;
-import org.apache.ignite.configuration.NearCacheConfiguration;
import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
import static org.apache.ignite.cache.CacheMode.PARTITIONED;
@@ -29,11 +28,6 @@ import static org.apache.ignite.cache.CacheMode.PARTITIONED;
*/
public class GridCacheTxPartitionedLocalStoreSelfTest extends GridCacheAbstractLocalStoreSelfTest {
/** {@inheritDoc} */
- @Override protected NearCacheConfiguration nearConfiguration() {
- return null;
- }
-
- /** {@inheritDoc} */
@Override protected CacheAtomicityMode getAtomicMode() {
return TRANSACTIONAL;
}
[2/2] ignite git commit: IGNITE-2864 Need update local store from
primary and backups
Posted by av...@apache.org.
IGNITE-2864 Need update local store from primary and backups
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/ea909bc5
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/ea909bc5
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/ea909bc5
Branch: refs/heads/ignite-1.6
Commit: ea909bc57361938bd72002a1aa2864a7938aa02f
Parents: 3319d2a
Author: Anton Vinogradov <av...@apache.org>
Authored: Fri May 13 18:00:46 2016 +0300
Committer: Anton Vinogradov <av...@apache.org>
Committed: Fri May 13 18:00:46 2016 +0300
----------------------------------------------------------------------
.../apache/ignite/IgniteSystemProperties.java | 3 +
.../cache/GridCacheSharedContext.java | 34 ++
.../GridDistributedTxRemoteAdapter.java | 7 +
.../dht/atomic/GridDhtAtomicCache.java | 17 +-
.../dht/atomic/GridDhtAtomicUpdateFuture.java | 6 +-
.../dht/atomic/GridDhtAtomicUpdateRequest.java | 110 +++--
.../store/GridCacheStoreManagerAdapter.java | 28 ++
.../cache/transactions/IgniteInternalTx.java | 5 +
.../cache/transactions/IgniteTxAdapter.java | 252 ++++++++++
.../IgniteTxImplicitSingleStateImpl.java | 9 +-
.../transactions/IgniteTxLocalAdapter.java | 237 ---------
.../cache/transactions/IgniteTxLocalEx.java | 5 -
.../IgniteTxRemoteSingleStateImpl.java | 17 +
.../IgniteTxRemoteStateAdapter.java | 9 -
.../transactions/IgniteTxRemoteStateImpl.java | 34 +-
.../cache/transactions/IgniteTxStateImpl.java | 2 +-
.../CacheStoreUsageMultinodeAbstractTest.java | 5 +-
.../GridCacheAbstractLocalStoreSelfTest.java | 478 +++++++++++++++++--
.../GridCachePartitionedLocalStoreSelfTest.java | 6 -
...chePartitionedOffHeapLocalStoreSelfTest.java | 6 -
.../GridCacheReplicatedLocalStoreSelfTest.java | 6 -
...ridCacheTxPartitionedLocalStoreSelfTest.java | 6 -
22 files changed, 906 insertions(+), 376 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/ea909bc5/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java b/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java
index b203a82..804cd43 100644
--- a/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java
+++ b/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java
@@ -383,6 +383,9 @@ public final class IgniteSystemProperties {
/** Maximum number of nested listener calls before listener notification becomes asynchronous. */
public static final String IGNITE_MAX_NESTED_LISTENER_CALLS = "IGNITE_MAX_NESTED_LISTENER_CALLS";
+ /** Indicating whether local store keeps primary only. Backward compatibility flag. */
+ public static final String IGNITE_LOCAL_STORE_KEEPS_PRIMARY_ONLY = "IGNITE_LOCAL_STORE_KEEPS_PRIMARY_ONLY";
+
/**
* Manages {@link OptimizedMarshaller} behavior of {@code serialVersionUID} computation for
* {@link Serializable} classes.
http://git-wip-us.apache.org/repos/asf/ignite/blob/ea909bc5/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheSharedContext.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheSharedContext.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheSharedContext.java
index 9c7a047..f27d114 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheSharedContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheSharedContext.java
@@ -24,8 +24,10 @@ import java.util.ListIterator;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.IgniteSystemProperties;
import org.apache.ignite.cache.store.CacheStoreSessionListener;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.IgniteConfiguration;
@@ -53,6 +55,8 @@ import org.apache.ignite.lang.IgniteFuture;
import org.apache.ignite.marshaller.Marshaller;
import org.jetbrains.annotations.Nullable;
+import static org.apache.ignite.IgniteSystemProperties.IGNITE_LOCAL_STORE_KEEPS_PRIMARY_ONLY;
+
/**
* Shared context.
*/
@@ -97,6 +101,12 @@ public class GridCacheSharedContext<K, V> {
/** Store session listeners. */
private Collection<CacheStoreSessionListener> storeSesLsnrs;
+ /** Local store count. */
+ private final AtomicInteger locStoreCnt;
+
+ /** Indicating whether local store keeps primary only. */
+ private final boolean locStorePrimaryOnly = IgniteSystemProperties.getBoolean(IGNITE_LOCAL_STORE_KEEPS_PRIMARY_ONLY);
+
/**
* @param kernalCtx Context.
* @param txMgr Transaction manager.
@@ -130,6 +140,8 @@ public class GridCacheSharedContext<K, V> {
txMetrics = new TransactionMetricsAdapter();
ctxMap = new ConcurrentHashMap<>();
+
+ locStoreCnt = new AtomicInteger();
}
/**
@@ -251,6 +263,11 @@ public class GridCacheSharedContext<K, V> {
", conflictingCacheName=" + existing.name() + ']');
}
+ CacheStoreManager mgr = cacheCtx.store();
+
+ if (mgr.configured() && mgr.isLocal())
+ locStoreCnt.incrementAndGet();
+
ctxMap.put(cacheCtx.cacheId(), cacheCtx);
}
@@ -262,6 +279,11 @@ public class GridCacheSharedContext<K, V> {
ctxMap.remove(cacheId, cacheCtx);
+ CacheStoreManager mgr = cacheCtx.store();
+
+ if (mgr.configured() && mgr.isLocal())
+ locStoreCnt.decrementAndGet();
+
// Safely clean up the message listeners.
ioMgr.removeHandlers(cacheId);
}
@@ -480,6 +502,13 @@ public class GridCacheSharedContext<K, V> {
}
/**
+ * @return Count of caches with configured local stores.
+ */
+ public int getLocalStoreCount() {
+ return locStoreCnt.get();
+ }
+
+ /**
* @param nodeId Node ID.
* @return Node or {@code null}.
*/
@@ -487,6 +516,11 @@ public class GridCacheSharedContext<K, V> {
return kernalCtx.discovery().node(nodeId);
}
+ /** Indicating whether local store keeps primary only. */
+ public boolean localStorePrimaryOnly() {
+ return locStorePrimaryOnly;
+ }
+
/**
* Gets grid logger for given class.
*
http://git-wip-us.apache.org/repos/asf/ignite/blob/ea909bc5/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
index 2bb6bf3..c56d1f7 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
@@ -455,6 +455,8 @@ public class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
AffinityTopologyVersion topVer = topologyVersion();
+ batchStoreCommit(writeMap().values());
+
// Node that for near transactions we grab all entries.
for (IgniteTxEntry txEntry : (near() ? allEntries() : writeEntries())) {
GridCacheContext cacheCtx = txEntry.context();
@@ -773,6 +775,11 @@ public class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
return explicitVers == null ? Collections.<GridCacheVersion>emptyList() : explicitVers;
}
+ /** {@inheritDoc} */
+ @Override public void commitError(Throwable e) {
+ // No-op.
+ }
+
/**
* Adds explicit version if there is one.
*
http://git-wip-us.apache.org/repos/asf/ignite/blob/ea909bc5/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
index ee909e6..1f01a76 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
@@ -141,7 +141,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
@GridToStringExclude
private CI2<GridNearAtomicUpdateRequest, GridNearAtomicUpdateResponse> updateReplyClos;
- /** Pending */
+ /** Pending */
private ConcurrentMap<UUID, DeferredResponseBuffer> pendingResponses = new ConcurrentHashMap8<>();
/** */
@@ -1566,10 +1566,11 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
GridCacheReturn retVal = null;
- if (keys.size() > 1 && // Several keys ...
- writeThrough() && !req.skipStore() && // and store is enabled ...
- !ctx.store().isLocal() && // and this is not local store ...
- !ctx.dr().receiveEnabled() // and no DR.
+ if (keys.size() > 1 && // Several keys ...
+ writeThrough() && !req.skipStore() && // and store is enabled ...
+ !ctx.store().isLocal() && // and this is not local store ...
+ // (conflict resolver should be used for local store)
+ !ctx.dr().receiveEnabled() // and no DR.
) {
// This method can only be used when there are no replicated entries in the batch.
UpdateBatchResult updRes = updateWithBatch(node,
@@ -2201,7 +2202,8 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
op,
writeVal,
req.invokeArguments(),
- primary && writeThrough() && !req.skipStore(),
+ (primary || (ctx.store().isLocal() && !ctx.shared().localStorePrimaryOnly()))
+ && writeThrough() && !req.skipStore(),
!req.skipStore(),
sndPrevVal || req.returnValue(),
req.keepBinary(),
@@ -2964,7 +2966,8 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
op,
op == TRANSFORM ? entryProcessor : val,
op == TRANSFORM ? req.invokeArguments() : null,
- /*write-through*/false,
+ /*write-through*/(ctx.store().isLocal() && !ctx.shared().localStorePrimaryOnly())
+ && writeThrough() && !req.skipStore(),
/*read-through*/false,
/*retval*/false,
req.keepBinary(),
http://git-wip-us.apache.org/repos/asf/ignite/blob/ea909bc5/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
index fa69f10..570f043 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
@@ -259,7 +259,8 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
this.updateReq.taskNameHash(),
forceTransformBackups ? this.updateReq.invokeArguments() : null,
cctx.deploymentEnabled(),
- this.updateReq.keepBinary());
+ this.updateReq.keepBinary(),
+ this.updateReq.skipStore());
mappings.put(nodeId, updateReq);
}
@@ -320,7 +321,8 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
this.updateReq.taskNameHash(),
forceTransformBackups ? this.updateReq.invokeArguments() : null,
cctx.deploymentEnabled(),
- this.updateReq.keepBinary());
+ this.updateReq.keepBinary(),
+ this.updateReq.skipStore());
mappings.put(nodeId, updateReq);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/ea909bc5/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java
index 6b050b1..b1a94de 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java
@@ -46,6 +46,8 @@ import org.apache.ignite.plugin.extensions.communication.MessageWriter;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
+import static org.apache.ignite.internal.processors.cache.GridCacheUtils.SKIP_STORE_FLAG_MASK;
+
/**
* Lite dht cache backup update request.
*/
@@ -57,6 +59,7 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
public static final int CACHE_MSG_IDX = nextIndexId();
/** Node ID. */
+ @GridDirectTransient
private UUID nodeId;
/** Future version. */
@@ -159,6 +162,11 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
private boolean keepBinary;
/**
+ * Additional flags.
+ */
+ private byte flags;
+
+ /**
* Empty constructor required by {@link Externalizable}.
*/
public GridDhtAtomicUpdateRequest() {
@@ -192,7 +200,8 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
int taskNameHash,
Object[] invokeArgs,
boolean addDepInfo,
- boolean keepBinary
+ boolean keepBinary,
+ boolean skipStore
) {
assert invokeArgs == null || forceTransformBackups;
@@ -209,6 +218,9 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
this.addDepInfo = addDepInfo;
this.keepBinary = keepBinary;
+ if (skipStore)
+ flags = (byte)(flags | SKIP_STORE_FLAG_MASK);
+
keys = new ArrayList<>();
partIds = new ArrayList<>();
@@ -465,6 +477,13 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
}
/**
+ * @return Skip write-through to a persistent storage.
+ */
+ public boolean skipStore() {
+ return (flags & SKIP_STORE_FLAG_MASK) == SKIP_STORE_FLAG_MASK;
+ }
+
+ /**
* @param updCntr Update counter.
* @return Update counter.
*/
@@ -729,121 +748,121 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
writer.incrementState();
case 6:
- if (!writer.writeBoolean("forceTransformBackups", forceTransformBackups))
+ if (!writer.writeByte("flags", flags))
return false;
writer.incrementState();
case 7:
- if (!writer.writeMessage("futVer", futVer))
+ if (!writer.writeBoolean("forceTransformBackups", forceTransformBackups))
return false;
writer.incrementState();
case 8:
- if (!writer.writeObjectArray("invokeArgsBytes", invokeArgsBytes, MessageCollectionItemType.BYTE_ARR))
+ if (!writer.writeMessage("futVer", futVer))
return false;
writer.incrementState();
case 9:
- if (!writer.writeBoolean("keepBinary", keepBinary))
+ if (!writer.writeObjectArray("invokeArgsBytes", invokeArgsBytes, MessageCollectionItemType.BYTE_ARR))
return false;
writer.incrementState();
case 10:
- if (!writer.writeCollection("keys", keys, MessageCollectionItemType.MSG))
+ if (!writer.writeBoolean("keepBinary", keepBinary))
return false;
writer.incrementState();
case 11:
- if (!writer.writeCollection("nearEntryProcessorsBytes", nearEntryProcessorsBytes, MessageCollectionItemType.BYTE_ARR))
+ if (!writer.writeCollection("keys", keys, MessageCollectionItemType.MSG))
return false;
writer.incrementState();
case 12:
- if (!writer.writeMessage("nearExpireTimes", nearExpireTimes))
+ if (!writer.writeCollection("nearEntryProcessorsBytes", nearEntryProcessorsBytes, MessageCollectionItemType.BYTE_ARR))
return false;
writer.incrementState();
case 13:
- if (!writer.writeCollection("nearKeys", nearKeys, MessageCollectionItemType.MSG))
+ if (!writer.writeMessage("nearExpireTimes", nearExpireTimes))
return false;
writer.incrementState();
case 14:
- if (!writer.writeMessage("nearTtls", nearTtls))
+ if (!writer.writeCollection("nearKeys", nearKeys, MessageCollectionItemType.MSG))
return false;
writer.incrementState();
case 15:
- if (!writer.writeCollection("nearVals", nearVals, MessageCollectionItemType.MSG))
+ if (!writer.writeMessage("nearTtls", nearTtls))
return false;
writer.incrementState();
case 16:
- if (!writer.writeCollection("prevVals", prevVals, MessageCollectionItemType.MSG))
+ if (!writer.writeCollection("nearVals", nearVals, MessageCollectionItemType.MSG))
return false;
writer.incrementState();
case 17:
- if (!writer.writeUuid("subjId", subjId))
+ if (!writer.writeCollection("prevVals", prevVals, MessageCollectionItemType.MSG))
return false;
writer.incrementState();
case 18:
- if (!writer.writeByte("syncMode", syncMode != null ? (byte)syncMode.ordinal() : -1))
+ if (!writer.writeUuid("subjId", subjId))
return false;
writer.incrementState();
case 19:
- if (!writer.writeInt("taskNameHash", taskNameHash))
+ if (!writer.writeByte("syncMode", syncMode != null ? (byte)syncMode.ordinal() : -1))
return false;
writer.incrementState();
case 20:
- if (!writer.writeMessage("topVer", topVer))
+ if (!writer.writeInt("taskNameHash", taskNameHash))
return false;
writer.incrementState();
case 21:
- if (!writer.writeMessage("ttls", ttls))
+ if (!writer.writeMessage("topVer", topVer))
return false;
writer.incrementState();
case 22:
- if (!writer.writeMessage("updateCntrs", updateCntrs))
+ if (!writer.writeMessage("ttls", ttls))
return false;
writer.incrementState();
case 23:
- if (!writer.writeCollection("vals", vals, MessageCollectionItemType.MSG))
+ if (!writer.writeMessage("updateCntrs", updateCntrs))
return false;
writer.incrementState();
case 24:
- if (!writer.writeMessage("writeVer", writeVer))
+ if (!writer.writeCollection("vals", vals, MessageCollectionItemType.MSG))
return false;
writer.incrementState();
case 25:
- if (!writer.writeCollection("partIds", partIds, MessageCollectionItemType.INT))
+ if (!writer.writeMessage("writeVer", writeVer))
return false;
writer.incrementState();
@@ -889,7 +908,7 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
reader.incrementState();
case 6:
- forceTransformBackups = reader.readBoolean("forceTransformBackups");
+ flags = reader.readByte("flags");
if (!reader.isLastRead())
return false;
@@ -897,7 +916,7 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
reader.incrementState();
case 7:
- futVer = reader.readMessage("futVer");
+ forceTransformBackups = reader.readBoolean("forceTransformBackups");
if (!reader.isLastRead())
return false;
@@ -905,7 +924,7 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
reader.incrementState();
case 8:
- invokeArgsBytes = reader.readObjectArray("invokeArgsBytes", MessageCollectionItemType.BYTE_ARR, byte[].class);
+ futVer = reader.readMessage("futVer");
if (!reader.isLastRead())
return false;
@@ -913,7 +932,7 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
reader.incrementState();
case 9:
- keepBinary = reader.readBoolean("keepBinary");
+ invokeArgsBytes = reader.readObjectArray("invokeArgsBytes", MessageCollectionItemType.BYTE_ARR, byte[].class);
if (!reader.isLastRead())
return false;
@@ -921,7 +940,7 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
reader.incrementState();
case 10:
- keys = reader.readCollection("keys", MessageCollectionItemType.MSG);
+ keepBinary = reader.readBoolean("keepBinary");
if (!reader.isLastRead())
return false;
@@ -929,7 +948,7 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
reader.incrementState();
case 11:
- nearEntryProcessorsBytes = reader.readCollection("nearEntryProcessorsBytes", MessageCollectionItemType.BYTE_ARR);
+ keys = reader.readCollection("keys", MessageCollectionItemType.MSG);
if (!reader.isLastRead())
return false;
@@ -937,7 +956,7 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
reader.incrementState();
case 12:
- nearExpireTimes = reader.readMessage("nearExpireTimes");
+ nearEntryProcessorsBytes = reader.readCollection("nearEntryProcessorsBytes", MessageCollectionItemType.BYTE_ARR);
if (!reader.isLastRead())
return false;
@@ -945,7 +964,7 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
reader.incrementState();
case 13:
- nearKeys = reader.readCollection("nearKeys", MessageCollectionItemType.MSG);
+ nearExpireTimes = reader.readMessage("nearExpireTimes");
if (!reader.isLastRead())
return false;
@@ -953,7 +972,7 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
reader.incrementState();
case 14:
- nearTtls = reader.readMessage("nearTtls");
+ nearKeys = reader.readCollection("nearKeys", MessageCollectionItemType.MSG);
if (!reader.isLastRead())
return false;
@@ -961,7 +980,7 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
reader.incrementState();
case 15:
- nearVals = reader.readCollection("nearVals", MessageCollectionItemType.MSG);
+ nearTtls = reader.readMessage("nearTtls");
if (!reader.isLastRead())
return false;
@@ -969,7 +988,7 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
reader.incrementState();
case 16:
- prevVals = reader.readCollection("prevVals", MessageCollectionItemType.MSG);
+ nearVals = reader.readCollection("nearVals", MessageCollectionItemType.MSG);
if (!reader.isLastRead())
return false;
@@ -977,7 +996,7 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
reader.incrementState();
case 17:
- subjId = reader.readUuid("subjId");
+ prevVals = reader.readCollection("prevVals", MessageCollectionItemType.MSG);
if (!reader.isLastRead())
return false;
@@ -985,27 +1004,27 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
reader.incrementState();
case 18:
- byte syncModeOrd;
-
- syncModeOrd = reader.readByte("syncMode");
+ subjId = reader.readUuid("subjId");
if (!reader.isLastRead())
return false;
- syncMode = CacheWriteSynchronizationMode.fromOrdinal(syncModeOrd);
-
reader.incrementState();
case 19:
- taskNameHash = reader.readInt("taskNameHash");
+ byte syncModeOrd;
+
+ syncModeOrd = reader.readByte("syncMode");
if (!reader.isLastRead())
return false;
+ syncMode = CacheWriteSynchronizationMode.fromOrdinal(syncModeOrd);
+
reader.incrementState();
case 20:
- topVer = reader.readMessage("topVer");
+ taskNameHash = reader.readInt("taskNameHash");
if (!reader.isLastRead())
return false;
@@ -1013,7 +1032,7 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
reader.incrementState();
case 21:
- ttls = reader.readMessage("ttls");
+ topVer = reader.readMessage("topVer");
if (!reader.isLastRead())
return false;
@@ -1021,7 +1040,7 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
reader.incrementState();
case 22:
- updateCntrs = reader.readMessage("updateCntrs");
+ ttls = reader.readMessage("ttls");
if (!reader.isLastRead())
return false;
@@ -1029,7 +1048,7 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
reader.incrementState();
case 23:
- vals = reader.readCollection("vals", MessageCollectionItemType.MSG);
+ updateCntrs = reader.readMessage("updateCntrs");
if (!reader.isLastRead())
return false;
@@ -1037,7 +1056,7 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
reader.incrementState();
case 24:
- writeVer = reader.readMessage("writeVer");
+ vals = reader.readCollection("vals", MessageCollectionItemType.MSG);
if (!reader.isLastRead())
return false;
@@ -1045,12 +1064,13 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
reader.incrementState();
case 25:
- partIds = reader.readCollection("partIds", MessageCollectionItemType.INT);
+ writeVer = reader.readMessage("writeVer");
if (!reader.isLastRead())
return false;
reader.incrementState();
+
}
return reader.afterMessageRead(GridDhtAtomicUpdateRequest.class);
http://git-wip-us.apache.org/repos/asf/ignite/blob/ea909bc5/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 467f4b9..cd0c50f 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
@@ -31,9 +31,11 @@ import javax.cache.Cache;
import javax.cache.integration.CacheLoaderException;
import javax.cache.integration.CacheWriterException;
import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteSystemProperties;
import org.apache.ignite.cache.store.CacheStore;
import org.apache.ignite.cache.store.CacheStoreSession;
import org.apache.ignite.cache.store.CacheStoreSessionListener;
+import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.processors.cache.CacheEntryImpl;
@@ -61,11 +63,14 @@ import org.apache.ignite.internal.util.typedef.internal.SB;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteBiInClosure;
import org.apache.ignite.lang.IgniteBiTuple;
+import org.apache.ignite.lang.IgniteProductVersion;
import org.apache.ignite.lifecycle.LifecycleAware;
import org.apache.ignite.transactions.Transaction;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
+import static org.apache.ignite.IgniteSystemProperties.IGNITE_LOCAL_STORE_KEEPS_PRIMARY_ONLY;
+
/**
* Store manager.
*/
@@ -74,6 +79,13 @@ public abstract class GridCacheStoreManagerAdapter extends GridCacheManagerAdapt
/** */
private static final int SES_ATTR = GridMetadataAwareAdapter.EntryKey.CACHE_STORE_MANAGER_KEY.key();
+ /**
+ * Behavior can be changed by setting {@link IgniteSystemProperties#IGNITE_LOCAL_STORE_KEEPS_PRIMARY_ONLY} property
+ * to {@code True}.
+ */
+ private static final IgniteProductVersion LOCAL_STORE_KEEPS_PRIMARY_AND_BACKUPS_SINCE =
+ IgniteProductVersion.fromString("1.5.22");
+
/** */
protected CacheStore<Object, Object> store;
@@ -217,6 +229,22 @@ public abstract class GridCacheStoreManagerAdapter extends GridCacheManagerAdapt
globalSesLsnrs = true;
}
+
+ if (isLocal()) {
+ for (ClusterNode node : cctx.kernalContext().cluster().get().forRemotes().nodes()) {
+ if (LOCAL_STORE_KEEPS_PRIMARY_AND_BACKUPS_SINCE.compareTo(node.version()) > 0 &&
+ !IgniteSystemProperties.getBoolean(IGNITE_LOCAL_STORE_KEEPS_PRIMARY_ONLY)) {
+ IgniteProductVersion v = LOCAL_STORE_KEEPS_PRIMARY_AND_BACKUPS_SINCE;
+
+ log.warning("Since Ignite " + v.major() + "." + v.minor() + "." + v.maintenance() +
+ " Local Store keeps primary and backup partitions. " +
+ "To keep primary partitions only please set system property " +
+ IGNITE_LOCAL_STORE_KEEPS_PRIMARY_ONLY + " to 'true'.");
+
+ break;
+ }
+ }
+ }
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/ea909bc5/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteInternalTx.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteInternalTx.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteInternalTx.java
index 87b8e07..8c0425d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteInternalTx.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteInternalTx.java
@@ -698,4 +698,9 @@ public interface IgniteInternalTx extends AutoCloseable, GridTimeoutObject {
* @param topVer New topology version.
*/
public void onRemap(AffinityTopologyVersion topVer);
+
+ /**
+ * @param e Commit error.
+ */
+ public void commitError(Throwable e);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/ea909bc5/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java
index ebddd50..86a1989 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java
@@ -22,10 +22,13 @@ import java.io.IOException;
import java.io.ObjectInput;
import java.io.ObjectOutput;
import java.io.ObjectStreamException;
+import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -41,6 +44,7 @@ import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.CacheInvokeEntry;
+import org.apache.ignite.internal.processors.cache.CacheLazyEntry;
import org.apache.ignite.internal.processors.cache.CacheObject;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
@@ -50,6 +54,7 @@ import org.apache.ignite.internal.processors.cache.GridCacheOperation;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearCacheEntry;
+import org.apache.ignite.internal.processors.cache.store.CacheStoreManager;
import org.apache.ignite.internal.processors.cache.version.GridCachePlainVersionedEntry;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersionConflictContext;
@@ -1212,6 +1217,248 @@ public abstract class IgniteTxAdapter extends GridMetadataAwareAdapter implement
}
/**
+ * @param stores Store managers.
+ * @return If {@code isWriteToStoreFromDht} value same for all stores.
+ */
+ protected boolean isWriteToStoreFromDhtValid(Collection<CacheStoreManager> stores) {
+ if (stores != null && !stores.isEmpty()) {
+ boolean exp = F.first(stores).isWriteToStoreFromDht();
+
+ for (CacheStoreManager store : stores) {
+ if (store.isWriteToStoreFromDht() != exp)
+ return false;
+ }
+ }
+
+ return true;
+ }
+
+ /**
+ * @param stores Store managers.
+ * @param commit Commit flag.
+ * @throws IgniteCheckedException In case of error.
+ */
+ protected void sessionEnd(Collection<CacheStoreManager> stores, boolean commit) throws IgniteCheckedException {
+ Iterator<CacheStoreManager> it = stores.iterator();
+
+ while (it.hasNext()) {
+ CacheStoreManager store = it.next();
+
+ store.sessionEnd(this, commit, !it.hasNext());
+ }
+ }
+
+ /**
+ * Performs batch database operations. This commit must be called
+ * before cache update. This way if there is a DB failure,
+ * cache transaction can still be rolled back.
+ *
+ * @param writeEntries Transaction write set.
+ * @throws IgniteCheckedException If batch update failed.
+ */
+ @SuppressWarnings({"CatchGenericClass"})
+ protected void batchStoreCommit(Iterable<IgniteTxEntry> writeEntries) throws IgniteCheckedException {
+ if (!storeEnabled() || internal() ||
+ (!local() && near())) // No need to work with local store at GridNearTxRemote.
+ return;
+
+ Collection<CacheStoreManager> stores = txState().stores(cctx);
+
+ if (stores == null || stores.isEmpty())
+ return;
+
+ assert isWriteToStoreFromDhtValid(stores) : "isWriteToStoreFromDht can't be different within one transaction";
+
+ CacheStoreManager first = F.first(stores);
+
+ boolean isWriteToStoreFromDht = first.isWriteToStoreFromDht();
+
+ if ((local() || first.isLocal()) && (near() || isWriteToStoreFromDht)) {
+ try {
+ if (writeEntries != null) {
+ Map<Object, IgniteBiTuple<Object, GridCacheVersion>> putMap = null;
+ List<Object> rmvCol = null;
+ CacheStoreManager writeStore = null;
+
+ boolean skipNonPrimary = near() && isWriteToStoreFromDht;
+
+ for (IgniteTxEntry e : writeEntries) {
+ boolean skip = e.skipStore();
+
+ if (!skip && skipNonPrimary) {
+ skip = e.cached().isNear() ||
+ e.cached().detached() ||
+ !e.context().affinity().primary(e.cached().partition(), topologyVersion()).isLocal();
+ }
+
+ if (!skip && !local() && // Update local store at backups only if needed.
+ cctx.localStorePrimaryOnly())
+ skip = true;
+
+ if (skip)
+ continue;
+
+ boolean intercept = e.context().config().getInterceptor() != null;
+
+ if (intercept || !F.isEmpty(e.entryProcessors()))
+ e.cached().unswap(false);
+
+ IgniteBiTuple<GridCacheOperation, CacheObject> res = applyTransformClosures(e, false);
+
+ GridCacheContext cacheCtx = e.context();
+
+ GridCacheOperation op = res.get1();
+ KeyCacheObject key = e.key();
+ CacheObject val = res.get2();
+ GridCacheVersion ver = writeVersion();
+
+ if (op == CREATE || op == UPDATE) {
+ // Batch-process all removes if needed.
+ if (rmvCol != null && !rmvCol.isEmpty()) {
+ assert writeStore != null;
+
+ writeStore.removeAll(this, rmvCol);
+
+ // Reset.
+ rmvCol.clear();
+
+ writeStore = null;
+ }
+
+ // Batch-process puts if cache ID has changed.
+ if (writeStore != null && writeStore != cacheCtx.store()) {
+ if (putMap != null && !putMap.isEmpty()) {
+ writeStore.putAll(this, putMap);
+
+ // Reset.
+ putMap.clear();
+ }
+
+ writeStore = null;
+ }
+
+ if (intercept) {
+ Object interceptorVal = cacheCtx.config().getInterceptor().onBeforePut(
+ new CacheLazyEntry(
+ cacheCtx,
+ key,
+ e.cached().rawGetOrUnmarshal(true),
+ e.keepBinary()),
+ cacheCtx.cacheObjectContext().unwrapBinaryIfNeeded(val, e.keepBinary(), false));
+
+ if (interceptorVal == null)
+ continue;
+
+ val = cacheCtx.toCacheObject(cacheCtx.unwrapTemporary(interceptorVal));
+ }
+
+ if (writeStore == null)
+ writeStore = cacheCtx.store();
+
+ if (writeStore.isWriteThrough()) {
+ if (putMap == null)
+ putMap = new LinkedHashMap<>(writeMap().size(), 1.0f);
+
+ putMap.put(key, F.<Object, GridCacheVersion>t(val, ver));
+ }
+ }
+ else if (op == DELETE) {
+ // Batch-process all puts if needed.
+ if (putMap != null && !putMap.isEmpty()) {
+ assert writeStore != null;
+
+ writeStore.putAll(this, putMap);
+
+ // Reset.
+ putMap.clear();
+
+ writeStore = null;
+ }
+
+ if (writeStore != null && writeStore != cacheCtx.store()) {
+ if (rmvCol != null && !rmvCol.isEmpty()) {
+ writeStore.removeAll(this, rmvCol);
+
+ // Reset.
+ rmvCol.clear();
+ }
+
+ writeStore = null;
+ }
+
+ if (intercept) {
+ IgniteBiTuple<Boolean, Object> t = cacheCtx.config().getInterceptor().onBeforeRemove(
+ new CacheLazyEntry(cacheCtx, key, e.cached().rawGetOrUnmarshal(true), e.keepBinary()));
+
+ if (cacheCtx.cancelRemove(t))
+ continue;
+ }
+
+ if (writeStore == null)
+ writeStore = cacheCtx.store();
+
+ if (writeStore.isWriteThrough()) {
+ if (rmvCol == null)
+ rmvCol = new ArrayList<>();
+
+ rmvCol.add(key);
+ }
+ }
+ else if (log.isDebugEnabled())
+ log.debug("Ignoring NOOP entry for batch store commit: " + e);
+ }
+
+ if (putMap != null && !putMap.isEmpty()) {
+ assert rmvCol == null || rmvCol.isEmpty();
+ assert writeStore != null;
+
+ // Batch put at the end of transaction.
+ writeStore.putAll(this, putMap);
+ }
+
+ if (rmvCol != null && !rmvCol.isEmpty()) {
+ assert putMap == null || putMap.isEmpty();
+ assert writeStore != null;
+
+ // Batch remove at the end of transaction.
+ writeStore.removeAll(this, rmvCol);
+ }
+ }
+
+ // Commit while locks are held.
+ sessionEnd(stores, true);
+ }
+ catch (IgniteCheckedException ex) {
+ commitError(ex);
+
+ errorWhenCommitting();
+
+ // Safe to remove transaction from committed tx list because nothing was committed yet.
+ cctx.tm().removeCommittedTx(this);
+
+ throw ex;
+ }
+ catch (Throwable ex) {
+ commitError(ex);
+
+ errorWhenCommitting();
+
+ // Safe to remove transaction from committed tx list because nothing was committed yet.
+ cctx.tm().removeCommittedTx(this);
+
+ if (ex instanceof Error)
+ throw (Error)ex;
+
+ throw new IgniteCheckedException("Failed to commit transaction to database: " + this, ex);
+ }
+ finally {
+ if (isRollbackOnly())
+ sessionEnd(stores, false);
+ }
+ }
+ }
+
+ /**
* @param txEntry Entry to process.
* @param metrics {@code True} if metrics should be updated.
* @return Tuple containing transformation results.
@@ -1794,6 +2041,11 @@ public abstract class IgniteTxAdapter extends GridMetadataAwareAdapter implement
}
/** {@inheritDoc} */
+ @Override public void commitError(Throwable e) {
+ throw new IllegalStateException("Deserialized transaction can only be used as read-only.");
+ }
+
+ /** {@inheritDoc} */
@Override public boolean empty() {
throw new IllegalStateException("Deserialized transaction can only be used as read-only.");
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/ea909bc5/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxImplicitSingleStateImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxImplicitSingleStateImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxImplicitSingleStateImpl.java
index 965ef2a..d6a32c2 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxImplicitSingleStateImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxImplicitSingleStateImpl.java
@@ -164,13 +164,8 @@ public class IgniteTxImplicitSingleStateImpl extends IgniteTxLocalStateAdapter {
CacheStoreManager store = cacheCtx.store();
- if (store.configured()) {
- HashSet<CacheStoreManager> set = new HashSet<>(3, 0.75f);
-
- set.add(store);
-
- return set;
- }
+ if (store.configured())
+ return Collections.singleton(store);
return null;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/ea909bc5/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
index e8aa7c8..d492051 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
@@ -25,7 +25,6 @@ import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashMap;
-import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
@@ -42,7 +41,6 @@ import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.CacheEntryPredicate;
import org.apache.ignite.internal.processors.cache.CacheInvokeEntry;
-import org.apache.ignite.internal.processors.cache.CacheLazyEntry;
import org.apache.ignite.internal.processors.cache.CacheObject;
import org.apache.ignite.internal.processors.cache.CacheOperationContext;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
@@ -625,209 +623,6 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
return cacheCtx.cache().entryEx(key.key(), topVer);
}
- /**
- * Performs batch database operations. This commit must be called
- * before {@link #userCommit()}. This way if there is a DB failure,
- * cache transaction can still be rolled back.
- *
- * @param writeEntries Transaction write set.
- * @throws IgniteCheckedException If batch update failed.
- */
- @SuppressWarnings({"CatchGenericClass"})
- protected void batchStoreCommit(Iterable<IgniteTxEntry> writeEntries) throws IgniteCheckedException {
- if (!storeEnabled() || internal())
- return;
-
- Collection<CacheStoreManager> stores = txState.stores(cctx);
-
- if (stores == null || stores.isEmpty())
- return;
-
- assert isWriteToStoreFromDhtValid(stores) : "isWriteToStoreFromDht can't be different within one transaction";
-
- boolean isWriteToStoreFromDht = F.first(stores).isWriteToStoreFromDht();
-
- if (near() || isWriteToStoreFromDht) {
- try {
- if (writeEntries != null) {
- Map<Object, IgniteBiTuple<Object, GridCacheVersion>> putMap = null;
- List<Object> rmvCol = null;
- CacheStoreManager writeStore = null;
-
- boolean skipNonPrimary = near() && isWriteToStoreFromDht;
-
- for (IgniteTxEntry e : writeEntries) {
- boolean skip = e.skipStore();
-
- if (!skip && skipNonPrimary) {
- skip = e.cached().isNear() ||
- e.cached().detached() ||
- !e.context().affinity().primary(e.cached().partition(), topologyVersion()).isLocal();
- }
-
- if (skip)
- continue;
-
- boolean intercept = e.context().config().getInterceptor() != null;
-
- if (intercept || !F.isEmpty(e.entryProcessors()))
- e.cached().unswap(false);
-
- IgniteBiTuple<GridCacheOperation, CacheObject> res = applyTransformClosures(e, false);
-
- GridCacheContext cacheCtx = e.context();
-
- GridCacheOperation op = res.get1();
- KeyCacheObject key = e.key();
- CacheObject val = res.get2();
- GridCacheVersion ver = writeVersion();
-
- if (op == CREATE || op == UPDATE) {
- // Batch-process all removes if needed.
- if (rmvCol != null && !rmvCol.isEmpty()) {
- assert writeStore != null;
-
- writeStore.removeAll(this, rmvCol);
-
- // Reset.
- rmvCol.clear();
-
- writeStore = null;
- }
-
- // Batch-process puts if cache ID has changed.
- if (writeStore != null && writeStore != cacheCtx.store()) {
- if (putMap != null && !putMap.isEmpty()) {
- writeStore.putAll(this, putMap);
-
- // Reset.
- putMap.clear();
- }
-
- writeStore = null;
- }
-
- if (intercept) {
- Object interceptorVal = cacheCtx.config().getInterceptor().onBeforePut(
- new CacheLazyEntry(
- cacheCtx,
- key,
- e.cached().rawGetOrUnmarshal(true),
- e.keepBinary()),
- cacheCtx.cacheObjectContext().unwrapBinaryIfNeeded(val, e.keepBinary(), false));
-
- if (interceptorVal == null)
- continue;
-
- val = cacheCtx.toCacheObject(cacheCtx.unwrapTemporary(interceptorVal));
- }
-
- if (writeStore == null)
- writeStore = cacheCtx.store();
-
- if (writeStore.isWriteThrough()) {
- if (putMap == null)
- putMap = new LinkedHashMap<>(writeMap().size(), 1.0f);
-
- putMap.put(key, F.<Object, GridCacheVersion>t(val, ver));
- }
- }
- else if (op == DELETE) {
- // Batch-process all puts if needed.
- if (putMap != null && !putMap.isEmpty()) {
- assert writeStore != null;
-
- writeStore.putAll(this, putMap);
-
- // Reset.
- putMap.clear();
-
- writeStore = null;
- }
-
- if (writeStore != null && writeStore != cacheCtx.store()) {
- if (rmvCol != null && !rmvCol.isEmpty()) {
- writeStore.removeAll(this, rmvCol);
-
- // Reset.
- rmvCol.clear();
- }
-
- writeStore = null;
- }
-
- if (intercept) {
- IgniteBiTuple<Boolean, Object> t = cacheCtx.config().getInterceptor().onBeforeRemove(
- new CacheLazyEntry(cacheCtx, key, e.cached().rawGetOrUnmarshal(true), e.keepBinary()));
-
- if (cacheCtx.cancelRemove(t))
- continue;
- }
-
- if (writeStore == null)
- writeStore = cacheCtx.store();
-
- if (writeStore.isWriteThrough()) {
- if (rmvCol == null)
- rmvCol = new ArrayList<>();
-
- rmvCol.add(key);
- }
- }
- else if (log.isDebugEnabled())
- log.debug("Ignoring NOOP entry for batch store commit: " + e);
- }
-
- if (putMap != null && !putMap.isEmpty()) {
- assert rmvCol == null || rmvCol.isEmpty();
- assert writeStore != null;
-
- // Batch put at the end of transaction.
- writeStore.putAll(this, putMap);
- }
-
- if (rmvCol != null && !rmvCol.isEmpty()) {
- assert putMap == null || putMap.isEmpty();
- assert writeStore != null;
-
- // Batch remove at the end of transaction.
- writeStore.removeAll(this, rmvCol);
- }
- }
-
- // Commit while locks are held.
- sessionEnd(stores, true);
- }
- catch (IgniteCheckedException ex) {
- commitError(ex);
-
- errorWhenCommitting();
-
- // Safe to remove transaction from committed tx list because nothing was committed yet.
- cctx.tm().removeCommittedTx(this);
-
- throw ex;
- }
- catch (Throwable ex) {
- commitError(ex);
-
- errorWhenCommitting();
-
- // Safe to remove transaction from committed tx list because nothing was committed yet.
- cctx.tm().removeCommittedTx(this);
-
- if (ex instanceof Error)
- throw (Error)ex;
-
- throw new IgniteCheckedException("Failed to commit transaction to database: " + this, ex);
- }
- finally {
- if (isRollbackOnly())
- sessionEnd(stores, false);
- }
- }
- }
-
/** {@inheritDoc} */
@SuppressWarnings({"CatchGenericClass"})
@Override public void userCommit() throws IgniteCheckedException {
@@ -1310,21 +1105,6 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
}
/**
- * @param stores Store managers.
- * @param commit Commit flag.
- * @throws IgniteCheckedException In case of error.
- */
- private void sessionEnd(Collection<CacheStoreManager> stores, boolean commit) throws IgniteCheckedException {
- Iterator<CacheStoreManager> it = stores.iterator();
-
- while (it.hasNext()) {
- CacheStoreManager store = it.next();
-
- store.sessionEnd(this, commit, !it.hasNext());
- }
- }
-
- /**
* @param entry Entry.
* @return {@code True} if local node is current primary for given entry.
*/
@@ -4040,23 +3820,6 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
}
/**
- * @param stores Store managers.
- * @return If {@code isWriteToStoreFromDht} value same for all stores.
- */
- private boolean isWriteToStoreFromDhtValid(Collection<CacheStoreManager> stores) {
- if (stores != null && !stores.isEmpty()) {
- boolean exp = F.first(stores).isWriteToStoreFromDht();
-
- for (CacheStoreManager store : stores) {
- if (store.isWriteToStoreFromDht() != exp)
- return false;
- }
- }
-
- return true;
- }
-
- /**
* Post-lock closure alias.
*
* @param <T> Return type.
http://git-wip-us.apache.org/repos/asf/ignite/blob/ea909bc5/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalEx.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalEx.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalEx.java
index 6741885..9fb3558 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalEx.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalEx.java
@@ -48,11 +48,6 @@ public interface IgniteTxLocalEx extends IgniteInternalTx {
@Nullable public Throwable commitError();
/**
- * @param e Commit error.
- */
- public void commitError(Throwable e);
-
- /**
* @throws IgniteCheckedException If commit failed.
*/
public void userCommit() throws IgniteCheckedException;
http://git-wip-us.apache.org/repos/asf/ignite/blob/ea909bc5/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxRemoteSingleStateImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxRemoteSingleStateImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxRemoteSingleStateImpl.java
index d356fb6..ae3c187 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxRemoteSingleStateImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxRemoteSingleStateImpl.java
@@ -21,6 +21,8 @@ import java.util.Collection;
import java.util.Collections;
import java.util.Map;
import java.util.Set;
+import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
+import org.apache.ignite.internal.processors.cache.store.CacheStoreManager;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.jetbrains.annotations.Nullable;
@@ -112,4 +114,19 @@ public class IgniteTxRemoteSingleStateImpl extends IgniteTxRemoteStateAdapter {
public String toString() {
return S.toString(IgniteTxRemoteSingleStateImpl.class, this);
}
+
+ /** {@inheritDoc} */
+ @Override public Collection<CacheStoreManager> stores(GridCacheSharedContext cctx) {
+ if (entry == null)
+ return null;
+
+ CacheStoreManager store = entry.context().store();
+
+ if (store.configured()
+ && store.isLocal()) { // Only local stores take part at tx on backup node.
+ return Collections.singleton(store);
+ }
+
+ return null;
+ }
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/ea909bc5/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxRemoteStateAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxRemoteStateAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxRemoteStateAdapter.java
index 79b4a74..1c2ccbe 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxRemoteStateAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxRemoteStateAdapter.java
@@ -17,13 +17,11 @@
package org.apache.ignite.internal.processors.cache.transactions;
-import java.util.Collection;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.cache.CacheWriteSynchronizationMode;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTopologyFuture;
-import org.apache.ignite.internal.processors.cache.store.CacheStoreManager;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.jetbrains.annotations.Nullable;
@@ -100,13 +98,6 @@ public abstract class IgniteTxRemoteStateAdapter implements IgniteTxRemoteState
}
/** {@inheritDoc} */
- @Override public Collection<CacheStoreManager> stores(GridCacheSharedContext cctx) {
- assert false;
-
- return null;
- }
-
- /** {@inheritDoc} */
@Nullable @Override public GridCacheContext singleCacheContext(GridCacheSharedContext cctx) {
return null;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/ea909bc5/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxRemoteStateImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxRemoteStateImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxRemoteStateImpl.java
index 3335b44..4f46831 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxRemoteStateImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxRemoteStateImpl.java
@@ -17,13 +17,15 @@
package org.apache.ignite.internal.processors.cache.transactions;
+import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
-
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
+import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
+import org.apache.ignite.internal.processors.cache.store.CacheStoreManager;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.S;
@@ -145,4 +147,34 @@ public class IgniteTxRemoteStateImpl extends IgniteTxRemoteStateAdapter {
public String toString() {
return S.toString(IgniteTxRemoteStateImpl.class, this);
}
+
+ /** {@inheritDoc} */
+ @Override public Collection<CacheStoreManager> stores(GridCacheSharedContext cctx) {
+ int locStoreCnt = cctx.getLocalStoreCount();
+
+ if (locStoreCnt > 0 && !writeMap.isEmpty()) {
+ Collection<CacheStoreManager> stores = null;
+
+ for (IgniteTxEntry e : writeMap.values()) {
+ if (e.skipStore())
+ continue;
+
+ CacheStoreManager store = e.context().store();
+
+ if (store.configured() && store.isLocal()) {
+ if (stores == null)
+ stores = new ArrayList<>(locStoreCnt);
+
+ stores.add(store);
+
+ if (stores.size() == locStoreCnt)
+ break;
+ }
+ }
+
+ return stores;
+ }
+
+ return null;
+ }
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/ea909bc5/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxStateImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxStateImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxStateImpl.java
index c826de1..c116d0d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxStateImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxStateImpl.java
@@ -50,8 +50,8 @@ import static org.apache.ignite.cache.CacheWriteSynchronizationMode.PRIMARY_SYNC
public class IgniteTxStateImpl extends IgniteTxLocalStateAdapter {
/** Active cache IDs. */
private GridLongList activeCacheIds = new GridLongList();
- /** Per-transaction read map. */
+ /** Per-transaction read map. */
@GridToStringInclude
protected Map<IgniteTxKey, IgniteTxEntry> txMap;
http://git-wip-us.apache.org/repos/asf/ignite/blob/ea909bc5/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheStoreUsageMultinodeAbstractTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheStoreUsageMultinodeAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheStoreUsageMultinodeAbstractTest.java
index 9c4b7b2..eaae2f9 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheStoreUsageMultinodeAbstractTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheStoreUsageMultinodeAbstractTest.java
@@ -242,9 +242,10 @@ public abstract class CacheStoreUsageMultinodeAbstractTest extends GridCommonAbs
assertTrue("Store is not updated", wait);
- assertEquals("Write on wrong node: " + writeMap, 1, writeMap.size());
+ assertEquals("Write on wrong node: " + writeMap, locStore ? 2 : 1, writeMap.size());
- assertEquals(expNode, writeMap.keySet().iterator().next());
+ if (!locStore)
+ assertEquals(expNode, writeMap.keySet().iterator().next());
writeMap.clear();
}