You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ag...@apache.org on 2020/04/09 13:43:21 UTC
[ignite] branch master updated: IGNITE-12760 Prevent AssertionError
on message unmarshalling when received from a dead node - Fixes #7538.
This is an automated email from the ASF dual-hosted git repository.
agoncharuk pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new 022b79d IGNITE-12760 Prevent AssertionError on message unmarshalling when received from a dead node - Fixes #7538.
022b79d is described below
commit 022b79dee847c5b292cd1cf72fcac6e99a53addb
Author: denis-chudov <mo...@gmail.com>
AuthorDate: Thu Apr 9 15:47:46 2020 +0300
IGNITE-12760 Prevent AssertionError on message unmarshalling when received from a dead node - Fixes #7538.
Signed-off-by: Alexey Goncharuk <al...@gmail.com>
---
.../ignite/internal/GridEventConsumeHandler.java | 44 +--
.../managers/deployment/GridDeploymentManager.java | 25 +-
.../deployment/GridDeploymentMetadata.java | 4 +-
.../cache/GridCacheDeploymentManager.java | 33 ++-
.../processors/cache/GridCacheIoManager.java | 30 +-
.../IgnitePeerToPeerClassLoadingException.java | 32 ++
.../continuous/CacheContinuousQueryHandler.java | 26 +-
.../p2p/ClassLoadingProblemExceptionTest.java | 330 +++++++++++++++++++++
.../apache/ignite/tests/p2p/P2PTestPredicate.java | 47 +++
.../testsuites/IgniteUriDeploymentTestSuite.java | 4 +-
10 files changed, 516 insertions(+), 59 deletions(-)
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/GridEventConsumeHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/GridEventConsumeHandler.java
index 017a1b8..f6caca5 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/GridEventConsumeHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/GridEventConsumeHandler.java
@@ -345,30 +345,30 @@ class GridEventConsumeHandler implements GridContinuousHandler {
ClassLoader ldr = null;
- if (cache != null) {
- GridCacheDeploymentManager depMgr = cache.context().deploy();
-
- GridDeploymentInfo depInfo = wrapper.depInfo;
-
- if (depInfo != null) {
- depMgr.p2pContext(
- nodeId,
- depInfo.classLoaderId(),
- depInfo.userVersion(),
- depInfo.deployMode(),
- depInfo.participants()
- );
- }
+ try {
+ if (cache != null) {
+ GridCacheDeploymentManager depMgr = cache.context().deploy();
+
+ GridDeploymentInfo depInfo = wrapper.depInfo;
+
+ if (depInfo != null) {
+ depMgr.p2pContext(
+ nodeId,
+ depInfo.classLoaderId(),
+ depInfo.userVersion(),
+ depInfo.deployMode(),
+ depInfo.participants()
+ );
+ }
- ldr = depMgr.globalLoader();
- }
- else {
- U.warn(ctx.log(getClass()), "Received cache event for cache that is not configured locally " +
- "when peer class loading is enabled: " + wrapper.cacheName + ". Will try to unmarshal " +
- "with default class loader.");
- }
+ ldr = depMgr.globalLoader();
+ }
+ else {
+ U.warn(ctx.log(getClass()), "Received cache event for cache that is not configured locally " +
+ "when peer class loading is enabled: " + wrapper.cacheName + ". Will try to unmarshal " +
+ "with default class loader.");
+ }
- try {
wrapper.p2pUnmarshal(ctx.config().getMarshaller(), U.resolveClassLoader(ldr, ctx.config()));
}
catch (IgniteCheckedException e) {
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/deployment/GridDeploymentManager.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/deployment/GridDeploymentManager.java
index b20dd70..f0817df 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/deployment/GridDeploymentManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/deployment/GridDeploymentManager.java
@@ -287,13 +287,13 @@ public class GridDeploymentManager extends GridManagerAdapter<DeploymentSpi> {
// Check for nested execution. In that case, if task
// is available locally by name, then we should ignore
// class loader ID.
- GridDeployment dep = locStore.getDeployment(meta);
+ GridDeployment dep = checkDeployment(locStore.getDeployment(meta), "local");
if (dep == null) {
- dep = ldrStore.getDeployment(ldr.classLoaderId());
+ dep = checkDeployment(ldrStore.getDeployment(ldr.classLoaderId()), "perLoader");
if (dep == null)
- dep = verStore.getDeployment(ldr.classLoaderId());
+ dep = checkDeployment(verStore.getDeployment(ldr.classLoaderId()), "perVersion");
}
return dep;
@@ -313,6 +313,25 @@ public class GridDeploymentManager extends GridManagerAdapter<DeploymentSpi> {
}
/**
+ * Checks and logs possibly incorrect deployments.
+ *
+ * @param deployment Deployment.
+ * @param store Store name.
+ * @return Deployment.
+ */
+ private GridDeployment checkDeployment(GridDeployment deployment, String store) {
+ if (deployment != null
+ && deployment.participants() == null
+ && !ctx.discovery().localNode().id().equals(deployment.classLoaderId().globalId())) {
+ log.warning("Possibly incorrect deployment detected in '" + store + "', can't be used to deserialize " +
+ "message on target node [participants=null, localNodeId=" + ctx.discovery().localNode().id() +
+ ", deployment=" + deployment + "]");
+ }
+
+ return deployment;
+ }
+
+ /**
* Gets any deployment by loader ID.
*
* @param ldrId Loader ID.
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/deployment/GridDeploymentMetadata.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/deployment/GridDeploymentMetadata.java
index 67cd213..370825c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/deployment/GridDeploymentMetadata.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/deployment/GridDeploymentMetadata.java
@@ -29,7 +29,7 @@ import org.apache.ignite.lang.IgniteUuid;
/**
* Deployment metadata.
*/
-class GridDeploymentMetadata {
+public class GridDeploymentMetadata {
/** Deployment mode. */
private DeploymentMode depMode;
@@ -203,7 +203,7 @@ class GridDeploymentMetadata {
*
* @param clsLdrId Property clsLdrId.
*/
- void classLoaderId(IgniteUuid clsLdrId) {
+ public void classLoaderId(IgniteUuid clsLdrId) {
this.clsLdrId = clsLdrId;
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheDeploymentManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheDeploymentManager.java
index c104909..707fea1 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheDeploymentManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheDeploymentManager.java
@@ -375,7 +375,7 @@ public class GridCacheDeploymentManager<K, V> extends GridCacheSharedManagerAdap
String userVer,
DeploymentMode mode,
Map<UUID, IgniteUuid> participants
- ) {
+ ) throws IgnitePeerToPeerClassLoadingException {
localLdrId.set(ldrId);
assert depEnabled;
@@ -427,6 +427,12 @@ public class GridCacheDeploymentManager<K, V> extends GridCacheSharedManagerAdap
depInfo = deps.get(ldrId);
if (depInfo == null) {
+ if (!sndId.equals(ldrId.globalId()) && participants == null) {
+ throw new IgnitePeerToPeerClassLoadingException("Failed to load class using class loader with " +
+ "given id (loader id doesn't match sender id and there are no more participants) " +
+ "[clsLdrId=" + ldrId + ", senderId=" + sndId + ", participants=null]");
+ }
+
depInfo = new CachedDeploymentInfo<>(sndId, ldrId, userVer, mode, participants);
CachedDeploymentInfo<K, V> old = deps.putIfAbsent(ldrId, depInfo);
@@ -668,7 +674,7 @@ public class GridCacheDeploymentManager<K, V> extends GridCacheSharedManagerAdap
*
* @param deployable Deployable object.
*/
- public void prepare(GridCacheDeployable deployable) {
+ public void prepare(GridCacheDeployable deployable) throws IgnitePeerToPeerClassLoadingException {
assert depEnabled;
// Only set deployment info if it was not set automatically.
@@ -684,8 +690,11 @@ public class GridCacheDeploymentManager<K, V> extends GridCacheSharedManagerAdap
}
}
- if (dep != null)
+ if (dep != null) {
+ checkDeploymentIsCorrect(dep, deployable);
+
deployable.prepare(dep);
+ }
if (log.isDebugEnabled())
log.debug("Prepared grid cache deployable [dep=" + dep + ", deployable=" + deployable + ']');
@@ -693,6 +702,24 @@ public class GridCacheDeploymentManager<K, V> extends GridCacheSharedManagerAdap
}
/**
+ * Checks if given deployment is correct to prepare a message.
+ *
+ * @param deployment Deployment.
+ * @param deployable Deployable message.
+ * @throws IgnitePeerToPeerClassLoadingException If deployment is incorrect.
+ */
+ private void checkDeploymentIsCorrect(GridDeploymentInfoBean deployment, GridCacheDeployable deployable)
+ throws IgnitePeerToPeerClassLoadingException {
+ if (deployment.participants() == null
+ && !cctx.localNode().id().equals(deployment.classLoaderId().globalId())) {
+ throw new IgnitePeerToPeerClassLoadingException("Failed to use deployment to prepare deployable, " +
+ "because local node id does not correspond with class loader id, and there are no more participants " +
+ "[localNodeId=" + cctx.localNode().id() + ", deployment=" + deployment + ", deployable=" + deployable +
+ ", locDep=" + locDep.get() + "]");
+ }
+ }
+
+ /**
* @return First global deployment.
*/
@Nullable public GridDeploymentInfoBean globalDeploymentInfo() {
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java
index 7170ac3..f2eab9e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java
@@ -1605,23 +1605,23 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
GridDeploymentInfo bean = cacheMsg.deployInfo();
- if (bean != null) {
- assert depEnabled : "Received deployment info while peer class loading is disabled [nodeId=" + nodeId +
- ", msg=" + cacheMsg + ']';
-
- cctx.deploy().p2pContext(
- nodeId,
- bean.classLoaderId(),
- bean.userVersion(),
- bean.deployMode(),
- bean.participants()
- );
+ try {
+ if (bean != null) {
+ assert depEnabled : "Received deployment info while peer class loading is disabled [nodeId=" + nodeId +
+ ", msg=" + cacheMsg + ']';
- if (log.isDebugEnabled())
- log.debug("Set P2P context [senderId=" + nodeId + ", msg=" + cacheMsg + ']');
- }
+ cctx.deploy().p2pContext(
+ nodeId,
+ bean.classLoaderId(),
+ bean.userVersion(),
+ bean.deployMode(),
+ bean.participants()
+ );
+
+ if (log.isDebugEnabled())
+ log.debug("Set P2P context [senderId=" + nodeId + ", msg=" + cacheMsg + ']');
+ }
- try {
cacheMsg.finishUnmarshal(cctx, cctx.deploy().globalLoader());
}
catch (IgniteCheckedException e) {
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgnitePeerToPeerClassLoadingException.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgnitePeerToPeerClassLoadingException.java
new file mode 100644
index 0000000..9a6158b
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgnitePeerToPeerClassLoadingException.java
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.ignite.internal.processors.cache;
+
+import org.apache.ignite.IgniteCheckedException;
+
+/**
+ * This exception is thrown in communication-related cases when it is impossible to load class from any peer.
+ */
+public class IgnitePeerToPeerClassLoadingException extends IgniteCheckedException {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** */
+ public IgnitePeerToPeerClassLoadingException(String s) {
+ super(s);
+ }
+}
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 57e53d2..faec73f 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
@@ -907,21 +907,21 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
ClassLoader ldr = depMgr.globalLoader();
- if (ctx.config().isPeerClassLoadingEnabled()) {
- GridDeploymentInfo depInfo = e.deployInfo();
-
- if (depInfo != null) {
- depMgr.p2pContext(
- nodeId,
- depInfo.classLoaderId(),
- depInfo.userVersion(),
- depInfo.deployMode(),
- depInfo.participants()
- );
+ try {
+ if (ctx.config().isPeerClassLoadingEnabled()) {
+ GridDeploymentInfo depInfo = e.deployInfo();
+
+ if (depInfo != null) {
+ depMgr.p2pContext(
+ nodeId,
+ depInfo.classLoaderId(),
+ depInfo.userVersion(),
+ depInfo.deployMode(),
+ depInfo.participants()
+ );
+ }
}
- }
- try {
e.unmarshal(cctx, ldr);
Collection<CacheEntryEvent<? extends K, ? extends V>> evts = handleEvent(ctx, e);
diff --git a/modules/core/src/test/java/org/apache/ignite/p2p/ClassLoadingProblemExceptionTest.java b/modules/core/src/test/java/org/apache/ignite/p2p/ClassLoadingProblemExceptionTest.java
new file mode 100644
index 0000000..f770f5b
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/p2p/ClassLoadingProblemExceptionTest.java
@@ -0,0 +1,330 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.ignite.p2p;
+
+import java.io.Serializable;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.IntFunction;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.cache.query.ScanQuery;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.compute.ComputeJob;
+import org.apache.ignite.compute.ComputeJobAdapter;
+import org.apache.ignite.compute.ComputeJobResult;
+import org.apache.ignite.compute.ComputeTaskSplitAdapter;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.managers.communication.GridIoMessage;
+import org.apache.ignite.internal.managers.deployment.GridDeployment;
+import org.apache.ignite.internal.managers.deployment.GridDeploymentInfoBean;
+import org.apache.ignite.internal.managers.deployment.GridDeploymentManager;
+import org.apache.ignite.internal.managers.deployment.GridDeploymentMetadata;
+import org.apache.ignite.internal.managers.deployment.GridDeploymentStore;
+import org.apache.ignite.internal.managers.deployment.GridTestDeployment;
+import org.apache.ignite.internal.marshaller.optimized.OptimizedMarshaller;
+import org.apache.ignite.internal.processors.cache.IgnitePeerToPeerClassLoadingException;
+import org.apache.ignite.internal.processors.cache.query.GridCacheQueryRequest;
+import org.apache.ignite.internal.util.typedef.X;
+import org.apache.ignite.lang.IgniteBiPredicate;
+import org.apache.ignite.lang.IgniteInClosure;
+import org.apache.ignite.lang.IgniteUuid;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.resources.IgniteInstanceResource;
+import org.apache.ignite.spi.IgniteSpiException;
+import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.jetbrains.annotations.Nullable;
+import org.junit.Test;
+
+import static org.apache.ignite.configuration.DeploymentMode.SHARED;
+
+/**
+ *
+ */
+public class ClassLoadingProblemExceptionTest extends GridCommonAbstractTest implements Serializable {
+ /** Test predicate class name. */
+ private static final String PREDICATE_NAME = "org.apache.ignite.tests.p2p.P2PTestPredicate";
+
+ /** Flag that allows to test the check on receiving message. */
+ private boolean spoilMsgOnSnd = false;
+
+ /** Flag that allows to test the check while preparing message on sender. */
+ private boolean spoilDeploymentBeforePrepare = false;
+
+ /** */
+ private static final String CACHE_NAME = "cache";
+
+ /** */
+ private static final String CLIENT_PREFIX = "client";
+
+ /** */
+ private static final String CLIENT_1 = CLIENT_PREFIX + "1";
+
+ /** */
+ private static final String CLIENT_2 = CLIENT_PREFIX + "2";
+
+ /** Container for exception that can be thrown inside of job. */
+ private static AtomicReference<Throwable> exceptionThrown = new AtomicReference<>(null);
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ cfg
+ .setClientMode(igniteInstanceName.startsWith(CLIENT_PREFIX))
+ .setPeerClassLoadingEnabled(true)
+ .setDeploymentMode(SHARED)
+ .setCommunicationSpi(new TestCommunicationSpi())
+ .setMarshaller(new OptimizedMarshaller());
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ super.beforeTest();
+
+ stopAllGrids();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ stopAllGrids();
+
+ super.afterTest();
+ }
+
+ /**
+ *
+ * @param spoilDeploymentBeforePrepare Test the check on preparing message.
+ * @param spoilMsgOnSnd Test the check on receiving message.
+ * @param eExp Exception expected.
+ * @throws Exception If failed.
+ */
+ private void doTest(
+ boolean spoilDeploymentBeforePrepare,
+ boolean spoilMsgOnSnd,
+ Class<? extends Throwable> eExp
+ ) throws Exception {
+ this.spoilDeploymentBeforePrepare = spoilDeploymentBeforePrepare;
+ this.spoilMsgOnSnd = spoilMsgOnSnd;
+ this.exceptionThrown.set(null);
+
+ IgniteEx crd = (IgniteEx)startGridsMultiThreaded(1);
+
+ crd.getOrCreateCache(CACHE_NAME);
+
+ IgniteEx client1 = startGrid(CLIENT_1);
+ IgniteEx client2 = startGrid(CLIENT_2);
+
+ awaitPartitionMapExchange();
+
+ GridDeploymentManager deploymentMgr = client2.context().deploy();
+
+ GridDeploymentStore store = GridTestUtils.getFieldValue(deploymentMgr, "locStore");
+
+ GridTestUtils.setFieldValue(deploymentMgr, "locStore", new GridDeploymentTestStore(store));
+
+ client1.compute(client1.cluster().forRemotes()).execute(new P2PDeploymentLongRunningTask(), null);
+
+ if (exceptionThrown.get() != null)
+ exceptionThrown.get().printStackTrace();
+
+ assertTrue(
+ "Wrong exception: " + exceptionThrown.get(),
+ exceptionThrown.get() == null && eExp == null || X.hasCause(exceptionThrown.get(), eExp)
+ );
+ }
+
+ /** */
+ @Test
+ public void testDontSpoil() throws Exception {
+ doTest(false, false, null);
+ }
+
+ /** */
+ @Test
+ public void testSpoilBeforePrepare() throws Exception {
+ doTest(true, false, IgnitePeerToPeerClassLoadingException.class);
+ }
+
+ /** */
+ @Test
+ public void testSpoilOnSend() throws Exception {
+ doTest(false, true, IgnitePeerToPeerClassLoadingException.class);
+ }
+
+ /** */
+ @Test
+ public void testSpoilBoth() throws Exception {
+ doTest(true, true, IgnitePeerToPeerClassLoadingException.class);
+ }
+
+ /**
+ *
+ */
+ private class TestCommunicationSpi extends TcpCommunicationSpi {
+ /** {@inheritDoc} */
+ @Override public void sendMessage(ClusterNode node, Message msg, IgniteInClosure<IgniteException> ackC)
+ throws IgniteSpiException {
+ if (spoilMsgOnSnd && msg instanceof GridIoMessage) {
+ GridIoMessage ioMsg = (GridIoMessage)msg;
+
+ Message m = ioMsg.message();
+
+ if (m instanceof GridCacheQueryRequest) {
+ GridCacheQueryRequest queryRequest = (GridCacheQueryRequest)m;
+
+ if (queryRequest.deployInfo() != null) {
+ queryRequest.prepare(
+ new GridDeploymentInfoBean(
+ IgniteUuid.fromUuid(UUID.randomUUID()),
+ queryRequest.deployInfo().userVersion(),
+ queryRequest.deployInfo().deployMode(),
+ null
+ )
+ );
+ }
+ }
+ }
+
+ super.sendMessage(node, msg, ackC);
+ }
+ }
+
+ /**
+ *
+ */
+ private class GridDeploymentTestStore implements GridDeploymentStore {
+ /** */
+ private final GridDeploymentStore store;
+
+ /** */
+ public GridDeploymentTestStore(GridDeploymentStore store) {
+ this.store = store;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void start() throws IgniteCheckedException {
+ store.start();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void stop() {
+ store.stop();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onKernalStart() throws IgniteCheckedException {
+ store.onKernalStart();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onKernalStop() {
+ store.onKernalStop();
+ }
+
+ /** {@inheritDoc} */
+ @Override public @Nullable GridDeployment getDeployment(GridDeploymentMetadata meta) {
+ return store.getDeployment(meta);
+ }
+
+ /** {@inheritDoc} */
+ @Override public @Nullable GridDeployment searchDeploymentCache(GridDeploymentMetadata meta) {
+ return store.searchDeploymentCache(meta);
+ }
+
+ /** {@inheritDoc} */
+ @Override public @Nullable GridDeployment getDeployment(IgniteUuid ldrId) {
+ return store.getDeployment(ldrId);
+ }
+
+ /** {@inheritDoc} */
+ @Override public Collection<GridDeployment> getDeployments() {
+ return store.getDeployments();
+ }
+
+ /** {@inheritDoc} */
+ @Override public GridDeployment explicitDeploy(Class<?> cls, ClassLoader clsLdr) throws IgniteCheckedException {
+ if (spoilDeploymentBeforePrepare) {
+ IgniteUuid ldrId = IgniteUuid.fromUuid(UUID.randomUUID());
+
+ return new GridTestDeployment(SHARED, getClass().getClassLoader(), ldrId, "0", PREDICATE_NAME, false);
+ }
+ else
+ return store.explicitDeploy(cls, clsLdr);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void explicitUndeploy(@Nullable UUID nodeId, String rsrcName) {
+ store.explicitUndeploy(nodeId, rsrcName);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void addParticipants(Map<UUID, IgniteUuid> allParticipants, Map<UUID, IgniteUuid> addedParticipants) {
+ store.addParticipants(allParticipants, addedParticipants);
+ }
+ }
+
+ /**
+ *
+ */
+ public static class P2PDeploymentLongRunningTask extends ComputeTaskSplitAdapter<Object, Object> {
+ /** {@inheritDoc} */
+ @Override protected Collection<? extends ComputeJob> split(int gridSize, Object arg) throws IgniteException {
+ IntFunction<ComputeJobAdapter> f = i -> new ComputeJobAdapter() {
+ /** */
+ @IgniteInstanceResource
+ private transient IgniteEx ignite;
+
+ /** {@inheritDoc} */
+ @Override public Object execute() {
+ try {
+ if (ignite.configuration().getIgniteInstanceName().equals(CLIENT_2)) {
+ IgniteCache<Integer, Integer> cache = ignite.getOrCreateCache(CACHE_NAME);
+
+ Class<IgniteBiPredicate> cls = (Class<IgniteBiPredicate>)getExternalClassLoader().loadClass(PREDICATE_NAME);
+
+ cache.query(new ScanQuery<IgniteBiPredicate, Integer>(cls.newInstance())).getAll();
+ }
+ }
+ catch (Throwable e) {
+ exceptionThrown.set(e);
+ }
+
+ return null;
+ }
+ };
+
+ return IntStream.range(0, gridSize).mapToObj(f).collect(Collectors.toList());
+ }
+
+ /** {@inheritDoc} */
+ @Nullable @Override public Object reduce(List<ComputeJobResult> results) throws IgniteException {
+ return null;
+ }
+ }
+}
diff --git a/modules/extdata/p2p/src/main/java/org/apache/ignite/tests/p2p/P2PTestPredicate.java b/modules/extdata/p2p/src/main/java/org/apache/ignite/tests/p2p/P2PTestPredicate.java
new file mode 100644
index 0000000..1bd6ea8
--- /dev/null
+++ b/modules/extdata/p2p/src/main/java/org/apache/ignite/tests/p2p/P2PTestPredicate.java
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.tests.p2p;
+
+import java.io.Serializable;
+import org.apache.ignite.internal.processors.cache.GridCacheDeployable;
+import org.apache.ignite.internal.processors.cache.GridCacheIdMessage;
+import org.apache.ignite.lang.IgniteBiPredicate;
+
+/**
+ * Test predicate for scan queries in p2p deployment tests.
+ */
+@SuppressWarnings("unused")
+public class P2PTestPredicate extends GridCacheIdMessage implements GridCacheDeployable, IgniteBiPredicate, Serializable {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** {@inheritDoc} */
+ @Override public boolean addDeploymentInfo() {
+ return true;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean apply(Object o, Object o2) {
+ return false;
+ }
+
+ /** {@inheritDoc} */
+ @Override public short directType() {
+ return 0;
+ }
+}
diff --git a/modules/urideploy/src/test/java/org/apache/ignite/testsuites/IgniteUriDeploymentTestSuite.java b/modules/urideploy/src/test/java/org/apache/ignite/testsuites/IgniteUriDeploymentTestSuite.java
index 1741179..3f4216a 100644
--- a/modules/urideploy/src/test/java/org/apache/ignite/testsuites/IgniteUriDeploymentTestSuite.java
+++ b/modules/urideploy/src/test/java/org/apache/ignite/testsuites/IgniteUriDeploymentTestSuite.java
@@ -18,6 +18,7 @@
package org.apache.ignite.testsuites;
import org.apache.ignite.internal.GridTaskUriDeploymentDeadlockSelfTest;
+import org.apache.ignite.p2p.ClassLoadingProblemExceptionTest;
import org.apache.ignite.p2p.GridP2PDisabledSelfTest;
import org.apache.ignite.spi.deployment.uri.GridUriDeploymentClassLoaderMultiThreadedSelfTest;
import org.apache.ignite.spi.deployment.uri.GridUriDeploymentClassLoaderSelfTest;
@@ -59,7 +60,8 @@ import org.junit.runners.Suite;
IgniteToolsSelfTestSuite.class,
GridTaskUriDeploymentDeadlockSelfTest.class,
- GridP2PDisabledSelfTest.class
+ GridP2PDisabledSelfTest.class,
+ ClassLoadingProblemExceptionTest.class
})
public class IgniteUriDeploymentTestSuite {
}