ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [03/23] incubator-ignite git commit: # ignite-537
Date Fri, 10 Apr 2015 21:57:11 GMT
# ignite-537


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

Branch: refs/heads/ignite-sprint-3
Commit: 88e1ff8810ebc617e08c8eea554ec451f2df603c
Parents: 5029aeb
Author: sboikov <semen.boikov@inria.fr>
Authored: Thu Apr 9 23:49:42 2015 +0300
Committer: sboikov <semen.boikov@inria.fr>
Committed: Thu Apr 9 23:49:42 2015 +0300

----------------------------------------------------------------------
 .../discovery/GridDiscoveryManager.java         |  41 +---
 .../continuous/GridContinuousProcessor.java     |  25 ++-
 .../spi/discovery/DiscoverySpiDataExchange.java |   6 +-
 .../discovery/tcp/TcpClientDiscoverySpi.java    |  11 +-
 .../spi/discovery/tcp/TcpDiscoverySpi.java      | 223 +++++++++++++++++--
 .../discovery/tcp/TcpDiscoverySpiAdapter.java   |  13 --
 .../continuous/GridEventConsumeSelfTest.java    |   5 +-
 .../discovery/AbstractDiscoverySelfTest.java    |   5 +-
 .../tcp/TcpDiscoverySpiStartStopSelfTest.java   |   5 +-
 .../junits/spi/GridSpiAbstractTest.java         |   5 +-
 10 files changed, 253 insertions(+), 86 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/88e1ff88/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
index 7662bcc..04ff423 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
@@ -36,7 +36,6 @@ import org.apache.ignite.internal.util.typedef.*;
 import org.apache.ignite.internal.util.typedef.internal.*;
 import org.apache.ignite.internal.util.worker.*;
 import org.apache.ignite.lang.*;
-import org.apache.ignite.marshaller.jdk.*;
 import org.apache.ignite.plugin.security.*;
 import org.apache.ignite.plugin.segmentation.*;
 import org.apache.ignite.spi.*;
@@ -172,9 +171,6 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi>
{
     /** Map of dynamic cache filters. */
     private Map<String, CachePredicate> registeredCaches = new HashMap<>();
 
-    /** */
-    private JdkMarshaller jdkMarsh = new JdkMarshaller();
-
     /** @param ctx Context. */
     public GridDiscoveryManager(GridKernalContext ctx) {
         super(ctx, ctx.config().getDiscoverySpi());
@@ -421,10 +417,10 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi>
{
         });
 
         spi.setDataExchange(new DiscoverySpiDataExchange() {
-            @Override public Map<Integer, byte[]> collect(UUID nodeId) {
+            @Override public Map<Integer, Object> collect(UUID nodeId) {
                 assert nodeId != null;
 
-                Map<Integer, byte[]> data = new HashMap<>();
+                Map<Integer, Object> data = new HashMap<>();
 
                 for (GridComponent comp : ctx.components()) {
                     Object compData = comp.collectDiscoveryData(nodeId);
@@ -432,48 +428,29 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi>
{
                     if (compData != null) {
                         assert comp.discoveryDataType() != null;
 
-                        try {
-                            byte[] bytes = jdkMarsh.marshal(compData);
-
-                            data.put(comp.discoveryDataType().ordinal(), bytes);
-                        }
-                        catch (IgniteCheckedException e) {
-                            U.error(log, "Failed to marshal discovery data " +
-                                "[comp=" + comp + ", data=" + compData + ']', e);
-                        }
+                        data.put(comp.discoveryDataType().ordinal(), compData);
                     }
                 }
 
                 return data;
             }
 
-            @Override public void onExchange(UUID joiningNodeId,
-                UUID nodeId,
-                Map<Integer, byte[]> data,
-                ClassLoader clsLdr) {
-                for (Map.Entry<Integer, byte[]> entry : data.entrySet()) {
+            @Override public void onExchange(UUID joiningNodeId, UUID nodeId, Map<Integer,
Object> data) {
+                for (Map.Entry<Integer, Object> e : data.entrySet()) {
                     GridComponent comp = null;
 
                     for (GridComponent c : ctx.components()) {
-                        if (c.discoveryDataType() != null && c.discoveryDataType().ordinal()
== entry.getKey()) {
+                        if (c.discoveryDataType() != null && c.discoveryDataType().ordinal()
== e.getKey()) {
                             comp = c;
 
                             break;
                         }
                     }
 
-                    if (comp != null) {
-                        try {
-                            Object compData = jdkMarsh.unmarshal(entry.getValue(), clsLdr);
-
-                            comp.onDiscoveryDataReceived(joiningNodeId, nodeId, compData);
-                        }
-                        catch (IgniteCheckedException e) {
-                            U.error(log, "Failed to unmarshal discovery data for component:
"  + comp, e);
-                        }
-                    }
+                    if (comp != null)
+                        comp.onDiscoveryDataReceived(joiningNodeId, nodeId, e.getValue());
                     else
-                        U.warn(log, "Received discovery data for unknown component: " + entry.getKey());
+                        U.warn(log, "Received discovery data for unknown component: " + e.getKey());
                 }
             }
         });

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/88e1ff88/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 54da8e6..99b9c0c 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
@@ -30,6 +30,7 @@ import org.apache.ignite.internal.processors.cache.*;
 import org.apache.ignite.internal.processors.timeout.*;
 import org.apache.ignite.internal.util.*;
 import org.apache.ignite.internal.util.future.*;
+import org.apache.ignite.internal.util.lang.*;
 import org.apache.ignite.internal.util.tostring.*;
 import org.apache.ignite.internal.util.typedef.*;
 import org.apache.ignite.internal.util.typedef.internal.*;
@@ -1559,7 +1560,7 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
     /**
      * Discovery data.
      */
-    private static class DiscoveryData implements Externalizable {
+    private static class DiscoveryData implements Externalizable, GridPeerDeployAware {
         /** */
         private static final long serialVersionUID = 0L;
 
@@ -1577,6 +1578,16 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
             // No-op.
         }
 
+        @Override
+        public Class<?> deployClass() {
+            return U.peerDeployAware0(items).deployClass();
+        }
+
+        @Override
+        public ClassLoader classLoader() {
+            return U.peerDeployAware0(items).classLoader();
+        }
+
         /**
          * @param nodeId Node ID.
          */
@@ -1616,7 +1627,7 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
     /**
      * Discovery data item.
      */
-    private static class DiscoveryDataItem implements Externalizable {
+    private static class DiscoveryDataItem implements Externalizable, GridPeerDeployAware
{
         /** */
         private static final long serialVersionUID = 0L;
 
@@ -1666,6 +1677,16 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
             this.interval = interval;
         }
 
+        @Override
+        public Class<?> deployClass() {
+            return prjPred.getClass();
+        }
+
+        @Override
+        public ClassLoader classLoader() {
+            return prjPred.getClass().getClassLoader();
+        }
+
         /** {@inheritDoc} */
         @Override public void writeExternal(ObjectOutput out) throws IOException {
             U.writeUuid(out, routineId);

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/88e1ff88/modules/core/src/main/java/org/apache/ignite/spi/discovery/DiscoverySpiDataExchange.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/DiscoverySpiDataExchange.java
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/DiscoverySpiDataExchange.java
index 06f1af8..1f64c87 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/DiscoverySpiDataExchange.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/DiscoverySpiDataExchange.java
@@ -33,7 +33,7 @@ public interface DiscoverySpiDataExchange {
      * @param joiningNodeId ID of new node that joins topology.
      * @return Collection of discovery data objects from different components.
      */
-    public Map<Integer, byte[]> collect(UUID joiningNodeId);
+    public Map<Integer, Object> collect(UUID joiningNodeId);
 
     /**
      * Notifies discovery manager about data received from remote node.
@@ -41,10 +41,8 @@ public interface DiscoverySpiDataExchange {
      * @param joiningNodeId Joining node ID.
      * @param nodeId Remote node ID for which data is provided.
      * @param data Collection of marshalled discovery data objects from different components.
-     * @param clsLdr Class loader to use for discovery data unmarshalling.
      */
     public void onExchange(UUID joiningNodeId,
         UUID nodeId,
-        Map<Integer, byte[]> data,
-        ClassLoader clsLdr);
+        Map<Integer, Object> data);
 }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/88e1ff88/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpi.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpi.java
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpi.java
index 6ced987..f613f4a 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpi.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpi.java
@@ -897,10 +897,9 @@ public class TcpClientDiscoverySpi extends TcpDiscoverySpiAdapter implements
Tcp
 
                         if (dataMap != null) {
                             for (Map.Entry<UUID, Map<Integer, byte[]>> entry
: dataMap.entrySet()) {
-                                exchange.onExchange(newNodeId,
-                                    entry.getKey(),
-                                    entry.getValue(),
-                                    exchangeClassLoader(newNodeId));
+//                                exchange.onExchange(newNodeId,
+//                                    entry.getKey(),
+//                                    entry.getValue());
                             }
                         }
 
@@ -923,8 +922,8 @@ public class TcpClientDiscoverySpi extends TcpDiscoverySpiAdapter implements
Tcp
 
                     Map<Integer, byte[]> data = msg.newNodeDiscoveryData();
 
-                    if (data != null)
-                        exchange.onExchange(newNodeId, newNodeId, data, exchangeClassLoader(newNodeId));
+//                    if (data != null)
+//                        exchange.onExchange(newNodeId, newNodeId, data);
                 }
             }
         }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/88e1ff88/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
index ef22291..3fa6fd1 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
@@ -1377,6 +1377,87 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements
TcpDiscov
             log.debug("Discovery SPI has been connected to topology with order: " + locNode.internalOrder());
     }
 
+    /** */
+    private GridConcurrentHashSet<ClassLoader> exchangeLdrs = new GridConcurrentHashSet();
+
+    /**
+     * @param nodeId Node ID>
+     * @return Marshalled exchange data.
+     * @throws IgniteSpiException If failed.
+     */
+    protected Map<Integer, byte[]> collectExchangeData(UUID nodeId) throws IgniteSpiException
{
+        boolean p2pEnaled = ignite.configuration().isPeerClassLoadingEnabled();
+
+        Map<Integer, Object> data = exchange.collect(nodeId);
+
+        Map<Integer, byte[]> data0 = U.newHashMap(data.size());
+
+        for (Map.Entry<Integer, Object> entry : data.entrySet()) {
+            try {
+                byte[] bytes = marsh.marshal(entry.getValue());
+
+                data0.put(entry.getKey(), bytes);
+
+                if (p2pEnaled) {
+                    ClassLoader ldr;
+
+                    if (entry.getValue() instanceof GridPeerDeployAware)
+                        ldr = ((GridPeerDeployAware)entry.getValue()).classLoader();
+                    else
+                        ldr = entry.getValue().getClass().getClassLoader();
+
+                    log.info("Add loader: " + entry.getValue() + " " + ldr);
+
+                    if (ldr != null) {
+                        if (exchangeLdrs.add(ldr))
+                            log.info("Added loader: " + ldr);
+                    }
+                }
+            }
+            catch (IgniteCheckedException e) {
+                U.error(log, "Failed to marshal discovery data " +
+                    "[comp=" + entry.getKey() + ", data=" + entry.getValue() + ']', e);
+
+                throw new IgniteSpiException("Failed to marshal discovery data.", e);
+            }
+        }
+
+        return data0;
+    }
+
+    /**
+     * @param joiningNode Joining node.
+     * @param nodeId Remote node ID for which data is provided.
+     * @param data Collection of marshalled discovery data objects from different components.
+     */
+    protected void onExchange(TcpDiscoveryNode joiningNode,
+        UUID nodeId,
+        Map<Integer, byte[]> data) {
+        TcpDiscoveryNode node;
+
+        if (joiningNode.id().equals(nodeId))
+            node = joiningNode;
+        else
+            node = ring.node(nodeId);
+
+        ClassLoader clsLdr = exchangeClassLoader(node);
+
+        Map<Integer, Object> data0 = U.newHashMap(data.size());
+
+        for (Map.Entry<Integer, byte[]> entry : data.entrySet()) {
+            try {
+                Object compData = marsh.unmarshal(entry.getValue(), clsLdr);
+
+                data0.put(entry.getKey(), compData);
+            }
+            catch (IgniteCheckedException e) {
+                U.error(log, "Failed to unmarshal discovery data for component: "  + entry.getKey(),
e);
+            }
+        }
+
+        exchange.onExchange(joiningNode.id(), nodeId, data0);
+    }
+
     /**
      * Tries to send join request message to a random node presenting in topology.
      * Address is provided by {@link TcpDiscoveryIpFinder} and message is
@@ -1388,7 +1469,7 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements
TcpDiscov
     @SuppressWarnings({"BusyWait"})
     private boolean sendJoinRequestMessage() throws IgniteSpiException {
         TcpDiscoveryAbstractMessage joinReq = new TcpDiscoveryJoinRequestMessage(locNode,
-            exchange.collect(getLocalNodeId()));
+            collectExchangeData(getLocalNodeId()));
 
         // Time when it has been detected, that addresses from IP finder do not respond.
         long noResStart = 0;
@@ -2198,7 +2279,11 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements
TcpDiscov
     private TcpDiscoveryGetClassResponse processGetClassRequest(TcpDiscoveryGetClassRequest
req) {
         assert !F.isEmpty(req.className()) : req;
 
-        InputStream in = getClass().getResourceAsStream(req.className());
+        String rsrc = U.classNameToResourceName(req.className());
+
+        log.info("Get class request: " + req.className() + " " + rsrc);
+
+        InputStream in = locLdr.getResourceAsStream(rsrc);
 
         byte[] clsBytes = null;
         String err = null;
@@ -3591,9 +3676,9 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements
TcpDiscov
                     Map<Integer, byte[]> data = msg.newNodeDiscoveryData();
 
                     if (data != null)
-                        exchange.onExchange(node.id(), node.id(), data, exchangeClassLoader(node.id()));
+                        onExchange(node, node.id(), data);
 
-                    msg.addDiscoveryData(locNodeId, exchange.collect(node.id()));
+                    msg.addDiscoveryData(locNodeId, collectExchangeData(node.id()));
                 }
 
                 if (log.isDebugEnabled())
@@ -3662,12 +3747,8 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements
TcpDiscov
 
                 // Notify outside of synchronized block.
                 if (dataMap != null) {
-                    for (Map.Entry<UUID, Map<Integer, byte[]>> entry : dataMap.entrySet())
{
-                        exchange.onExchange(node.id(),
-                            entry.getKey(),
-                            entry.getValue(),
-                            exchangeClassLoader(node.id()));
-                    }
+                    for (Map.Entry<UUID, Map<Integer, byte[]>> entry : dataMap.entrySet())
+                        onExchange(node, entry.getKey(), entry.getValue());
                 }
             }
 
@@ -5221,30 +5302,109 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements
TcpDiscov
             U.closeQuiet(sock);
         }
     }
+
+    /** */
+    private ConcurrentMap<UUID, DiscoveryDeploymentClassLoader> ldrs = new ConcurrentHashMap<>();
+
+    /** */
+    private LocalDeploymentClassLoader locLdr = new LocalDeploymentClassLoader(null);
+
+    /**
+     *
+     */
+    class LocalDeploymentClassLoader extends ClassLoader {
+        public LocalDeploymentClassLoader(ClassLoader parent) {
+            super(parent);
+        }
+
+        @Override
+        public InputStream getResourceAsStream(String name) {
+            log.info("Local getResourceAsStream: " + name);
+
+            for (ClassLoader ldr : exchangeLdrs) {
+                InputStream  in = ldr.getResourceAsStream(name);
+
+                if (in != null)
+                    return in;
+            }
+
+            return super.getResourceAsStream(name);
+        }
+
+        @Override public Class<?> loadClass(String name) throws ClassNotFoundException
{
+            log.info("Local loadClass: " + name);
+
+            for (ClassLoader ldr : exchangeLdrs) {
+                try {
+                    return ldr.loadClass(name);
+                }
+                catch (ClassNotFoundException ignore) {
+                }
+            }
+
+            return super.loadClass(name);
+        }
+    }
+
+    /**
+     * @param nodeId Node ID.
+     * @return Class loader for exchange data unmarshalling.
+     */
+    protected ClassLoader exchangeClassLoader(TcpDiscoveryNode node) {
+        assert ignite != null;
+
+        UUID nodeId = node.id();
+
+        if (!ignite.configuration().isPeerClassLoadingEnabled())
+            return null;
+
+        if (nodeId.equals(getLocalNodeId()))
+            return locLdr;
+
+        DiscoveryDeploymentClassLoader ldr = ldrs.get(nodeId);
+
+        if (ldr == null)
+            ldr = F.addIfAbsent(ldrs, nodeId, new DiscoveryDeploymentClassLoader(node));
+
+        return ldr;
+    }
+
     /**
      *
      */
     private class DiscoveryDeploymentClassLoader extends ClassLoader {
         /** */
-        private final UUID nodeId;
+        private final TcpDiscoveryNode node;
 
         /** */
         private Socket sock;
 
         /**
-         * @param nodeId Node ID.
+         * @param node Node.
          */
-        public DiscoveryDeploymentClassLoader(UUID nodeId) {
-            this.nodeId = nodeId;
+        public DiscoveryDeploymentClassLoader(TcpDiscoveryNode node) {
+            assert !node.id().equals(getLocalNodeId());
+
+            this.node = node;
+        }
+
+        @Override protected Class<?> loadClass(String name, boolean resolve) throws
ClassNotFoundException {
+            // log.info("P2p load class: " + name);
+
+            return super.loadClass(name, resolve);
         }
 
         /** {@inheritDoc} */
         @Override protected Class<?> findClass(String name) throws ClassNotFoundException
{
+            log.info("P2p find class: " + name);
+
             TcpDiscoveryGetClassResponse res = requestClass(name);
 
             if (res.error() != null)
                 throw new ClassNotFoundException(res.error());
 
+            log.info("P2p loaded: " + name);
+
             assert res.classBytes() != null;
 
             Class<?> cls = defineClass(name, res.classBytes(), 0, res.classBytes().length);
@@ -5272,16 +5432,45 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements
TcpDiscov
                 return res;
             }
             catch (IOException | IgniteCheckedException e) {
+                e.printStackTrace();
+
                 throw new ClassNotFoundException("Failed to load class: " + name, e);
             }
         }
 
         private Socket connect() {
             if (sock == null) {
-                TcpDiscoveryNode node = ring.node(nodeId);
+                for (InetSocketAddress addr : getNodeAddresses(node, U.sameMacs(locNode,
node))) {
+                    sock = connect(addr);
+
+                    if (sock != null)
+                        break;
+                }
+            }
+
+            return sock;
+        }
+
+        private Socket connect(InetSocketAddress addr) {
+            UUID locNodeId  = getLocalNodeId();
+
+            Socket sock = null;
+
+            for (int i = 0; i < reconCnt; i++) {
+                try {
+                    sock = openSocket(addr);
 
-                if (node == null)
-                    return null;
+                    writeToSocket(sock, new TcpDiscoveryHandshakeRequest(locNodeId));
+
+                    TcpDiscoveryHandshakeResponse res = readMessage(sock, null, netTimeout);
+
+                    break;
+                }
+                catch (IOException | IgniteCheckedException e) {
+                    e.printStackTrace();
+
+                    U.closeQuiet(sock);
+                }
             }
 
             return sock;

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/88e1ff88/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiAdapter.java
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiAdapter.java
index 5712d25..9cde198 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiAdapter.java
@@ -725,19 +725,6 @@ abstract class TcpDiscoverySpiAdapter extends IgniteSpiAdapter implements
Discov
     }
 
     /**
-     * @param nodeId Node ID.
-     * @return Class loader for exchange data unmarshalling.
-     */
-    protected ClassLoader exchangeClassLoader(UUID nodeId) {
-        assert ignite != null;
-
-        if (!ignite.configuration().isPeerClassLoadingEnabled() || nodeId.equals(getLocalNodeId()))
-            return null;
-
-        return null;
-    }
-
-    /**
      * @param msg Message.
      * @return Whether delivery of the message is ensured.
      */

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/88e1ff88/modules/core/src/test/java/org/apache/ignite/internal/processors/continuous/GridEventConsumeSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/continuous/GridEventConsumeSelfTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/continuous/GridEventConsumeSelfTest.java
index a51d1a8..4bd2901 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/continuous/GridEventConsumeSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/continuous/GridEventConsumeSelfTest.java
@@ -763,7 +763,7 @@ public class GridEventConsumeSelfTest extends GridCommonAbstractTest implements
      * @throws Exception If failed.
      */
     // TODO: GG-6730
-    public void _testNodeJoinWithP2P() throws Exception {
+    public void testNodeJoinWithP2P() throws Exception {
         final Collection<UUID> nodeIds = new HashSet<>();
         final AtomicInteger cnt = new AtomicInteger();
         final CountDownLatch latch = new CountDownLatch(GRID_CNT + 1);
@@ -800,8 +800,7 @@ public class GridEventConsumeSelfTest extends GridCommonAbstractTest implements
             assertEquals(GRID_CNT + 1, cnt.get());
         }
         finally {
-            stopGrid("anotherGrid1");
-            stopGrid("anotherGrid2");
+            stopGrid("anotherGrid");
 
             grid(0).events().stopRemoteListen(consumeId);
         }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/88e1ff88/modules/core/src/test/java/org/apache/ignite/spi/discovery/AbstractDiscoverySelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/discovery/AbstractDiscoverySelfTest.java
b/modules/core/src/test/java/org/apache/ignite/spi/discovery/AbstractDiscoverySelfTest.java
index 78710fe..91d3514 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/discovery/AbstractDiscoverySelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/discovery/AbstractDiscoverySelfTest.java
@@ -379,14 +379,13 @@ public abstract class AbstractDiscoverySelfTest<T extends IgniteSpi>
extends Gri
                 });
 
                 spi.setDataExchange(new DiscoverySpiDataExchange() {
-                    @Override public Map<Integer, byte[]> collect(UUID nodeId) {
+                    @Override public Map<Integer, Object> collect(UUID nodeId) {
                         return new HashMap<>();
                     }
 
                     @Override public void onExchange(UUID joiningNodeId,
                         UUID nodeId,
-                        Map<Integer, byte[]> data,
-                        ClassLoader clsLdr) {
+                        Map<Integer, Object> data) {
                         // No-op.
                     }
                 });

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/88e1ff88/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiStartStopSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiStartStopSelfTest.java
b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiStartStopSelfTest.java
index 6692046..058d908 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiStartStopSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiStartStopSelfTest.java
@@ -44,14 +44,13 @@ public class TcpDiscoverySpiStartStopSelfTest extends GridSpiStartStopAbstractTe
     @GridSpiTestConfig
     public DiscoverySpiDataExchange getDataExchange() {
         return new DiscoverySpiDataExchange() {
-            @Override public Map<Integer, byte[]> collect(UUID nodeId) {
+            @Override public Map<Integer, Object> collect(UUID nodeId) {
                 return null;
             }
 
             @Override public void onExchange(UUID joiningNodeId,
                 UUID nodeId,
-                Map<Integer, byte[]> data,
-                ClassLoader clsLdr) {
+                Map<Integer, Object> data) {
                 // No-op.
             }
         };

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/88e1ff88/modules/core/src/test/java/org/apache/ignite/testframework/junits/spi/GridSpiAbstractTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testframework/junits/spi/GridSpiAbstractTest.java
b/modules/core/src/test/java/org/apache/ignite/testframework/junits/spi/GridSpiAbstractTest.java
index 7b83979..82f3d6c 100644
--- a/modules/core/src/test/java/org/apache/ignite/testframework/junits/spi/GridSpiAbstractTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/testframework/junits/spi/GridSpiAbstractTest.java
@@ -215,14 +215,13 @@ public abstract class GridSpiAbstractTest<T extends IgniteSpi>
extends GridAbstr
             discoSpi.setMetricsProvider(createMetricsProvider());
 
             discoSpi.setDataExchange(new DiscoverySpiDataExchange() {
-                @Override public Map<Integer, byte[]> collect(UUID nodeId) {
+                @Override public Map<Integer, Object> collect(UUID nodeId) {
                     return new HashMap<>();
                 }
 
                 @Override public void onExchange(UUID joiningNodeId,
                     UUID nodeId,
-                    Map<Integer, byte[]> data,
-                    ClassLoader clsLdr) {
+                    Map<Integer, Object> data) {
                 }
             });
 


Mime
View raw message