ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From a.@apache.org
Subject [42/50] [abbrv] ignite git commit: IGNITE-9129 P2P class deployment with ZK discovery fixed.
Date Fri, 03 Aug 2018 09:58:55 GMT
IGNITE-9129 P2P class deployment with ZK discovery fixed.

Signed-off-by: agura <agura@apache.org>


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

Branch: refs/heads/ignite-8446
Commit: 0e9bb1c0e50ed8a29ccd3d6a9e2bd5311b6b7912
Parents: 0b39405
Author: Pavel Kovalenko <jokserfn@gmail.com>
Authored: Thu Aug 2 16:18:44 2018 +0300
Committer: agura <agura@apache.org>
Committed: Thu Aug 2 16:19:24 2018 +0300

----------------------------------------------------------------------
 .../continuous/GridContinuousProcessor.java     | 72 +++++++++++---------
 .../GridP2PContinuousDeploymentSelfTest.java    | 70 ++++++++++++++++++-
 .../zk/ZookeeperDiscoverySpiTestSuite2.java     |  3 +
 3 files changed, 109 insertions(+), 36 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/ignite/blob/0e9bb1c0/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 2d48b7d..6723ea4 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
@@ -544,7 +544,7 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
      * @param routineInfo Routine info.
      */
     private void startDiscoveryDataRoutine(ContinuousRoutineInfo routineInfo) {
-        IgnitePredicate<ClusterNode> nodeFilter = null;
+        IgnitePredicate<ClusterNode> nodeFilter;
 
         try {
             if (routineInfo.nodeFilter != null) {
@@ -552,6 +552,8 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
 
                 ctx.resource().injectGeneric(nodeFilter);
             }
+            else
+                nodeFilter = null;
         }
         catch (IgniteCheckedException e) {
             U.error(log, "Failed to unmarshal continuous routine filter, ignore routine ["
+
@@ -561,45 +563,47 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
             return;
         }
 
-        if (nodeFilter == null || nodeFilter.apply(ctx.discovery().localNode())) {
-            GridContinuousHandler hnd;
+        ctx.discovery().localJoinFuture().listen(f -> ctx.closure().runLocalSafe(() ->
{
+            if (nodeFilter == null || nodeFilter.apply(ctx.discovery().localNode())) {
+                GridContinuousHandler hnd;
 
-            try {
-                hnd = U.unmarshal(marsh, routineInfo.hnd, U.resolveClassLoader(ctx.config()));
+                try {
+                    hnd = U.unmarshal(marsh, routineInfo.hnd, U.resolveClassLoader(ctx.config()));
 
-                if (ctx.config().isPeerClassLoadingEnabled())
-                    hnd.p2pUnmarshal(routineInfo.srcNodeId, ctx);
-            }
-            catch (IgniteCheckedException e) {
-                U.error(log, "Failed to unmarshal continuous routine handler, ignore routine
[" +
-                    "routineId=" + routineInfo.routineId +
-                    ", srcNodeId=" + routineInfo.srcNodeId + ']', e);
+                    if (ctx.config().isPeerClassLoadingEnabled())
+                        hnd.p2pUnmarshal(routineInfo.srcNodeId, ctx);
+                }
+                catch (IgniteCheckedException e) {
+                    U.error(log, "Failed to unmarshal continuous routine handler, ignore
routine [" +
+                        "routineId=" + routineInfo.routineId +
+                        ", srcNodeId=" + routineInfo.srcNodeId + ']', e);
 
-                return;
-            }
+                    return;
+                }
 
-            try {
-                registerHandler(routineInfo.srcNodeId,
-                    routineInfo.routineId,
-                    hnd,
-                    routineInfo.bufSize,
-                    routineInfo.interval,
-                    routineInfo.autoUnsubscribe,
-                    false);
-            }
-            catch (IgniteCheckedException e) {
-                U.error(log, "Failed to register continuous routine handler, ignore routine
[" +
-                    "routineId=" + routineInfo.routineId +
-                    ", srcNodeId=" + routineInfo.srcNodeId + ']', e);
+                try {
+                    registerHandler(routineInfo.srcNodeId,
+                        routineInfo.routineId,
+                        hnd,
+                        routineInfo.bufSize,
+                        routineInfo.interval,
+                        routineInfo.autoUnsubscribe,
+                        false);
+                }
+                catch (IgniteCheckedException e) {
+                    U.error(log, "Failed to register continuous routine handler, ignore routine
[" +
+                        "routineId=" + routineInfo.routineId +
+                        ", srcNodeId=" + routineInfo.srcNodeId + ']', e);
+                }
             }
-        }
-        else {
-            if (log.isDebugEnabled()) {
-                log.debug("Do not register continuous routine, rejected by node filter ["
+
-                    "routineId=" + routineInfo.routineId +
-                    ", srcNodeId=" + routineInfo.srcNodeId + ']');
+            else {
+                if (log.isDebugEnabled()) {
+                    log.debug("Do not register continuous routine, rejected by node filter
[" +
+                        "routineId=" + routineInfo.routineId +
+                        ", srcNodeId=" + routineInfo.srcNodeId + ']');
+                }
             }
-        }
+        }));
     }
 
     /**

http://git-wip-us.apache.org/repos/asf/ignite/blob/0e9bb1c0/modules/core/src/test/java/org/apache/ignite/p2p/GridP2PContinuousDeploymentSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/p2p/GridP2PContinuousDeploymentSelfTest.java
b/modules/core/src/test/java/org/apache/ignite/p2p/GridP2PContinuousDeploymentSelfTest.java
index 4ea9037..28dab3f 100644
--- a/modules/core/src/test/java/org/apache/ignite/p2p/GridP2PContinuousDeploymentSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/p2p/GridP2PContinuousDeploymentSelfTest.java
@@ -17,9 +17,14 @@
 
 package org.apache.ignite.p2p;
 
+import java.util.UUID;
 import org.apache.ignite.Ignite;
 import org.apache.ignite.configuration.CacheConfiguration;
 import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.events.Event;
+import org.apache.ignite.events.EventType;
+import org.apache.ignite.lang.IgniteBiPredicate;
+import org.apache.ignite.lang.IgnitePredicate;
 import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
 import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
 import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
@@ -49,6 +54,12 @@ public class GridP2PContinuousDeploymentSelfTest extends GridCommonAbstractTest
     /** Second test task name. */
     private static final String TEST_TASK_2 = "org.apache.ignite.tests.p2p.GridP2PContinuousDeploymentTask2";
 
+    /** Test predicate. */
+    private static final String TEST_PREDICATE = "org.apache.ignite.tests.p2p.GridEventConsumeFilter";
+
+    /** Client mode. */
+    private boolean clientMode;
+
     /** {@inheritDoc} */
     @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws
Exception {
         IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
@@ -66,6 +77,11 @@ public class GridP2PContinuousDeploymentSelfTest extends GridCommonAbstractTest
 
         cfg.setDiscoverySpi(disco);
 
+        cfg.setPeerClassLoadingEnabled(true);
+
+        if (clientMode)
+            cfg.setClientMode(true);
+
         return cfg;
     }
 
@@ -85,8 +101,13 @@ public class GridP2PContinuousDeploymentSelfTest extends GridCommonAbstractTest
     }
 
     /** {@inheritDoc} */
-    @Override protected void beforeTestsStarted() throws Exception {
-        startGridsMultiThreaded(GRID_CNT);
+    @Override protected void beforeTest() throws Exception {
+        stopAllGrids();
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void afterTest() throws Exception {
+        stopAllGrids();
     }
 
     /**
@@ -94,6 +115,8 @@ public class GridP2PContinuousDeploymentSelfTest extends GridCommonAbstractTest
      */
     @SuppressWarnings("unchecked")
     public void testDeployment() throws Exception {
+        startGridsMultiThreaded(GRID_CNT);
+
         Ignite ignite = startGrid(IGNITE_INSTANCE_NAME);
 
         Class cls = getExternalClassLoader().loadClass(TEST_TASK_1);
@@ -110,4 +133,47 @@ public class GridP2PContinuousDeploymentSelfTest extends GridCommonAbstractTest
 
         stopGrid(IGNITE_INSTANCE_NAME);
     }
+
+    /**
+     * Tests that server node joins correctly to existing cluster if it has deployed user
class with enabled P2P.
+     *
+     * @throws Exception If failed.
+     */
+    public void testServerJoinWithP2PClassDeployedInCluster() throws Exception {
+        startGrids(GRID_CNT);
+
+        ClassLoader extLdr = getExternalClassLoader();
+
+        clientMode = true;
+
+        Ignite client = startGrid(2);
+
+        Class<?> cls = extLdr.loadClass(TEST_PREDICATE);
+
+        client.events().remoteListen(
+            new IgniteBiPredicate<UUID, Event>() {
+                @Override public boolean apply(UUID uuid, Event event) {
+                    return true;
+                }
+            },
+            (IgnitePredicate<Event>) cls.newInstance(),
+            EventType.EVT_CACHE_OBJECT_PUT
+        );
+
+        clientMode = false;
+
+        Ignite srv = startGrid(3);
+
+        srv.events().remoteListen(
+            new IgniteBiPredicate<UUID, Event>() {
+                @Override public boolean apply(UUID uuid, Event event) {
+                    return true;
+                }
+            },
+            (IgnitePredicate<Event>) cls.newInstance(),
+            EventType.EVT_CACHE_OBJECT_PUT
+        );
+
+        awaitPartitionMapExchange();
+    }
 }
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/ignite/blob/0e9bb1c0/modules/zookeeper/src/test/java/org/apache/ignite/spi/discovery/zk/ZookeeperDiscoverySpiTestSuite2.java
----------------------------------------------------------------------
diff --git a/modules/zookeeper/src/test/java/org/apache/ignite/spi/discovery/zk/ZookeeperDiscoverySpiTestSuite2.java
b/modules/zookeeper/src/test/java/org/apache/ignite/spi/discovery/zk/ZookeeperDiscoverySpiTestSuite2.java
index ddb003b..012366f 100644
--- a/modules/zookeeper/src/test/java/org/apache/ignite/spi/discovery/zk/ZookeeperDiscoverySpiTestSuite2.java
+++ b/modules/zookeeper/src/test/java/org/apache/ignite/spi/discovery/zk/ZookeeperDiscoverySpiTestSuite2.java
@@ -37,6 +37,7 @@ import org.apache.ignite.internal.processors.cache.distributed.replicated.Ignite
 import org.apache.ignite.internal.processors.cache.multijvm.GridCacheAtomicMultiJvmFullApiSelfTest;
 import org.apache.ignite.internal.processors.cache.multijvm.GridCachePartitionedMultiJvmFullApiSelfTest;
 import org.apache.ignite.internal.processors.continuous.GridEventConsumeSelfTest;
+import org.apache.ignite.p2p.GridP2PContinuousDeploymentSelfTest;
 import org.apache.ignite.util.GridCommandHandlerTest;
 
 /**
@@ -91,6 +92,8 @@ public class ZookeeperDiscoverySpiTestSuite2 extends ZookeeperDiscoverySpiAbstra
 
         suite.addTestSuite(GridCommandHandlerTest.class);
 
+        suite.addTestSuite(GridP2PContinuousDeploymentSelfTest.class);
+
         return suite;
     }
 }


Mime
View raw message