http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/main/java/org/apache/ignite/internal/processors/service/ServiceSingleNodeDeploymentResultBatch.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/ServiceSingleNodeDeploymentResultBatch.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/ServiceSingleNodeDeploymentResultBatch.java new file mode 100644 index 0000000..320c5a9 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/ServiceSingleNodeDeploymentResultBatch.java @@ -0,0 +1,155 @@ +/* + * 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.service; + +import java.nio.ByteBuffer; +import java.util.Map; +import org.apache.ignite.internal.util.tostring.GridToStringInclude; +import org.apache.ignite.internal.util.typedef.internal.S; +import org.apache.ignite.lang.IgniteUuid; +import org.apache.ignite.plugin.extensions.communication.Message; +import org.apache.ignite.plugin.extensions.communication.MessageReader; +import org.apache.ignite.plugin.extensions.communication.MessageWriter; +import org.jetbrains.annotations.NotNull; + +import static org.apache.ignite.plugin.extensions.communication.MessageCollectionItemType.IGNITE_UUID; +import static org.apache.ignite.plugin.extensions.communication.MessageCollectionItemType.MSG; + +/** + * Batch of service single node deployment result. + * <p/> + * Contains collection of {@link ServiceSingleNodeDeploymentResult} mapped services ids. + */ +public class ServiceSingleNodeDeploymentResultBatch implements Message { + /** */ + private static final long serialVersionUID = 0L; + + /** Deployment process id. */ + @GridToStringInclude + private ServiceDeploymentProcessId depId; + + /** Services deployments results. */ + @GridToStringInclude + private Map<IgniteUuid, ServiceSingleNodeDeploymentResult> results; + + /** + * Empty constructor for marshalling purposes. + */ + public ServiceSingleNodeDeploymentResultBatch() { + } + + /** + * @param depId Deployment process id. + * @param results Services deployments results. + */ + public ServiceSingleNodeDeploymentResultBatch(@NotNull ServiceDeploymentProcessId depId, + @NotNull Map<IgniteUuid, ServiceSingleNodeDeploymentResult> results) { + this.depId = depId; + this.results = results; + } + + /** + * @return Services deployments results. + */ + public Map<IgniteUuid, ServiceSingleNodeDeploymentResult> results() { + return results; + } + + /** + * @return Deployment process id. + */ + public ServiceDeploymentProcessId deploymentId() { + return depId; + } + + /** {@inheritDoc} */ + @Override public boolean writeTo(ByteBuffer buf, MessageWriter writer) { + writer.setBuffer(buf); + + if (!writer.isHeaderWritten()) { + if (!writer.writeHeader(directType(), fieldsCount())) + return false; + + writer.onHeaderWritten(); + } + + switch (writer.state()) { + case 0: + if (!writer.writeMessage("depId", depId)) + return false; + + writer.incrementState(); + + case 1: + if (!writer.writeMap("results", results, IGNITE_UUID, MSG)) + return false; + + writer.incrementState(); + } + + return true; + } + + /** {@inheritDoc} */ + @Override public boolean readFrom(ByteBuffer buf, MessageReader reader) { + reader.setBuffer(buf); + + if (!reader.beforeMessageRead()) + return false; + + switch (reader.state()) { + case 0: + depId = reader.readMessage("depId"); + + if (!reader.isLastRead()) + return false; + + reader.incrementState(); + + case 1: + results = reader.readMap("results", IGNITE_UUID, MSG, false); + + if (!reader.isLastRead()) + return false; + + reader.incrementState(); + } + + return reader.afterMessageRead(ServiceSingleNodeDeploymentResultBatch.class); + } + + /** {@inheritDoc} */ + @Override public short directType() { + return 168; + } + + /** {@inheritDoc} */ + @Override public byte fieldsCount() { + return 2; + } + + /** {@inheritDoc} */ + @Override public void onAckReceived() { + // No-op. + } + + /** {@inheritDoc} */ + @Override public String toString() { + return S.toString(ServiceSingleNodeDeploymentResultBatch.class, this); + } +}
http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/main/java/org/apache/ignite/internal/processors/service/ServiceUndeploymentRequest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/ServiceUndeploymentRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/ServiceUndeploymentRequest.java new file mode 100644 index 0000000..5482362 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/ServiceUndeploymentRequest.java @@ -0,0 +1,42 @@ +/* + * 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.service; + +import org.apache.ignite.internal.util.typedef.internal.S; +import org.apache.ignite.lang.IgniteUuid; +import org.jetbrains.annotations.NotNull; + +/** + * Service undeployment request. + */ +public class ServiceUndeploymentRequest extends ServiceChangeAbstractRequest { + /** */ + private static final long serialVersionUID = 0L; + + /** + * @param srvcId Service id. + */ + public ServiceUndeploymentRequest(@NotNull IgniteUuid srvcId) { + super(srvcId); + } + + /** {@inheritDoc} */ + @Override public String toString() { + return S.toString(ServiceUndeploymentRequest.class, this); + } +} http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/main/java/org/apache/ignite/services/ServiceConfiguration.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/services/ServiceConfiguration.java b/modules/core/src/main/java/org/apache/ignite/services/ServiceConfiguration.java index 37a794b..a517197 100644 --- a/modules/core/src/main/java/org/apache/ignite/services/ServiceConfiguration.java +++ b/modules/core/src/main/java/org/apache/ignite/services/ServiceConfiguration.java @@ -181,6 +181,8 @@ public class ServiceConfiguration implements Serializable { * Gets cache name used for key-to-node affinity calculation. * <p> * This parameter is optional and is set only when deploying service based on key-affinity. + * <p/> + * <b>NOTE:</b> If the cache is destroyed, the service will be undeployed automatically. * * @return Cache name, possibly {@code null}. */ http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java index e61d30a..0edb497 100644 --- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java +++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java @@ -150,6 +150,7 @@ import org.jetbrains.annotations.Nullable; import static org.apache.ignite.IgniteSystemProperties.IGNITE_BINARY_MARSHALLER_USE_STRING_SERIALIZATION_VER_2; import static org.apache.ignite.IgniteSystemProperties.IGNITE_DISCOVERY_CLIENT_RECONNECT_HISTORY_SIZE; import static org.apache.ignite.IgniteSystemProperties.IGNITE_OPTIMIZED_MARSHALLER_USE_DEFAULT_SUID; +import static org.apache.ignite.IgniteSystemProperties.IGNITE_EVENT_DRIVEN_SERVICE_PROCESSOR_ENABLED; import static org.apache.ignite.IgniteSystemProperties.getInteger; import static org.apache.ignite.events.EventType.EVT_NODE_FAILED; import static org.apache.ignite.events.EventType.EVT_NODE_JOINED; @@ -163,6 +164,7 @@ import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_MARSHALLER; import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_MARSHALLER_COMPACT_FOOTER; import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_MARSHALLER_USE_BINARY_STRING_SER_VER_2; import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_MARSHALLER_USE_DFLT_SUID; +import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_EVENT_DRIVEN_SERVICE_PROCESSOR_ENABLED; import static org.apache.ignite.spi.IgnitePortProtocol.TCP; import static org.apache.ignite.spi.discovery.tcp.internal.TcpDiscoverySpiState.AUTH_FAILED; import static org.apache.ignite.spi.discovery.tcp.internal.TcpDiscoverySpiState.CHECK_FAILED; @@ -4044,6 +4046,48 @@ class ServerImpl extends TcpDiscoveryImpl { return; } + final Boolean locSrvcProcModeAttr = locNode.attribute(ATTR_EVENT_DRIVEN_SERVICE_PROCESSOR_ENABLED); + // Can be null only in module tests of discovery spi (without node startup). + final Boolean locSrvcProcMode = locSrvcProcModeAttr != null ? locSrvcProcModeAttr : false; + + final Boolean rmtSrvcProcModeAttr = node.attribute(ATTR_EVENT_DRIVEN_SERVICE_PROCESSOR_ENABLED); + final boolean rmtSrvcProcMode = rmtSrvcProcModeAttr != null ? rmtSrvcProcModeAttr : false; + + if (!F.eq(locSrvcProcMode, rmtSrvcProcMode)) { + utilityPool.execute( + new Runnable() { + @Override public void run() { + String errMsg = "Local node's " + IGNITE_EVENT_DRIVEN_SERVICE_PROCESSOR_ENABLED + + " property value differs from remote node's value " + + "(to make sure all nodes in topology have identical service processor mode, " + + "configure system property explicitly) " + + "[locSrvcProcMode=" + locSrvcProcMode + + ", rmtSrvcProcMode=" + rmtSrvcProcMode + + ", locNodeAddrs=" + U.addressesAsString(locNode) + + ", rmtNodeAddrs=" + U.addressesAsString(node) + + ", locNodeId=" + locNode.id() + ", rmtNodeId=" + msg.creatorNodeId() + ']'; + + String sndMsg = "Local node's " + IGNITE_EVENT_DRIVEN_SERVICE_PROCESSOR_ENABLED + + " property value differs from remote node's value " + + "(to make sure all nodes in topology have identical service processor mode, " + + "configure system property explicitly) " + + "[locSrvcProcMode=" + rmtSrvcProcMode + + ", rmtSrvcProcMode=" + locSrvcProcMode + + ", locNodeAddrs=" + U.addressesAsString(node) + ", locPort=" + node.discoveryPort() + + ", rmtNodeAddr=" + U.addressesAsString(locNode) + ", locNodeId=" + node.id() + + ", rmtNodeId=" + locNode.id() + ']'; + + nodeCheckError( + node, + errMsg, + sndMsg); + } + }); + + // Ignore join request. + return; + } + // Handle join. node.internalOrder(ring.nextNodeOrder()); http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/test/java/org/apache/ignite/internal/GridDeploymentSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/GridDeploymentSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/GridDeploymentSelfTest.java index 000581a..2c41c8e 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/GridDeploymentSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/GridDeploymentSelfTest.java @@ -480,7 +480,7 @@ public class GridDeploymentSelfTest extends GridCommonAbstractTest { /** {@inheritDoc} */ @Override public boolean register(ClassLoader ldr, Class rsrc) throws IgniteSpiException { - if (super.register(ldr, rsrc)) { + if (super.register(ldr, rsrc) && ComputeTaskAdapter.class.isAssignableFrom(rsrc)) { deployCnt++; return true; http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectServicesTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectServicesTest.java b/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectServicesTest.java index 98803e0..a42f960 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectServicesTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectServicesTest.java @@ -30,6 +30,7 @@ import org.apache.ignite.resources.IgniteInstanceResource; import org.apache.ignite.services.Service; import org.apache.ignite.services.ServiceContext; import org.apache.ignite.testframework.GridTestUtils; +import org.junit.Assume; import org.junit.Test; import org.junit.runner.RunWith; import org.junit.runners.JUnit4; @@ -133,6 +134,8 @@ public class IgniteClientReconnectServicesTest extends IgniteClientReconnectAbst */ @Test public void testReconnectInDeploying() throws Exception { + Assume.assumeTrue(!isEventDrivenServiceProcessorEnabled()); + Ignite client = grid(serverCount()); assertTrue(client.cluster().localNode().isClient()); @@ -179,6 +182,52 @@ public class IgniteClientReconnectServicesTest extends IgniteClientReconnectAbst /** * @throws Exception If failed. */ + @SuppressWarnings("ThrowableResultOfMethodCallIgnored") + @Test + public void testReconnectInDeployingNew() throws Exception { + Assume.assumeTrue(isEventDrivenServiceProcessorEnabled()); + + IgniteEx client = grid(serverCount()); + + assertTrue(client.cluster().localNode().isClient()); + + final IgniteServices services = client.services(); + + Ignite srv = ignite(0); + + final IgniteInternalFuture<Object> fut = GridTestUtils.runAsync(new Callable<Object>() { + @Override public Object call() throws Exception { + try { + services.deployClusterSingleton("testReconnectInDeploying", new TestServiceImpl()); + } + catch (IgniteClientDisconnectedException e) { + checkAndWait(e); + + return true; + } + + return false; + } + }); + + reconnectClientNode(client, srv, () -> { + // Check that client waiting operation. + GridTestUtils.assertThrows(log, () -> fut.get(200), IgniteFutureTimeoutCheckedException.class, null); + + try { + assertNotDone(fut); + } + catch (Exception e) { + fail("Unexpected exception has been thrown, err=" + e.getMessage()); + } + }); + + assertTrue((Boolean)fut.get(2, TimeUnit.SECONDS)); + } + + /** + * @throws Exception If failed. + */ @Test public void testReconnectInProgress() throws Exception { Ignite client = grid(serverCount()); http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/test/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManagerAttributesSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManagerAttributesSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManagerAttributesSelfTest.java index 8db1551..cc88a6a 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManagerAttributesSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManagerAttributesSelfTest.java @@ -35,6 +35,7 @@ import org.junit.runners.JUnit4; import static org.apache.ignite.IgniteSystemProperties.IGNITE_BINARY_MARSHALLER_USE_STRING_SERIALIZATION_VER_2; import static org.apache.ignite.IgniteSystemProperties.IGNITE_OPTIMIZED_MARSHALLER_USE_DEFAULT_SUID; import static org.apache.ignite.IgniteSystemProperties.IGNITE_SECURITY_COMPATIBILITY_MODE; +import static org.apache.ignite.IgniteSystemProperties.IGNITE_EVENT_DRIVEN_SERVICE_PROCESSOR_ENABLED; import static org.apache.ignite.configuration.DeploymentMode.CONTINUOUS; import static org.apache.ignite.configuration.DeploymentMode.SHARED; @@ -246,6 +247,17 @@ public abstract class GridDiscoveryManagerAttributesSelfTest extends GridCommonA * @throws Exception If failed. */ @Test + public void testServiceProcessorModeProperty() throws Exception { + doTestCompatibilityEnabled(IGNITE_EVENT_DRIVEN_SERVICE_PROCESSOR_ENABLED, true, false, true); + doTestCompatibilityEnabled(IGNITE_EVENT_DRIVEN_SERVICE_PROCESSOR_ENABLED, false, true, true); + doTestCompatibilityEnabled(IGNITE_EVENT_DRIVEN_SERVICE_PROCESSOR_ENABLED, true, true, false); + doTestCompatibilityEnabled(IGNITE_EVENT_DRIVEN_SERVICE_PROCESSOR_ENABLED, false, false, false); + } + + /** + * @throws Exception If failed. + */ + @Test public void testSecurityCompatibilityEnabled() throws Exception { TestReconnectPluginProvider.enabled = true; TestReconnectProcessor.enabled = true; http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/GridCacheContinuousQueryAbstractSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/GridCacheContinuousQueryAbstractSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/GridCacheContinuousQueryAbstractSelfTest.java index 7e66584..b26e4d4 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/GridCacheContinuousQueryAbstractSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/GridCacheContinuousQueryAbstractSelfTest.java @@ -58,8 +58,10 @@ import org.apache.ignite.configuration.NearCacheConfiguration; import org.apache.ignite.events.CacheQueryExecutedEvent; import org.apache.ignite.events.CacheQueryReadEvent; import org.apache.ignite.events.Event; +import org.apache.ignite.internal.GridKernalContext; import org.apache.ignite.internal.processors.continuous.GridContinuousProcessor; import org.apache.ignite.internal.processors.datastructures.GridCacheInternalKeyImpl; +import org.apache.ignite.internal.processors.service.GridServiceProcessor; import org.apache.ignite.internal.util.typedef.F; import org.apache.ignite.internal.util.typedef.P2; import org.apache.ignite.internal.util.typedef.PA; @@ -206,9 +208,12 @@ public abstract class GridCacheContinuousQueryAbstractSelfTest extends GridCommo }, 3000); for (int i = 0; i < gridCount(); i++) { - GridContinuousProcessor proc = grid(i).context().continuous(); + GridKernalContext ctx = grid(i).context(); + GridContinuousProcessor proc = ctx.continuous(); - assertEquals(String.valueOf(i), 1, ((Map)U.field(proc, "locInfos")).size()); + final int locInfosCnt = ctx.service() instanceof GridServiceProcessor ? 1 : 0; + + assertEquals(String.valueOf(i), locInfosCnt, ((Map)U.field(proc, "locInfos")).size()); assertEquals(String.valueOf(i), 0, ((Map)U.field(proc, "rmtInfos")).size()); assertEquals(String.valueOf(i), 0, ((Map)U.field(proc, "startFuts")).size()); assertEquals(String.valueOf(i), 0, ((Map)U.field(proc, "stopFuts")).size()); http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceContinuousQueryRedeployTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceContinuousQueryRedeployTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceContinuousQueryRedeployTest.java index e4157e2..47bdf61 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceContinuousQueryRedeployTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceContinuousQueryRedeployTest.java @@ -132,11 +132,7 @@ public class GridServiceContinuousQueryRedeployTest extends GridCommonAbstractTe svcCfg.setName(SERVICE_NAME); svcCfg.setTotalCount(1); svcCfg.setMaxPerNodeCount(1); - svcCfg.setNodeFilter(new IgnitePredicate<ClusterNode>() { - @Override public boolean apply(ClusterNode node) { - return !node.isClient(); - } - }); + svcCfg.setNodeFilter((IgnitePredicate<ClusterNode>)node -> !node.isClient()); ignite.services().deploy(svcCfg); } http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceDeploymentCompoundFutureSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceDeploymentCompoundFutureSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceDeploymentCompoundFutureSelfTest.java index f2563d2..5a09aa2 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceDeploymentCompoundFutureSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceDeploymentCompoundFutureSelfTest.java @@ -25,6 +25,7 @@ import org.apache.ignite.IgniteException; import org.apache.ignite.internal.IgniteFutureTimeoutCheckedException; import org.apache.ignite.internal.util.future.GridFutureAdapter; import org.apache.ignite.internal.util.typedef.internal.U; +import org.apache.ignite.lang.IgniteUuid; import org.apache.ignite.services.ServiceConfiguration; import org.apache.ignite.services.ServiceDeploymentException; import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; @@ -40,7 +41,7 @@ public class GridServiceDeploymentCompoundFutureSelfTest extends GridCommonAbstr */ @Test public void testWaitForCompletionOnFailingFuture() throws Exception { - GridServiceDeploymentCompoundFuture compFut = new GridServiceDeploymentCompoundFuture(); + GridServiceDeploymentCompoundFuture<IgniteUuid> compFut = new GridServiceDeploymentCompoundFuture<>(); int failingFutsNum = 2; @@ -51,7 +52,7 @@ public class GridServiceDeploymentCompoundFutureSelfTest extends GridCommonAbstr for (int i = 0; i < failingFutsNum; i++) { ServiceConfiguration failingCfg = config("Failed-" + i); - GridServiceDeploymentFuture failingFut = new GridServiceDeploymentFuture(failingCfg); + GridServiceDeploymentFuture<IgniteUuid> failingFut = new GridServiceDeploymentFuture<>(failingCfg, IgniteUuid.randomUuid()); failingFuts.add(failingFut); @@ -61,7 +62,7 @@ public class GridServiceDeploymentCompoundFutureSelfTest extends GridCommonAbstr List<GridFutureAdapter<Object>> futs = new ArrayList<>(completingFutsNum); for (int i = 0; i < completingFutsNum; i++) { - GridServiceDeploymentFuture fut = new GridServiceDeploymentFuture(config(String.valueOf(i))); + GridServiceDeploymentFuture<IgniteUuid> fut = new GridServiceDeploymentFuture<>(config(String.valueOf(i)), IgniteUuid.randomUuid()); futs.add(fut); http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceDeploymentExceptionPropagationTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceDeploymentExceptionPropagationTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceDeploymentExceptionPropagationTest.java index d5df828..60f632a 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceDeploymentExceptionPropagationTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceDeploymentExceptionPropagationTest.java @@ -18,11 +18,13 @@ package org.apache.ignite.internal.processors.service; import org.apache.ignite.Ignite; -import org.apache.ignite.IgniteException; +import org.apache.ignite.internal.IgniteEx; import org.apache.ignite.services.Service; import org.apache.ignite.services.ServiceContext; -import org.apache.ignite.testframework.GridStringLogger; +import org.apache.ignite.services.ServiceDeploymentException; import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; +import org.junit.Assume; +import org.junit.BeforeClass; import org.junit.Test; import org.junit.runner.RunWith; import org.junit.runners.JUnit4; @@ -31,26 +33,37 @@ import org.junit.runners.JUnit4; @RunWith(JUnit4.class) public class GridServiceDeploymentExceptionPropagationTest extends GridCommonAbstractTest { /** */ - @SuppressWarnings("unused") + @BeforeClass + public static void check() { + Assume.assumeTrue(isEventDrivenServiceProcessorEnabled()); + } + + /** */ @Test public void testExceptionPropagation() throws Exception { - try (Ignite srv = startGrid("server")) { - - GridStringLogger log = new GridStringLogger(); - - try (Ignite client = startGrid("client", getConfiguration("client").setGridLogger(log).setClientMode(true))) { + try (IgniteEx srv = startGrid("server")) { + try (Ignite client = startGrid("client", getConfiguration("client").setClientMode(true))) { + final String srvcName = "my-service"; try { - client.services().deployClusterSingleton("my-service", new ServiceImpl()); - } - catch (IgniteException ignored) { - assertTrue(log.toString().contains("ServiceImpl init exception")); + client.services().deployClusterSingleton(srvcName, new ServiceImpl()); - return; // Exception is what we expect. + fail("Deployment exception has been expected."); } + catch (ServiceDeploymentException ex) { + String errMsg = ex.getSuppressed()[0].getMessage(); + + // Check that message contains cause node id + assertTrue(errMsg.contains(srv.cluster().localNode().id().toString())); + + // Check that message contains service name + assertTrue(errMsg.contains(srvcName)); - // Fail explicitly if we've managed to get here though we shouldn't have. - fail("https://issues.apache.org/jira/browse/IGNITE-3392";); + Throwable cause = ex.getSuppressed()[0].getCause(); + + // Check that error's cause contains users message + assertTrue(cause.getMessage().contains("ServiceImpl init exception")); + } } } } http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceProcessorAbstractSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceProcessorAbstractSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceProcessorAbstractSelfTest.java index 9197d93..f25a1e4 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceProcessorAbstractSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceProcessorAbstractSelfTest.java @@ -30,7 +30,9 @@ import org.apache.ignite.cache.CacheMode; import org.apache.ignite.cluster.ClusterNode; import org.apache.ignite.configuration.CacheConfiguration; import org.apache.ignite.configuration.IgniteConfiguration; +import org.apache.ignite.internal.IgniteEx; import org.apache.ignite.internal.IgniteInterruptedCheckedException; +import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion; import org.apache.ignite.internal.processors.affinity.GridAffinityProcessor; import org.apache.ignite.internal.util.lang.GridAbsPredicateX; import org.apache.ignite.internal.util.typedef.CA; @@ -139,7 +141,7 @@ public abstract class GridServiceProcessorAbstractSelfTest extends GridCommonAbs /** * @return Random grid. */ - protected Ignite randomGrid() { + protected IgniteEx randomGrid() { return grid(RAND.nextInt(nodeCount())); } @@ -249,11 +251,11 @@ public abstract class GridServiceProcessorAbstractSelfTest extends GridCommonAbs info("Deployed service: " + name); - fut1.get(); + try { + fut1.get(); - info("Finished waiting for service future: " + name); + info("Finished waiting for service future: " + name); - try { fut2.get(); fail("Failed to receive mismatching configuration exception."); @@ -764,6 +766,19 @@ public abstract class GridServiceProcessorAbstractSelfTest extends GridCommonAbs /** * @param svcName Service name. + * @param ignite Ignite instance. + * @param cnt Expected count. + */ + protected void checkCount(String svcName, IgniteEx ignite, int cnt) throws IgniteInterruptedCheckedException { + AffinityTopologyVersion topVer = ignite.context().discovery().topologyVersionEx(); + + waitForServicesReadyTopology(ignite, topVer); + + assertEquals(cnt, actualCount(svcName, ignite.services().serviceDescriptors())); + } + + /** + * @param svcName Service name. * @param descs Descriptors. * @param cnt Expected count. */ @@ -825,6 +840,32 @@ public abstract class GridServiceProcessorAbstractSelfTest extends GridCommonAbs } /** + * @param ignite Ignite instance. + * @param srvcName Affinity service name. + */ + protected void checkAffinityServiceDeployment(Ignite ignite, String srvcName) { + ServiceDescriptor desc = null; + + for (ServiceDescriptor d : ignite.services().serviceDescriptors()) { + if (d.name().equals(srvcName)) { + desc = d; + + break; + } + } + + assertNotNull(desc); + + assertEquals(1, desc.topologySnapshot().size()); + + ClusterNode n = ignite.affinity(desc.cacheName()).mapKeyToNode(desc.affinityKey()); + + assertNotNull(n); + + assertTrue(desc.topologySnapshot().containsKey(n.id())); + } + + /** * Affinity service. */ protected static class AffinityService implements Service { @@ -853,11 +894,6 @@ public abstract class GridServiceProcessorAbstractSelfTest extends GridCommonAbs /** {@inheritDoc} */ @Override public void init(ServiceContext ctx) throws Exception { X.println("Initializing affinity service for key: " + affKey); - - ClusterNode n = g.affinity(CACHE_NAME).mapKeyToNode(affKey); - - assertNotNull(n); - assertTrue(n.isLocal()); } /** {@inheritDoc} */ http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceProcessorBatchDeploySelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceProcessorBatchDeploySelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceProcessorBatchDeploySelfTest.java index b77144c..625b287 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceProcessorBatchDeploySelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceProcessorBatchDeploySelfTest.java @@ -29,6 +29,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import org.apache.ignite.Ignite; import org.apache.ignite.cluster.ClusterNode; +import org.apache.ignite.internal.IgniteEx; import org.apache.ignite.internal.IgniteInternalFuture; import org.apache.ignite.internal.IgniteNodeAttributes; import org.apache.ignite.lang.IgniteFuture; @@ -37,6 +38,7 @@ import org.apache.ignite.services.ServiceConfiguration; import org.apache.ignite.services.ServiceDeploymentException; import org.apache.ignite.services.ServiceDescriptor; import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; +import org.junit.Assume; import org.junit.Ignore; import org.junit.Test; import org.junit.runner.RunWith; @@ -132,16 +134,7 @@ public class GridServiceProcessorBatchDeploySelfTest extends GridCommonAbstractT CountDownLatch latch = new CountDownLatch(numServices); - IgnitePredicate<ClusterNode> depPred = client.cluster().forServers() - .forPredicate(new IgnitePredicate<ClusterNode>() { - @Override public boolean apply(ClusterNode node) { - String gridName = node.attribute(IgniteNodeAttributes.ATTR_IGNITE_INSTANCE_NAME); - - assert gridName != null; - - return gridName.startsWith(getTestIgniteInstanceName()); - } - }).predicate(); + IgnitePredicate<ClusterNode> depPred = new TestPredicate(getTestIgniteInstanceName()); List<ServiceConfiguration> cfgs = getConfigs(depPred, numServices); @@ -185,16 +178,7 @@ public class GridServiceProcessorBatchDeploySelfTest extends GridCommonAbstractT CountDownLatch latch = new CountDownLatch(numServices); - IgnitePredicate<ClusterNode> depPred = client.cluster().forServers() - .forPredicate(new IgnitePredicate<ClusterNode>() { - @Override public boolean apply(ClusterNode node) { - String gridName = node.attribute(IgniteNodeAttributes.ATTR_IGNITE_INSTANCE_NAME); - - assert gridName != null; - - return gridName.startsWith(getTestIgniteInstanceName()); - } - }).predicate(); + IgnitePredicate<ClusterNode> depPred = new TestPredicate(getTestIgniteInstanceName()); List<ServiceConfiguration> cfgs = getConfigs(depPred, numServices); @@ -430,10 +414,12 @@ public class GridServiceProcessorBatchDeploySelfTest extends GridCommonAbstractT /** * @throws Exception If failed. */ - @Ignore("https://issues.apache.org/jira/browse/IGNITE-10021";) @Test public void testCancelAllTopologyChange() throws Exception { - Ignite client = grid(CLIENT_NODE_NAME); + IgniteEx client = grid(CLIENT_NODE_NAME); + + Assume.assumeFalse("https://issues.apache.org/jira/browse/IGNITE-10021";, + client.context().service() instanceof GridServiceProcessor); int numServices = 500; @@ -661,4 +647,28 @@ public class GridServiceProcessorBatchDeploySelfTest extends GridCommonAbstractT } }); } + + /** + * Test predicate. + */ + private static class TestPredicate implements IgnitePredicate<ClusterNode> { + /** */ + private final String namePrefix; + + /** + * @param namePrefix Prefix to match instances name. + */ + public TestPredicate(String namePrefix) { + this.namePrefix = namePrefix; + } + + /** {@inheritDoc} */ + @Override public boolean apply(ClusterNode node) { + String gridName = node.attribute(IgniteNodeAttributes.ATTR_IGNITE_INSTANCE_NAME); + + assert gridName != null; + + return !node.isClient() && gridName.startsWith(namePrefix); + } + } } http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceProcessorMultiNodeConfigSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceProcessorMultiNodeConfigSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceProcessorMultiNodeConfigSelfTest.java index d328db2..b1e557f 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceProcessorMultiNodeConfigSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceProcessorMultiNodeConfigSelfTest.java @@ -20,9 +20,9 @@ package org.apache.ignite.internal.processors.service; import java.util.ArrayList; import java.util.List; import java.util.concurrent.CountDownLatch; -import org.apache.ignite.Ignite; import org.apache.ignite.configuration.CacheConfiguration; import org.apache.ignite.configuration.IgniteConfiguration; +import org.apache.ignite.internal.IgniteEx; import org.apache.ignite.internal.util.lang.GridAbsPredicateX; import org.apache.ignite.services.ServiceConfiguration; import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi; @@ -192,22 +192,28 @@ public class GridServiceProcessorMultiNodeConfigSelfTest extends GridServiceProc */ @Test public void testAffinityUpdateTopology() throws Exception { - Ignite g = randomGrid(); + IgniteEx g = randomGrid(); - checkCount(AFFINITY, g.services().serviceDescriptors(), 1); + checkCount(AFFINITY, g, 1); + + checkAffinityServiceDeployment(g, AFFINITY); int nodeCnt = 2; startExtraNodes(nodeCnt); try { - checkCount(AFFINITY, g.services().serviceDescriptors(), 1); + checkCount(AFFINITY, g, 1); + + checkAffinityServiceDeployment(g, AFFINITY); } finally { stopExtraNodes(nodeCnt); } - checkCount(AFFINITY, g.services().serviceDescriptors(), 1); + checkCount(AFFINITY, g, 1); + + checkAffinityServiceDeployment(g, AFFINITY); } /** @@ -215,13 +221,13 @@ public class GridServiceProcessorMultiNodeConfigSelfTest extends GridServiceProc */ @Test public void testDeployLimits() throws Exception { - final Ignite g = randomGrid(); + final IgniteEx g = randomGrid(); final String name = NODE_SINGLE_WITH_LIMIT; waitForDeployment(name, nodeCount()); - checkCount(name, g.services().serviceDescriptors(), nodeCount()); + checkCount(name, g, nodeCount()); CountDownLatch latch = new CountDownLatch(1); @@ -236,7 +242,7 @@ public class GridServiceProcessorMultiNodeConfigSelfTest extends GridServiceProc waitForDeployment(name, nodeCount() + 1); - checkCount(name, g.services().serviceDescriptors(), nodeCount() + 1); + checkCount(name, g, nodeCount() + 1); } finally { stopExtraNodes(extraNodes); @@ -247,7 +253,7 @@ public class GridServiceProcessorMultiNodeConfigSelfTest extends GridServiceProc // Service can be redeployed when nodes is stopping one-by-one. assertEquals(0, DummyService.started(name) - DummyService.cancelled(name)); - checkCount(name, g.services().serviceDescriptors(), nodeCount()); + checkCount(name, g, nodeCount()); } /** @@ -255,7 +261,7 @@ public class GridServiceProcessorMultiNodeConfigSelfTest extends GridServiceProc * @throws Exception If failed. */ private void checkSingletonUpdateTopology(String name) throws Exception { - Ignite g = randomGrid(); + IgniteEx g = randomGrid(); startExtraNodes(2, 2); @@ -265,7 +271,7 @@ public class GridServiceProcessorMultiNodeConfigSelfTest extends GridServiceProc info(">>> Passed checks."); - checkCount(name, g.services().serviceDescriptors(), 1); + checkCount(name, g, 1); } finally { stopExtraNodes(4); @@ -277,7 +283,7 @@ public class GridServiceProcessorMultiNodeConfigSelfTest extends GridServiceProc * @throws Exception If failed. */ private void checkDeployOnEachNodeUpdateTopology(String name) throws Exception { - Ignite g = randomGrid(); + IgniteEx g = randomGrid(); int newNodes = 4; @@ -297,7 +303,7 @@ public class GridServiceProcessorMultiNodeConfigSelfTest extends GridServiceProc // not start and cancel events individually. assertEquals(name, newNodes, DummyService.started(name) - DummyService.cancelled(name)); - checkCount(name, g.services().serviceDescriptors(), nodeCount() + newNodes); + checkCount(name, g, nodeCount() + newNodes); } finally { stopExtraNodes(newNodes); @@ -305,7 +311,7 @@ public class GridServiceProcessorMultiNodeConfigSelfTest extends GridServiceProc waitForDeployment(name, nodeCount()); - checkCount(name, g.services().serviceDescriptors(), nodeCount()); + checkCount(name, g, nodeCount()); } /** @@ -313,7 +319,7 @@ public class GridServiceProcessorMultiNodeConfigSelfTest extends GridServiceProc * @throws Exception If failed. */ private void checkDeployOnEachNodeButClientUpdateTopology(String name) throws Exception { - Ignite g = randomGrid(); + IgniteEx g = randomGrid(); int servers = 2; @@ -335,7 +341,7 @@ public class GridServiceProcessorMultiNodeConfigSelfTest extends GridServiceProc // not start and cancel events individually. assertEquals(name, servers, DummyService.started(name) - DummyService.cancelled(name)); - checkCount(name, g.services().serviceDescriptors(), nodeCount() + servers); + checkCount(name, g, nodeCount() + servers); } finally { stopExtraNodes(servers + clients); @@ -343,6 +349,6 @@ public class GridServiceProcessorMultiNodeConfigSelfTest extends GridServiceProc waitForDeployment(name, nodeCount()); - checkCount(name, g.services().serviceDescriptors(), nodeCount()); + checkCount(name, g, nodeCount()); } } http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceProcessorMultiNodeSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceProcessorMultiNodeSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceProcessorMultiNodeSelfTest.java index ed331fa..c6bb223 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceProcessorMultiNodeSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceProcessorMultiNodeSelfTest.java @@ -22,6 +22,7 @@ import junit.framework.TestCase; import org.apache.ignite.Ignite; import org.apache.ignite.IgniteServices; import org.apache.ignite.configuration.CacheConfiguration; +import org.apache.ignite.internal.IgniteEx; import org.apache.ignite.lang.IgniteFuture; import org.apache.ignite.services.Service; import org.apache.ignite.services.ServiceConfiguration; @@ -46,7 +47,7 @@ public class GridServiceProcessorMultiNodeSelfTest extends GridServiceProcessorA public void testSingletonUpdateTopology() throws Exception { String name = "serviceSingletonUpdateTopology"; - Ignite g = randomGrid(); + IgniteEx g = randomGrid(); CountDownLatch latch = new CountDownLatch(1); @@ -77,7 +78,7 @@ public class GridServiceProcessorMultiNodeSelfTest extends GridServiceProcessorA info(">>> Passed checks."); - checkCount(name, g.services().serviceDescriptors(), 1); + checkCount(name, g, 1); } finally { stopExtraNodes(nodeCnt); @@ -89,7 +90,7 @@ public class GridServiceProcessorMultiNodeSelfTest extends GridServiceProcessorA */ @Test public void testAffinityDeployUpdateTopology() throws Exception { - Ignite g = randomGrid(); + IgniteEx g = randomGrid(); final Integer affKey = 1; @@ -116,7 +117,7 @@ public class GridServiceProcessorMultiNodeSelfTest extends GridServiceProcessorA startExtraNodes(nodeCnt); try { - checkCount(name, g.services().serviceDescriptors(), 1); + checkCount(name, g, 1); } finally { stopExtraNodes(nodeCnt); @@ -134,7 +135,7 @@ public class GridServiceProcessorMultiNodeSelfTest extends GridServiceProcessorA try { final String name = "serviceOnEachNodeButClientUpdateTopology"; - Ignite g = randomGrid(); + IgniteEx g = randomGrid(); CountDownLatch latch = new CountDownLatch(nodeCount()); @@ -178,7 +179,7 @@ public class GridServiceProcessorMultiNodeSelfTest extends GridServiceProcessorA // not start and cancel events individually. assertEquals(name, nodeCount() + servers, DummyService.started(name) - DummyService.cancelled(name)); - checkCount(name, g.services().serviceDescriptors(), nodeCount() + servers); + checkCount(name, g, nodeCount() + servers); } finally { stopExtraNodes(servers + clients); @@ -200,7 +201,7 @@ public class GridServiceProcessorMultiNodeSelfTest extends GridServiceProcessorA try { final String name = "serviceOnEachProjectionNodeUpdateTopology"; - Ignite g = randomGrid(); + IgniteEx g = randomGrid(); int prestartedSrvcs = 1; @@ -246,7 +247,7 @@ public class GridServiceProcessorMultiNodeSelfTest extends GridServiceProcessorA // not start and cancel events individually. assertEquals(name, clients + prestartedSrvcs, DummyService.started(name) - DummyService.cancelled(name)); - checkCount(name, g.services().serviceDescriptors(), clients + prestartedSrvcs); + checkCount(name, g, clients + prestartedSrvcs); } finally { stopExtraNodes(servers + clients); @@ -268,7 +269,7 @@ public class GridServiceProcessorMultiNodeSelfTest extends GridServiceProcessorA try { final String name = "serviceOnEachNodeUpdateTopology"; - Ignite g = randomGrid(); + IgniteEx g = randomGrid(); final int prestartedNodes = nodeCount() + 1; @@ -323,7 +324,7 @@ public class GridServiceProcessorMultiNodeSelfTest extends GridServiceProcessorA assertEquals(name, prestartedNodes + extraNodes, DummyService.started(name) - DummyService.cancelled(name)); - checkCount(name, g.services().serviceDescriptors(), prestartedNodes + extraNodes); + checkCount(name, g, prestartedNodes + extraNodes); } finally { stopExtraNodes(extraNodes); @@ -342,7 +343,7 @@ public class GridServiceProcessorMultiNodeSelfTest extends GridServiceProcessorA public void testDeployLimits() throws Exception { final String name = "serviceWithLimitsUpdateTopology"; - Ignite g = randomGrid(); + IgniteEx g = randomGrid(); final int totalInstances = nodeCount() + 1; @@ -394,7 +395,7 @@ public class GridServiceProcessorMultiNodeSelfTest extends GridServiceProcessorA // not start and cancel events individually. assertEquals(name, totalInstances, DummyService.started(name) - DummyService.cancelled(name)); - checkCount(name, g.services().serviceDescriptors(), totalInstances); + checkCount(name, g, totalInstances); } finally { stopExtraNodes(extraNodes); http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceProcessorProxySelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceProcessorProxySelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceProcessorProxySelfTest.java index b022622..33db5fa 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceProcessorProxySelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceProcessorProxySelfTest.java @@ -230,7 +230,8 @@ public class GridServiceProcessorProxySelfTest extends GridServiceProcessorAbstr ignite.services(ignite.cluster().forLocal()).deployClusterSingleton(name, new MapServiceImpl<String, Integer>()); for (int i = 1; i < nodeCount(); i++) { - MapService<Integer, String> svc = grid(i).services().serviceProxy(name, MapService.class, false); + MapService<Integer, String> svc = grid(i).services() + .serviceProxy(name, MapService.class, false, 1_000L); // Make sure service is a proxy. assertFalse(svc instanceof Service); @@ -445,7 +446,7 @@ public class GridServiceProcessorProxySelfTest extends GridServiceProcessorAbstr /** * */ - protected class ErrorServiceImpl implements ErrorService { + protected static class ErrorServiceImpl implements ErrorService { /** {@inheritDoc} */ @Override public void cancel(ServiceContext ctx) { // No-op. http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceProcessorSingleNodeSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceProcessorSingleNodeSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceProcessorSingleNodeSelfTest.java index e2e5a5d..56d8a34 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceProcessorSingleNodeSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceProcessorSingleNodeSelfTest.java @@ -17,8 +17,8 @@ package org.apache.ignite.internal.processors.service; -import org.apache.ignite.Ignite; import org.apache.ignite.IgniteException; +import org.apache.ignite.internal.IgniteEx; import org.junit.Test; import org.junit.runner.RunWith; import org.junit.runners.JUnit4; @@ -33,7 +33,6 @@ public class GridServiceProcessorSingleNodeSelfTest extends GridServiceProcessor return 1; } - /** * @throws Exception If failed. */ @@ -41,17 +40,20 @@ public class GridServiceProcessorSingleNodeSelfTest extends GridServiceProcessor public void testNodeSingletonNotDeployedProxy() throws Exception { String name = "testNodeSingletonNotDeployedProxy"; - Ignite ignite = randomGrid(); + IgniteEx ignite = randomGrid(); + + try { + // Deploy only on remote nodes. + ignite.services(ignite.cluster().forRemotes()).deployNodeSingleton(name, new CounterServiceImpl()); - // Deploy only on remote nodes. - ignite.services(ignite.cluster().forRemotes()).deployNodeSingleton(name, new CounterServiceImpl()); + assertFalse("Should not reach here in this mode, because exception should be thrown.", + ignite.context().service() instanceof IgniteServiceProcessor); - info("Deployed service: " + name); + info("Deployed service: " + name); - // Get local proxy. - CounterService svc = ignite.services().serviceProxy(name, CounterService.class, false); + // Get local proxy. + CounterService svc = ignite.services().serviceProxy(name, CounterService.class, false); - try { svc.increment(); fail("Should never reach here."); http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceProcessorStopSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceProcessorStopSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceProcessorStopSelfTest.java index d17b015..ac3aaa6 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceProcessorStopSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceProcessorStopSelfTest.java @@ -180,7 +180,7 @@ public class GridServiceProcessorStopSelfTest extends GridCommonAbstractTest { /** * */ - public class TestServiceImpl implements Service, TestService { + public static class TestServiceImpl implements Service, TestService { /** Serial version UID. */ private static final long serialVersionUID = 0L; http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceReassignmentSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceReassignmentSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceReassignmentSelfTest.java index 532728b..096c9be 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceReassignmentSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceReassignmentSelfTest.java @@ -26,7 +26,6 @@ import java.util.concurrent.CountDownLatch; import org.apache.ignite.Ignite; import org.apache.ignite.IgniteCheckedException; import org.apache.ignite.internal.IgniteEx; -import org.apache.ignite.internal.processors.cache.IgniteInternalCache; import org.apache.ignite.internal.util.typedef.F; import org.apache.ignite.internal.util.typedef.internal.U; import org.apache.ignite.testframework.GridTestUtils; @@ -42,6 +41,9 @@ public class GridServiceReassignmentSelfTest extends GridServiceProcessorAbstrac /** */ private static final String SERVICE_NAME = "testService"; + /** */ + private static final long SERVICE_TOP_WAIT_TIMEOUT = 2_000L; + /** {@inheritDoc} */ @Override protected int nodeCount() { return 1; @@ -164,17 +166,17 @@ public class GridServiceReassignmentSelfTest extends GridServiceProcessorAbstrac private boolean checkServices(int total, int maxPerNode, int gridIdx, boolean lastTry) throws Exception { IgniteEx grid = grid(gridIdx); - IgniteInternalCache<GridServiceAssignmentsKey, GridServiceAssignments> cache = grid.utilityCache(); + waitForServicesReadyTopology(grid, grid.context().discovery().topologyVersionEx()); - GridServiceAssignments assignments = cache.get(new GridServiceAssignmentsKey(SERVICE_NAME)); + Map<UUID, Integer> srvcTop = grid.context().service().serviceTopology(SERVICE_NAME, SERVICE_TOP_WAIT_TIMEOUT); - Collection<UUID> nodes = F.viewReadOnly(grid.cluster().nodes(), F.node2id()); + Collection<UUID> nodes = F.viewReadOnly(grid.context().discovery().aliveServerNodes(), F.node2id()); - assertNotNull("Grid assignments object is null", assignments); + assertNotNull("Grid assignments object is null", srvcTop); int sum = 0; - for (Map.Entry<UUID, Integer> entry : assignments.assigns().entrySet()) { + for (Map.Entry<UUID, Integer> entry : srvcTop.entrySet()) { UUID nodeId = entry.getKey(); if (!lastTry && !nodes.contains(nodeId)) @@ -193,9 +195,9 @@ public class GridServiceReassignmentSelfTest extends GridServiceProcessorAbstrac if (total > 0) assertTrue("Total number of services limit exceeded [sum=" + sum + - ", assigns=" + assignments.assigns() + ']', sum <= total); + ", assigns=" + srvcTop + ']', sum <= total); else - assertEquals("Reassign per node failed.", nodes.size(), assignments.assigns().size()); + assertEquals("Reassign per node failed.", nodes.size(), srvcTop.size()); if (!lastTry && proxy(grid).get() != 10) return false; http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceSerializationSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceSerializationSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceSerializationSelfTest.java index 92a7449..9c9cc73 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceSerializationSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceSerializationSelfTest.java @@ -54,7 +54,7 @@ public class GridServiceSerializationSelfTest extends GridCommonAbstractTest { server.services(server.cluster().forServers()) .deployClusterSingleton("my-service", new MyServiceImpl()); - MyService svc = client.services().serviceProxy("my-service", MyService.class, false); + MyService svc = client.services().serviceProxy("my-service", MyService.class, false, 2_000); svc.hello(); http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/test/java/org/apache/ignite/internal/processors/service/IgniteServiceConfigVariationsFullApiTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/service/IgniteServiceConfigVariationsFullApiTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/IgniteServiceConfigVariationsFullApiTest.java index 0ebee6e..f64cef4 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/service/IgniteServiceConfigVariationsFullApiTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/IgniteServiceConfigVariationsFullApiTest.java @@ -21,18 +21,14 @@ import java.io.Externalizable; import java.io.IOException; import java.io.ObjectInput; import java.io.ObjectOutput; -import java.io.Serializable; import java.util.concurrent.ThreadLocalRandom; import javax.cache.configuration.Factory; import org.apache.ignite.IgniteCache; import org.apache.ignite.IgniteCheckedException; import org.apache.ignite.IgniteException; import org.apache.ignite.IgniteServices; -import org.apache.ignite.binary.BinaryObjectException; -import org.apache.ignite.binary.BinaryReader; -import org.apache.ignite.binary.BinaryWriter; -import org.apache.ignite.binary.Binarylizable; import org.apache.ignite.internal.IgniteEx; +import org.apache.ignite.internal.IgniteInterruptedCheckedException; import org.apache.ignite.services.Service; import org.apache.ignite.services.ServiceConfiguration; import org.apache.ignite.services.ServiceContext; @@ -67,9 +63,18 @@ public class IgniteServiceConfigVariationsFullApiTest extends IgniteConfigVariat private static final Factory[] serviceFactories = new Factory[] { Parameters.factory(TestServiceImpl.class), Parameters.factory(TestServiceImplExternalizable.class), - Parameters.factory(TestServiceImplBinarylizable.class) }; + /** */ + private static boolean isEventDrivenServiceProcessorEnabled; + + /** {@inheritDoc} */ + @Override protected void beforeTestsStarted() throws Exception { + super.beforeTestsStarted(); + + isEventDrivenServiceProcessorEnabled = grid(0).context().service() instanceof IgniteServiceProcessor; + } + /** {@inheritDoc} */ @Override protected boolean expectedClient(String testGridName) { int i = testsCfg.gridCount(); @@ -93,8 +98,7 @@ public class IgniteServiceConfigVariationsFullApiTest extends IgniteConfigVariat @Override public void run(IgniteServices services, String svcName, TestService svc) throws Exception { services.deployNodeSingleton(svcName, (Service)svc); - // TODO: Waiting for deployment should be removed after IEP-17 completion - GridTestUtils.waitForCondition(() -> services.service(svcName) != null, DEPLOYMENT_WAIT_TIMEOUT); + waitForServiceDeploymentIfNeeded(services, svcName); } })); } @@ -110,8 +114,7 @@ public class IgniteServiceConfigVariationsFullApiTest extends IgniteConfigVariat @Override public void run(IgniteServices services, String svcName, TestService svc) throws Exception { services.deployClusterSingleton(svcName, (Service)svc); - // TODO: Waiting for deployment should be removed after IEP-17 completion - GridTestUtils.waitForCondition(() -> services.service(svcName) != null, DEPLOYMENT_WAIT_TIMEOUT); + waitForServiceDeploymentIfNeeded(services, svcName); } })); } @@ -145,8 +148,10 @@ public class IgniteServiceConfigVariationsFullApiTest extends IgniteConfigVariat @Test public void testMultipleDeploy() throws Exception { runInAllDataModes(new ServiceTestRunnable(true, new DeployClosure() { - @Override public void run(IgniteServices services, String svcName, TestService svc) { + @Override public void run(IgniteServices services, String svcName, TestService svc) throws Exception { services.deployMultiple(svcName, (Service)svc, 0, 1); + + waitForServiceDeploymentIfNeeded(services, svcName); } })); } @@ -176,8 +181,7 @@ public class IgniteServiceConfigVariationsFullApiTest extends IgniteConfigVariat services.deploy(cfg); - // TODO: Waiting for deployment should be removed after IEP-17 completion - GridTestUtils.waitForCondition(() -> services.service(svcName) != null, DEPLOYMENT_WAIT_TIMEOUT); + waitForServiceDeploymentIfNeeded(services, svcName); } })); } @@ -263,7 +267,7 @@ public class IgniteServiceConfigVariationsFullApiTest extends IgniteConfigVariat // Expect correct value after being read back. int r = 1000; - while(r-- > 0) + while (r-- > 0) assertEquals(expected, proxy.getValue()); assertEquals("Expected 1 deployed service", 1, services.serviceDescriptors().size()); @@ -279,6 +283,29 @@ public class IgniteServiceConfigVariationsFullApiTest extends IgniteConfigVariat } } + /** {@inheritDoc} */ + @Override protected boolean isCompatible() throws Exception { + switch (dataMode) { + case SERIALIZABLE: + case CUSTOM_SERIALIZABLE: + case EXTERNALIZABLE: + return true; + } + + return false; + } + + /** + * @param services Ignite services. + * @param srvcName Service name to wait. + * @throws IgniteInterruptedCheckedException If interrupted. + */ + private void waitForServiceDeploymentIfNeeded(IgniteServices services, + String srvcName) throws IgniteInterruptedCheckedException { + if (!isEventDrivenServiceProcessorEnabled) + GridTestUtils.waitForCondition(() -> services.service(srvcName) != null, DEPLOYMENT_WAIT_TIMEOUT); + } + /** * Test service */ @@ -298,7 +325,7 @@ public class IgniteServiceConfigVariationsFullApiTest extends IgniteConfigVariat /** * Implementation for {@link TestService} */ - public static class TestServiceImpl implements Service, TestService, Serializable { + public static class TestServiceImpl implements Service, TestService { /** Test value. */ protected Object val; @@ -357,27 +384,4 @@ public class IgniteServiceConfigVariationsFullApiTest extends IgniteConfigVariat val = in.readObject(); } } - - /** - * Echo service, binarylizable object - */ - @SuppressWarnings({"PublicInnerClass"}) - public static class TestServiceImplBinarylizable extends TestServiceImpl implements Binarylizable { - /** - * Default constructor. - */ - public TestServiceImplBinarylizable() { - // No-op. - } - - /** {@inheritDoc} */ - @Override public void writeBinary(BinaryWriter writer) throws BinaryObjectException { - writer.writeObject("arg", val); - } - - /** {@inheritDoc} */ - @Override public void readBinary(BinaryReader reader) throws BinaryObjectException { - val = reader.readObject("arg"); - } - } } http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/test/java/org/apache/ignite/internal/processors/service/IgniteServiceDynamicCachesSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/service/IgniteServiceDynamicCachesSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/IgniteServiceDynamicCachesSelfTest.java index 790beb3..a648834 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/service/IgniteServiceDynamicCachesSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/IgniteServiceDynamicCachesSelfTest.java @@ -21,10 +21,12 @@ import org.apache.ignite.Ignite; import org.apache.ignite.IgniteLogger; import org.apache.ignite.IgniteServices; import org.apache.ignite.configuration.CacheConfiguration; +import org.apache.ignite.internal.IgniteEx; import org.apache.ignite.internal.util.typedef.PA; import org.apache.ignite.resources.LoggerResource; import org.apache.ignite.services.Service; import org.apache.ignite.services.ServiceContext; +import org.apache.ignite.services.ServiceDeploymentException; import org.apache.ignite.testframework.GridTestUtils; import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; import org.junit.Test; @@ -94,13 +96,14 @@ public class IgniteServiceDynamicCachesSelfTest extends GridCommonAbstractTest { * @throws Exception If failed. */ @Test + @SuppressWarnings("ThrowableResultOfMethodCallIgnored") public void testDeployCalledBeforeCacheStart() throws Exception { String cacheName = "cache"; CacheConfiguration ccfg = new CacheConfiguration(cacheName); ccfg.setBackups(1); - Ignite ig = ignite(0); + IgniteEx ig = grid(0); final IgniteServices svcs = ig.services(); @@ -114,11 +117,26 @@ public class IgniteServiceDynamicCachesSelfTest extends GridCommonAbstractTest { awaitPartitionMapExchange(); - svcs.deployKeyAffinitySingleton(svcName, new TestService(), cacheName, key); + if (ig.context().service() instanceof GridServiceProcessor) { + svcs.deployKeyAffinitySingleton(svcName, new TestService(), cacheName, key); - assert svcs.service(svcName) == null; + assertNull(svcs.service(svcName)); - ig.createCache(ccfg); + ig.createCache(ccfg); + } + else if (ig.context().service() instanceof IgniteServiceProcessor) { + GridTestUtils.assertThrowsWithCause(() -> { + svcs.deployKeyAffinitySingleton(svcName, new TestService(), cacheName, key); + + return null; + }, ServiceDeploymentException.class); + + ig.createCache(ccfg); + + svcs.deployKeyAffinitySingleton(svcName, new TestService(), cacheName, key); + } + else + fail("Unexpected service implementation."); try { boolean res = GridTestUtils.waitForCondition(new PA() { http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/test/java/org/apache/ignite/internal/processors/service/IgniteServiceReassignmentTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/service/IgniteServiceReassignmentTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/IgniteServiceReassignmentTest.java index c118d6d..ff30c29 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/service/IgniteServiceReassignmentTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/IgniteServiceReassignmentTest.java @@ -35,6 +35,7 @@ import org.apache.ignite.services.ServiceContext; import org.apache.ignite.testframework.GridStringLogger; import org.apache.ignite.testframework.GridTestUtils; import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; +import org.junit.Assume; import org.junit.Test; import org.junit.runner.RunWith; import org.junit.runners.JUnit4; @@ -87,13 +88,15 @@ public class IgniteServiceReassignmentTest extends GridCommonAbstractTest { public void testNodeRestart1() throws Exception { srvcCfg = serviceConfiguration(); - Ignite node1 = startGrid(1); + IgniteEx node1 = startGrid(1); + + waitForService(node1); assertEquals(42, serviceProxy(node1).foo()); srvcCfg = serviceConfiguration(); - Ignite node2 = startGrid(2); + IgniteEx node2 = startGrid(2); node1.close(); @@ -103,7 +106,9 @@ public class IgniteServiceReassignmentTest extends GridCommonAbstractTest { srvcCfg = serviceConfiguration(); - Ignite node3 = startGrid(3); + IgniteEx node3 = startGrid(3); + + waitForService(node3); assertEquals(42, serviceProxy(node3).foo()); @@ -111,6 +116,10 @@ public class IgniteServiceReassignmentTest extends GridCommonAbstractTest { node1 = startGrid(1); + waitForService(node1); + waitForService(node2); + waitForService(node3); + assertEquals(42, serviceProxy(node1).foo()); assertEquals(42, serviceProxy(node2).foo()); assertEquals(42, serviceProxy(node3).foo()); @@ -118,6 +127,7 @@ public class IgniteServiceReassignmentTest extends GridCommonAbstractTest { node2.close(); waitForService(node1); + waitForService(node3); assertEquals(42, serviceProxy(node1).foo()); assertEquals(42, serviceProxy(node3).foo()); @@ -174,7 +184,7 @@ public class IgniteServiceReassignmentTest extends GridCommonAbstractTest { if (nodeIdx == stopIdx) continue; - waitForService(ignite(nodeIdx)); + waitForService(grid(nodeIdx)); assertEquals(42, serviceProxy(ignite(nodeIdx)).foo()); } @@ -191,6 +201,8 @@ public class IgniteServiceReassignmentTest extends GridCommonAbstractTest { */ @Test public void testZombieAssignmentsCleanup() throws Exception { + Assume.assumeTrue(!isEventDrivenServiceProcessorEnabled()); + useStrLog = true; final int nodesCnt = 2; @@ -248,6 +260,8 @@ public class IgniteServiceReassignmentTest extends GridCommonAbstractTest { */ @Test public void testNodeStopWhileThereAreCacheActivitiesInServiceProcessor() throws Exception { + Assume.assumeTrue(!isEventDrivenServiceProcessorEnabled()); + final int nodesCnt = 2; final int maxSvc = 1024; @@ -286,19 +300,23 @@ public class IgniteServiceReassignmentTest extends GridCommonAbstractTest { * @param node Node. * @throws Exception If failed. */ - private void waitForService(final Ignite node) throws Exception { - assertTrue(GridTestUtils.waitForCondition(new PA() { - @Override public boolean apply() { - try { - serviceProxy(node).foo(); - - return true; + private void waitForService(final IgniteEx node) throws Exception { + if (node.context().service() instanceof IgniteServiceProcessor) + waitForServicesReadyTopology(node, node.context().discovery().topologyVersionEx()); + else { + assertTrue(GridTestUtils.waitForCondition(new PA() { + @Override public boolean apply() { + try { + serviceProxy(node).foo(); + + return true; + } + catch (IgniteException ignored) { + return false; + } } - catch (IgniteException ignored) { - return false; - } - } - }, 5000)); + }, 5000)); + } } /** http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/test/java/org/apache/ignite/internal/processors/service/ServiceDeploymentDiscoveryListenerNotificationOrderTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/service/ServiceDeploymentDiscoveryListenerNotificationOrderTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/ServiceDeploymentDiscoveryListenerNotificationOrderTest.java new file mode 100644 index 0000000..b07ecb7 --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/ServiceDeploymentDiscoveryListenerNotificationOrderTest.java @@ -0,0 +1,115 @@ + /* + * 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.service; + + import java.util.Collection; + import java.util.EventListener; + import java.util.List; + import java.util.concurrent.ConcurrentMap; + import org.apache.ignite.internal.IgniteEx; + import org.apache.ignite.internal.events.DiscoveryCustomEvent; + import org.apache.ignite.internal.managers.eventstorage.GridEventStorageManager; + import org.apache.ignite.internal.processors.cache.GridCachePartitionExchangeManager; + import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsExchangeFuture; + import org.apache.ignite.internal.util.GridConcurrentLinkedHashSet; + import org.apache.ignite.testframework.GridTestUtils; + import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; + import org.junit.Assume; + import org.junit.BeforeClass; + import org.junit.Test; + import org.junit.runner.RunWith; + import org.junit.runners.JUnit4; + + /** + * <b>Tests in the class strongly depend on implementation of {@link GridEventStorageManager} and internal logic of + * services and PME discovery listeners.</b> + * <p/> + * Tests that discovery listener registered by {@link ServiceDeploymentManager} will be notified earlier than discovery + * listener registered by {@link GridCachePartitionExchangeManager}. It allows service manager capture custom message + * because it may be nullified in PME process at the end of exchange in {@link GridDhtPartitionsExchangeFuture#onDone()}. + */ + @RunWith(JUnit4.class) + public class ServiceDeploymentDiscoveryListenerNotificationOrderTest extends GridCommonAbstractTest { + /** */ + @BeforeClass + public static void check() { + Assume.assumeTrue(isEventDrivenServiceProcessorEnabled()); + } + + /** + * <b>Strongly depends on internal implementation of {@link GridEventStorageManager}.</b> + * <p/> + * Tests that discovery listener registered by {@link ServiceDeploymentManager} is in collection of hight priority + * listeners, at the same time discovery listener registered by {@link GridCachePartitionExchangeManager} is in + * collection of usual listeners. + * <p/> + * This guarantees that service deployment discovery listener will be notified earlier that PME's discovery listener + * and will be able to capture custom messages which may be nullified in PME process. + * + * @throws Exception In case of an error. + */ + @Test + public void testServiceDiscoveryListenerNotifiedEarlierThanPME() throws Exception { + try { + IgniteEx ignite = startGrid(0); + + ConcurrentMap<Integer, Object> lsnrs = GridTestUtils + .getFieldValue(ignite.context().event(), "lsnrs"); + + Object customLsnrs = lsnrs.get(DiscoveryCustomEvent.EVT_DISCOVERY_CUSTOM_EVT); + + List<EventListener> highPriorityLsnrs = GridTestUtils.getFieldValue(customLsnrs, "highPriorityLsnrs"); + + GridConcurrentLinkedHashSet<EventListener> usualLsnrs = GridTestUtils + .getFieldValue(customLsnrs, "lsnrs"); + + assertTrue("Failed to find service deployment manager's listener in high priority listeners.", + containsListener(highPriorityLsnrs, ServiceDeploymentManager.class)); + + assertFalse("Service deployment manager's listener shoud not be registered in usual listeners.", + containsListener(usualLsnrs, ServiceDeploymentManager.class)); + + assertTrue("Failed to find PME manager's discovery listener in usual listeners.", + containsListener(usualLsnrs, GridCachePartitionExchangeManager.class)); + + assertFalse("PME manager's discovery listener shoud not be registered in high priority listeners.", + containsListener(highPriorityLsnrs, GridCachePartitionExchangeManager.class)); + } + finally { + stopAllGrids(); + } + } + + /** + * @param lsnrs Collection of listeners. + * @param cls Listener class. + * @return {@code true} if given collection contains expected listener, otherwise {@code false}. + */ + private boolean containsListener(Collection<EventListener> lsnrs, Class cls) { + for (Object wrapper : lsnrs) { + Object lsnr = GridTestUtils.getFieldValue(wrapper, "lsnr"); + + // We can't use 'instance of' or other reflection tools to check the type of class, because, service + // deployment discovery listener is a private non-static class, PME's listener is an anonymous class. + if (lsnr.getClass().getName().startsWith(cls.getName())) + return true; + } + + return false; + } + } \ No newline at end of file http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/test/java/org/apache/ignite/internal/processors/service/ServiceDeploymentNonSerializableStaticConfigurationTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/service/ServiceDeploymentNonSerializableStaticConfigurationTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/ServiceDeploymentNonSerializableStaticConfigurationTest.java new file mode 100644 index 0000000..4d08d10 --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/ServiceDeploymentNonSerializableStaticConfigurationTest.java @@ -0,0 +1,119 @@ +/* + * 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.service; + +import org.apache.ignite.configuration.IgniteConfiguration; +import org.apache.ignite.internal.IgniteEx; +import org.apache.ignite.services.Service; +import org.apache.ignite.services.ServiceConfiguration; +import org.apache.ignite.services.ServiceContext; +import org.apache.ignite.testframework.ListeningTestLogger; +import org.apache.ignite.testframework.LogListener; +import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; +import org.junit.Assume; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +/** + * + */ +@RunWith(JUnit4.class) +public class ServiceDeploymentNonSerializableStaticConfigurationTest extends GridCommonAbstractTest { + /** */ + private static final String TEST_SERVICE_NAME = "nonSerializableService"; + + /** */ + private final ListeningTestLogger log = new ListeningTestLogger(false, super.log); + + /** */ + @BeforeClass + public static void check() { + Assume.assumeTrue(isEventDrivenServiceProcessorEnabled()); + } + + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception { + IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName); + + cfg.setGridLogger(log); + + ServiceConfiguration srvcCfg = new ServiceConfiguration(); + + srvcCfg.setName(TEST_SERVICE_NAME); + srvcCfg.setMaxPerNodeCount(1); + srvcCfg.setService(new NonSerializableService()); + + cfg.setServiceConfiguration(srvcCfg); + + return cfg; + } + + /** + * @throws Exception In case of an error. + */ + @Test + public void testNonSerializableStaticServiceValidationFailure() throws Exception { + LogListener lsnr = LogListener + .matches(s -> s.startsWith("Failed to marshal service with configured marshaller [name=" + TEST_SERVICE_NAME)) + .atLeast(2) + .build(); + + log.registerListener(lsnr); + + try { + startGrid(0); + + IgniteEx ignite = startGrid(1); + + assertEquals(2, ignite.context().discovery().topologyVersion()); + + assertTrue(lsnr.check()); + } + finally { + stopAllGrids(); + } + } + + /** */ + private static class NonSerializableService implements Service { + /** */ + @SuppressWarnings("unused") + private NonSerializableObject nonSerializableField = new NonSerializableObject(); + + /** {@inheritDoc} */ + @Override public void cancel(ServiceContext ctx) { + // No-op. + } + + /** {@inheritDoc} */ + @Override public void init(ServiceContext ctx) throws Exception { + // No-op. + } + + /** {@inheritDoc} */ + @Override public void execute(ServiceContext ctx) throws Exception { + // No-op. + } + } + + /** */ + private static class NonSerializableObject { + } +}
