ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From a.@apache.org
Subject [24/40] ignite git commit: IGNITE-3972: Fixed a bug causing continuous queries to be lost for ATOMIC cache when key's primary node leaves topology. This closes #1133.
Date Mon, 17 Oct 2016 17:59:21 GMT
IGNITE-3972: Fixed a bug causing continuous queries to be lost for ATOMIC cache when key's
primary node leaves topology. This closes #1133.


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

Branch: refs/heads/ignite-ssl-hotfix
Commit: 0659bebe04dc9c0b0a163bc95061519c553e678c
Parents: 01ca6db
Author: Andrey V. Mashenkov <andrey.mashenkov@gmail.com>
Authored: Wed Oct 12 14:49:36 2016 +0300
Committer: vozerov-gridgain <vozerov@gridgain.com>
Committed: Wed Oct 12 14:49:36 2016 +0300

----------------------------------------------------------------------
 .../continuous/CacheContinuousQueryHandler.java |   8 +-
 .../continuous/GridContinuousProcessor.java     |   6 +
 ...eContinuousQueryAsyncFailoverTxSelfTest.java |   5 +
 ...ContinuousQueryFailoverAbstractSelfTest.java | 225 ++++++++++++++++++-
 .../CacheContinuousQueryFailoverTxSelfTest.java |   5 +
 5 files changed, 237 insertions(+), 12 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/ignite/blob/0659bebe/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java
index 4b5074c..304d031 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java
@@ -595,12 +595,12 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
         assert objs != null;
         assert ctx != null;
 
-        final List<CacheContinuousQueryEntry> entries = (List<CacheContinuousQueryEntry>)objs;
-
-        if (entries.isEmpty())
+        if (objs.isEmpty())
             return;
 
         if (asyncCallback) {
+            final List<CacheContinuousQueryEntry> entries = objs instanceof List ?
(List)objs : new ArrayList(objs);
+
             IgniteStripedThreadPoolExecutor asyncPool = ctx.asyncCallbackPool();
 
             int threadId = asyncPool.threadId(entries.get(0).partition());
@@ -639,7 +639,7 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
             }, threadId);
         }
         else
-            notifyCallback0(nodeId, ctx, entries);
+            notifyCallback0(nodeId, ctx, (Collection)objs);
     }
 
     /**

http://git-wip-us.apache.org/repos/asf/ignite/blob/0659bebe/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java
index f078b1b..3a559e7 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java
@@ -820,6 +820,12 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
             if (!toSnd.isEmpty())
                 sendNotification(nodeId, routineId, null, toSnd, orderedTopic, true, null);
         }
+        else {
+            LocalRoutineInfo localRoutineInfo = locInfos.get(routineId);
+
+            if (localRoutineInfo != null)
+                localRoutineInfo.handler().notifyCallback(nodeId, routineId, objs, ctx);
+        }
     }
 
     /**

http://git-wip-us.apache.org/repos/asf/ignite/blob/0659bebe/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryAsyncFailoverTxSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryAsyncFailoverTxSelfTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryAsyncFailoverTxSelfTest.java
index 8f0bd0e..900abc8 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryAsyncFailoverTxSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryAsyncFailoverTxSelfTest.java
@@ -41,4 +41,9 @@ public class CacheContinuousQueryAsyncFailoverTxSelfTest extends CacheContinuous
     @Override protected boolean asyncCallback() {
         return true;
     }
+
+    /** {@inheritDoc} */
+    public void testNoEventLossOnTopologyChange() throws Exception {
+        fail("https://issues.apache.org/jira/browse/IGNITE-4015");
+    }
 }

http://git-wip-us.apache.org/repos/asf/ignite/blob/0659bebe/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverAbstractSelfTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverAbstractSelfTest.java
index 1376be1..1b7fe2b 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverAbstractSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverAbstractSelfTest.java
@@ -35,6 +35,7 @@ import java.util.concurrent.CyclicBarrier;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 import javax.cache.Cache;
 import javax.cache.CacheException;
@@ -51,6 +52,7 @@ import org.apache.ignite.IgniteCache;
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.IgniteException;
 import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.Ignition;
 import org.apache.ignite.cache.CacheAtomicWriteOrderMode;
 import org.apache.ignite.cache.CacheAtomicityMode;
 import org.apache.ignite.cache.CacheEntryEventSerializableFilter;
@@ -69,6 +71,7 @@ import org.apache.ignite.configuration.IgniteConfiguration;
 import org.apache.ignite.configuration.NearCacheConfiguration;
 import org.apache.ignite.internal.IgniteEx;
 import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.IgniteInterruptedCheckedException;
 import org.apache.ignite.internal.IgniteKernal;
 import org.apache.ignite.internal.managers.communication.GridIoMessage;
 import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
@@ -84,9 +87,11 @@ import org.apache.ignite.internal.util.typedef.PA;
 import org.apache.ignite.internal.util.typedef.PAX;
 import org.apache.ignite.internal.util.typedef.T2;
 import org.apache.ignite.internal.util.typedef.T3;
+import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.lang.IgniteAsyncCallback;
 import org.apache.ignite.lang.IgniteInClosure;
 import org.apache.ignite.lang.IgniteOutClosure;
+import org.apache.ignite.logger.NullLogger;
 import org.apache.ignite.plugin.extensions.communication.Message;
 import org.apache.ignite.resources.LoggerResource;
 import org.apache.ignite.spi.IgniteSpiException;
@@ -141,10 +146,10 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest extends
GridC
 
         cfg.setCommunicationSpi(commSpi);
 
-        MemoryEventStorageSpi eventSpi = new MemoryEventStorageSpi();
-        eventSpi.setExpireCount(50);
+        MemoryEventStorageSpi evtSpi = new MemoryEventStorageSpi();
+        evtSpi.setExpireCount(50);
 
-        cfg.setEventStorageSpi(eventSpi);
+        cfg.setEventStorageSpi(evtSpi);
 
         CacheConfiguration ccfg = new CacheConfiguration();
 
@@ -1180,7 +1185,7 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest extends
GridC
                     CacheEntryEvent<?, ?> e = iter.next();
 
                     if ((exp.get2() != null && e.getValue() != null && exp.get2().equals(e.getValue()))
-                            && equalOldValue(e, exp)) {
+                        && equalOldValue(e, exp)) {
                         found = true;
 
                         iter.remove();
@@ -1254,12 +1259,12 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest
extends GridC
      * @param e Event
      * @param expVals expected value
      * @return {@code True} if entries has the same key, value and oldValue. If cache start
without backups
-     *          than oldValue ignoring in comparison.
+     * than oldValue ignoring in comparison.
      */
     private boolean equalOldValue(CacheEntryEvent<?, ?> e, T3<Object, Object, Object>
expVals) {
         return (e.getOldValue() == null && expVals.get3() == null) // Both null
             || (e.getOldValue() != null && expVals.get3() != null  // Equals
-                && e.getOldValue().equals(expVals.get3()))
+            && e.getOldValue().equals(expVals.get3()))
             || (backups == 0); // If we start without backup than oldValue might be lose.
     }
 
@@ -2040,7 +2045,7 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest extends
GridC
                             bar.await(1, MINUTES);
                     }
                 }
-                catch (Exception e){
+                catch (Exception e) {
                     log.error("Failed.", e);
 
                     err = true;
@@ -2251,6 +2256,164 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest
extends GridC
     }
 
     /**
+     * This is failover test detecting CQ event loss while topology changing.
+     *
+     * @throws Exception If failed.
+     */
+    public void testNoEventLossOnTopologyChange() throws Exception {
+        final int stableNodeCnt = 1;
+
+        final int batchLoadSize = 2000;
+
+        final int restartCycles = 5;
+
+        Ignite qryClient = startGridsMultiThreaded(stableNodeCnt);
+
+        final CacheEventListener4 lsnr = new CacheEventListener4(atomicityMode() == CacheAtomicityMode.ATOMIC);
+
+        ContinuousQuery<Integer, Integer> qry = new ContinuousQuery<>();
+
+        qry.setLocalListener(lsnr);
+
+        IgniteCache<Integer, Integer> cache = qryClient.cache(null);
+
+        QueryCursor<?> cur = cache.query(qry);
+
+        int iteration = 0;
+
+        int putCnt = 0;
+
+        int ignoredDupEvts = 0;
+
+        Thread nodeRestartThread = nodeRestartThread(restartCycles, 2_000, 1_000);
+
+        try {
+            nodeRestartThread.start();
+
+            while (!Thread.interrupted() && nodeRestartThread.isAlive()) {
+                iteration++;
+
+                for (int i = 0; i < batchLoadSize; i++)
+                    cache.put(i, iteration);
+
+                putCnt += batchLoadSize;
+
+                log.info("Batch loaded. Iteration: " + iteration);
+
+                final long cnt = lsnr.count();
+
+                final long expCnt = putCnt * stableNodeCnt + ignoredDupEvts;
+
+                GridTestUtils.waitForCondition(new GridAbsPredicate() {
+                    @Override public boolean apply() {
+                        return cnt == expCnt;
+                    }
+                }, 6_000);
+
+                if (cnt != expCnt) {
+                    StringBuilder sb = new StringBuilder();
+
+                    for (int i = 0; i < batchLoadSize; i++) {
+                        Integer key = i;
+                        Integer val = cache.get(key);
+
+                        if (!F.eq(val, iteration))
+                            sb.append("\n\t").append(">>> WRONG CACHE VALUE (lost
data?) [key=").append(key)
+                                .append(", val=").append(val).append(']');
+                    }
+
+                    for (Map.Entry<Integer, Integer> entry : lsnr.eventMap().entrySet())
{
+                        Integer key = entry.getKey();
+                        Integer val = entry.getValue();
+
+                        if (!F.eq(val, iteration))
+                            sb.append("\n\t").append(">>> WRONG LISTENER VALUE (lost
event?) [key=").append(key)
+                                .append(", val=").append(val).append(']');
+                    }
+
+                    String msg = sb.toString();
+
+                    // In atomic mode CQ can receive duplicate update events if update retried
after fails.
+                    // E.g. topology change
+                    if (atomicityMode() == CacheAtomicityMode.ATOMIC && msg.isEmpty()
&& cnt > expCnt)
+                        ignoredDupEvts += cnt - expCnt;
+                    else
+                        fail("Unexpected event updates count: EXPECTED=" + expCnt + ", ACTUAL="
+ cnt + ", " +
+                            "ITERATION=" + iteration + msg);
+                }
+
+                sleep(500);
+            }
+        }
+        finally {
+            nodeRestartThread.interrupt();
+
+            cur.close();
+
+            nodeRestartThread.join(3_000);
+        }
+    }
+
+    /**
+     * Starts thread which restarts a node over and over again.
+     */
+    private Thread nodeRestartThread(final int restartCycles, final long initDelay, final
long restartDelay) {
+        Thread t = new Thread(new Runnable() {
+            public void run() {
+                sleep(initDelay);
+
+                try {
+                    for (int i = 1; i <= restartCycles && !Thread.interrupted();
i++) {
+
+                        IgniteConfiguration cfg = optimize(getConfiguration("restartNode")).
+                            setGridLogger(new NullLogger());
+
+                        log.info("Node restart cycle started: " + i);
+
+                        try (Ignite ignored = Ignition.start(cfg)) {
+                            awaitPartitionMapExchange();
+
+                            sleep(restartDelay);
+                        }
+
+                        log.info("Node restart cycle finished: " + i);
+
+                        awaitPartitionMapExchange();
+
+                        sleep(restartDelay);
+                    }
+                }
+                catch (Exception e) {
+                    log.error("Unexpected error.", e);
+                }
+            }
+        });
+
+        t.setName("flapping-node-thread");
+
+        t.setDaemon(true);
+
+        return t;
+    }
+
+    /**
+     * Sleep quietly
+     *
+     * @param sleepTime Sleep time.
+     */
+    private void sleep(long sleepTime) {
+        try {
+            if (Thread.currentThread().isInterrupted())
+                return;
+
+            U.sleep(sleepTime);
+        }
+        catch (IgniteInterruptedCheckedException e) {
+            Thread.interrupted();
+        }
+    }
+
+    /**
      *
      */
     @IgniteAsyncCallback
@@ -2363,7 +2526,7 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest extends
GridC
 
         /** {@inheritDoc} */
         @Override public synchronized void onUpdated(Iterable<CacheEntryEvent<?, ?>>
evts)
-            throws CacheEntryListenerException  {
+            throws CacheEntryListenerException {
             try {
                 for (CacheEntryEvent<?, ?> evt : evts) {
                     Integer key = (Integer)evt.getKey();
@@ -2439,6 +2602,52 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest
extends GridC
     }
 
     /**
+     * Listener.
+     */
+    private static class CacheEventListener4 implements CacheEntryUpdatedListener<Integer,
Integer> {
+        /** Listener count. */
+        private final AtomicLong cntr = new AtomicLong();
+
+        /** Listener map. */
+        private final Map<Integer, Integer> evtMap = new ConcurrentHashMap<>();
+
+        /** Atomicity mode flag. */
+        private final boolean atomicModeFlag;
+
+        /** Constructor */
+        public CacheEventListener4(boolean atomicModeFlag) {
+            this.atomicModeFlag = atomicModeFlag;
+        }
+
+        /** {@inheritDoc} */
+        @SuppressWarnings("EqualsBetweenInconvertibleTypes")
+        @Override public void onUpdated(Iterable<CacheEntryEvent<? extends Integer,
? extends Integer>> evts)
+            throws CacheEntryListenerException {
+            for (CacheEntryEvent<? extends Integer, ? extends Integer> evt : evts)
{
+                Integer prev = evtMap.put(evt.getKey(), evt.getValue());
+
+                //Atomic cache allows duplicate events if cache update operation fails, e.g.
due to topology change.
+                if (!atomicModeFlag || prev == null || !prev.equals(evt))
+                    cntr.incrementAndGet();
+            }
+        }
+
+        /**
+         * @return Events count.
+         */
+        public long count() {
+            return cntr.get();
+        }
+
+        /**
+         * @return Event map.
+         */
+        Map<Integer, Integer> eventMap() {
+            return evtMap;
+        }
+    }
+
+    /**
      *
      */
     @IgniteAsyncCallback

http://git-wip-us.apache.org/repos/asf/ignite/blob/0659bebe/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverTxSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverTxSelfTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverTxSelfTest.java
index 789a105..c5240da 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverTxSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverTxSelfTest.java
@@ -36,4 +36,9 @@ public class CacheContinuousQueryFailoverTxSelfTest extends CacheContinuousQuery
     @Override protected CacheAtomicityMode atomicityMode() {
         return TRANSACTIONAL;
     }
+
+    /** {@inheritDoc} */
+    public void testNoEventLossOnTopologyChange() throws Exception {
+        fail("https://issues.apache.org/jira/browse/IGNITE-4015");
+    }
 }


Mime
View raw message