You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sb...@apache.org on 2018/12/27 16:40:28 UTC
[22/38] ignite git commit: IGNITE-9607: Service Grid redesign - Phase
1 - Fixes #4434.
IGNITE-9607: Service Grid redesign - Phase 1 - Fixes #4434.
Signed-off-by: Nikolay Izhikov <ni...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/62c560a5
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/62c560a5
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/62c560a5
Branch: refs/heads/ignite-601
Commit: 62c560a544a81a5e5b90099fa9da7ab51fc585fa
Parents: 457090a
Author: Vyacheslav Daradur <da...@gmail.com>
Authored: Wed Dec 26 23:17:30 2018 +0300
Committer: Nikolay Izhikov <ni...@apache.org>
Committed: Wed Dec 26 23:17:30 2018 +0300
----------------------------------------------------------------------
.../apache/ignite/IgniteSystemProperties.java | 13 +
.../apache/ignite/internal/GridComponent.java | 5 +-
.../ignite/internal/GridKernalContext.java | 4 +-
.../ignite/internal/GridKernalContextImpl.java | 12 +-
.../org/apache/ignite/internal/GridTopic.java | 5 +-
.../apache/ignite/internal/IgniteKernal.java | 23 +-
.../ignite/internal/IgniteNodeAttributes.java | 4 +
.../communication/GridIoMessageFactory.java | 18 +
.../discovery/GridDiscoveryManager.java | 23 +
.../cache/DynamicCacheChangeBatch.java | 20 +
.../processors/cache/GridCacheProcessor.java | 7 +-
.../GridDhtPartitionsExchangeFuture.java | 4 +-
.../cluster/ChangeGlobalStateMessage.java | 19 +
.../cluster/GridClusterStateProcessor.java | 9 +-
.../continuous/GridContinuousProcessor.java | 4 +-
.../service/GridServiceAssignments.java | 3 +
.../service/GridServiceAssignmentsKey.java | 3 +
.../service/GridServiceDeployment.java | 3 +
.../GridServiceDeploymentCompoundFuture.java | 15 +-
.../service/GridServiceDeploymentFuture.java | 24 +-
.../service/GridServiceDeploymentKey.java | 3 +
.../service/GridServiceProcessor.java | 187 +-
.../processors/service/GridServiceProxy.java | 10 +-
.../service/IgniteServiceProcessor.java | 1784 ++++++++++++++++++
.../service/PreparedConfigurations.java | 7 +-
.../service/ServiceChangeAbstractRequest.java | 53 +
.../service/ServiceChangeBatchRequest.java | 113 ++
.../service/ServiceClusterDeploymentResult.java | 73 +
.../ServiceClusterDeploymentResultBatch.java | 125 ++
.../service/ServiceDeploymentActions.java | 115 ++
.../service/ServiceDeploymentManager.java | 562 ++++++
.../service/ServiceDeploymentProcessId.java | 172 ++
.../service/ServiceDeploymentRequest.java | 56 +
.../service/ServiceDeploymentTask.java | 859 +++++++++
.../service/ServiceDescriptorImpl.java | 4 +
.../processors/service/ServiceInfo.java | 170 ++
.../service/ServiceProcessorAdapter.java | 162 ++
.../ServiceProcessorCommonDiscoveryData.java | 55 +
.../ServiceProcessorJoinNodeDiscoveryData.java | 53 +
.../ServiceSingleNodeDeploymentResult.java | 162 ++
.../ServiceSingleNodeDeploymentResultBatch.java | 155 ++
.../service/ServiceUndeploymentRequest.java | 42 +
.../ignite/services/ServiceConfiguration.java | 2 +
.../ignite/spi/discovery/tcp/ServerImpl.java | 44 +
.../ignite/internal/GridDeploymentSelfTest.java | 2 +-
.../IgniteClientReconnectServicesTest.java | 49 +
.../GridDiscoveryManagerAttributesSelfTest.java | 12 +
...ridCacheContinuousQueryAbstractSelfTest.java | 9 +-
.../GridServiceContinuousQueryRedeployTest.java | 6 +-
...ServiceDeploymentCompoundFutureSelfTest.java | 7 +-
...rviceDeploymentExceptionPropagationTest.java | 43 +-
.../GridServiceProcessorAbstractSelfTest.java | 54 +-
...GridServiceProcessorBatchDeploySelfTest.java | 54 +-
...ServiceProcessorMultiNodeConfigSelfTest.java | 40 +-
.../GridServiceProcessorMultiNodeSelfTest.java | 25 +-
.../GridServiceProcessorProxySelfTest.java | 5 +-
.../GridServiceProcessorSingleNodeSelfTest.java | 20 +-
.../GridServiceProcessorStopSelfTest.java | 2 +-
.../GridServiceReassignmentSelfTest.java | 18 +-
.../GridServiceSerializationSelfTest.java | 2 +-
...gniteServiceConfigVariationsFullApiTest.java | 80 +-
.../IgniteServiceDynamicCachesSelfTest.java | 26 +-
.../service/IgniteServiceReassignmentTest.java | 50 +-
...tDiscoveryListenerNotificationOrderTest.java | 115 ++
...tNonSerializableStaticConfigurationTest.java | 119 ++
.../ServiceDeploymentOnActivationTest.java | 110 +-
...ServiceDeploymentOnClientDisconnectTest.java | 228 +++
.../ServiceDeploymentOutsideBaselineTest.java | 12 +-
.../ServiceDeploymentProcessAbstractTest.java | 127 ++
.../ServiceDeploymentProcessIdSelfTest.java | 105 ++
...ploymentProcessingOnCoordinatorFailTest.java | 34 +
...ploymentProcessingOnCoordinatorLeftTest.java | 125 ++
...viceDeploymentProcessingOnNodesFailTest.java | 34 +
...viceDeploymentProcessingOnNodesLeftTest.java | 117 ++
.../processors/service/ServiceInfoSelfTest.java | 137 ++
.../ServicePredicateAccessCacheTest.java | 52 +-
.../ServiceReassignmentFunctionSelfTest.java | 220 +++
.../inner/LongInitializedTestService.java | 52 +
.../junits/common/GridCommonAbstractTest.java | 33 +
.../testsuites/IgniteKernalSelfTestSuite.java | 25 +-
...ServiceConfigVariationsFullApiTestSuite.java | 9 -
.../mvcc/CacheMvccBasicContinuousQueryTest.java | 9 +-
.../Services/ServicesTest.cs | 16 +-
83 files changed, 6919 insertions(+), 450 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java b/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java
index 4cfb361..f58f1aa 100644
--- a/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java
+++ b/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java
@@ -1059,6 +1059,19 @@ public final class IgniteSystemProperties {
public static final String IGNITE_DEFAULT_DATA_STORAGE_PAGE_SIZE = "IGNITE_DEFAULT_DATA_STORAGE_PAGE_SIZE";
/**
+ * Manages the type of the implementation of the service processor (implementation of the {@link IgniteServices}).
+ * All nodes in the cluster must have the same value of this property.
+ * <p/>
+ * If the property is {@code true} then event-driven implementation of the service processor will be used.
+ * <p/>
+ * If the property is {@code false} then internal cache based implementation of service processor will be used.
+ * <p/>
+ * Default is {@code true}.
+ */
+ public static final String IGNITE_EVENT_DRIVEN_SERVICE_PROCESSOR_ENABLED
+ = "IGNITE_EVENT_DRIVEN_SERVICE_PROCESSOR_ENABLED";
+
+ /**
* When set to {@code true}, cache metrics are not included into the discovery metrics update message (in this
* case message contains only cluster metrics). By default cache metrics are included into the message and
* calculated each time the message is sent.
http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/main/java/org/apache/ignite/internal/GridComponent.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/GridComponent.java b/modules/core/src/main/java/org/apache/ignite/internal/GridComponent.java
index 607217e..1f9e02e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/GridComponent.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/GridComponent.java
@@ -70,7 +70,10 @@ public interface GridComponent {
CACHE_CRD_PROC,
/** Encryption manager. */
- ENCRYPTION_MGR
+ ENCRYPTION_MGR,
+
+ /** Service processor. */
+ SERVICE_PROC
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java b/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java
index fdb8ebc..691fe37 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java
@@ -36,6 +36,7 @@ import org.apache.ignite.internal.managers.loadbalancer.GridLoadBalancerManager;
import org.apache.ignite.internal.processors.cache.mvcc.MvccProcessor;
import org.apache.ignite.internal.stat.IoStatisticsManager;
import org.apache.ignite.internal.processors.compress.CompressionProcessor;
+import org.apache.ignite.internal.processors.service.ServiceProcessorAdapter;
import org.apache.ignite.internal.worker.WorkersRegistry;
import org.apache.ignite.internal.processors.affinity.GridAffinityProcessor;
import org.apache.ignite.internal.processors.authentication.IgniteAuthenticationProcessor;
@@ -67,7 +68,6 @@ import org.apache.ignite.internal.processors.rest.GridRestProcessor;
import org.apache.ignite.internal.processors.schedule.IgniteScheduleProcessorAdapter;
import org.apache.ignite.internal.processors.security.GridSecurityProcessor;
import org.apache.ignite.internal.processors.segmentation.GridSegmentationProcessor;
-import org.apache.ignite.internal.processors.service.GridServiceProcessor;
import org.apache.ignite.internal.processors.session.GridTaskSessionProcessor;
import org.apache.ignite.internal.processors.subscription.GridInternalSubscriptionProcessor;
import org.apache.ignite.internal.processors.task.GridTaskProcessor;
@@ -224,7 +224,7 @@ public interface GridKernalContext extends Iterable<GridComponent> {
*
* @return Service processor.
*/
- public GridServiceProcessor service();
+ public ServiceProcessorAdapter service();
/**
* Gets port processor.
http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java
index 8131899..1219d00 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java
@@ -48,6 +48,7 @@ import org.apache.ignite.internal.managers.eventstorage.GridEventStorageManager;
import org.apache.ignite.internal.managers.failover.GridFailoverManager;
import org.apache.ignite.internal.managers.indexing.GridIndexingManager;
import org.apache.ignite.internal.managers.loadbalancer.GridLoadBalancerManager;
+import org.apache.ignite.internal.processors.service.ServiceProcessorAdapter;
import org.apache.ignite.internal.processors.affinity.GridAffinityProcessor;
import org.apache.ignite.internal.processors.authentication.IgniteAuthenticationProcessor;
import org.apache.ignite.internal.processors.cache.CacheConflictResolutionManager;
@@ -84,7 +85,6 @@ import org.apache.ignite.internal.processors.rest.GridRestProcessor;
import org.apache.ignite.internal.processors.schedule.IgniteScheduleProcessorAdapter;
import org.apache.ignite.internal.processors.security.GridSecurityProcessor;
import org.apache.ignite.internal.processors.segmentation.GridSegmentationProcessor;
-import org.apache.ignite.internal.processors.service.GridServiceProcessor;
import org.apache.ignite.internal.processors.session.GridTaskSessionProcessor;
import org.apache.ignite.internal.processors.subscription.GridInternalSubscriptionProcessor;
import org.apache.ignite.internal.processors.task.GridTaskProcessor;
@@ -208,7 +208,7 @@ public class GridKernalContextImpl implements GridKernalContext, Externalizable
/** */
@GridToStringInclude
- private GridServiceProcessor svcProc;
+ private ServiceProcessorAdapter srvcProc;
/** */
@GridToStringInclude
@@ -608,8 +608,8 @@ public class GridKernalContextImpl implements GridKernalContext, Externalizable
portProc = (GridPortProcessor)comp;
else if (comp instanceof GridClosureProcessor)
closProc = (GridClosureProcessor)comp;
- else if (comp instanceof GridServiceProcessor)
- svcProc = (GridServiceProcessor)comp;
+ else if (comp instanceof ServiceProcessorAdapter)
+ srvcProc = (ServiceProcessorAdapter)comp;
else if (comp instanceof IgniteScheduleProcessorAdapter)
scheduleProc = (IgniteScheduleProcessorAdapter)comp;
else if (comp instanceof GridSegmentationProcessor)
@@ -762,8 +762,8 @@ public class GridKernalContextImpl implements GridKernalContext, Externalizable
}
/** {@inheritDoc} */
- @Override public GridServiceProcessor service() {
- return svcProc;
+ @Override public ServiceProcessorAdapter service() {
+ return srvcProc;
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/main/java/org/apache/ignite/internal/GridTopic.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/GridTopic.java b/modules/core/src/main/java/org/apache/ignite/internal/GridTopic.java
index 95d7717..437ee4d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/GridTopic.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/GridTopic.java
@@ -136,7 +136,10 @@ public enum GridTopic {
TOPIC_CACHE_COORDINATOR,
/** */
- TOPIC_GEN_ENC_KEY;
+ TOPIC_GEN_ENC_KEY,
+
+ /** */
+ TOPIC_SERVICES;
/** Enum values. */
private static final GridTopic[] VALS = values();
http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java b/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
index 95bf49c..b84771a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
@@ -116,6 +116,7 @@ import org.apache.ignite.internal.managers.indexing.GridIndexingManager;
import org.apache.ignite.internal.managers.loadbalancer.GridLoadBalancerManager;
import org.apache.ignite.internal.marshaller.optimized.OptimizedMarshaller;
import org.apache.ignite.internal.processors.GridProcessor;
+import org.apache.ignite.internal.processors.GridProcessorAdapter;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.affinity.GridAffinityProcessor;
import org.apache.ignite.internal.processors.authentication.IgniteAuthenticationProcessor;
@@ -163,6 +164,7 @@ import org.apache.ignite.internal.processors.rest.GridRestProcessor;
import org.apache.ignite.internal.processors.security.GridSecurityProcessor;
import org.apache.ignite.internal.processors.segmentation.GridSegmentationProcessor;
import org.apache.ignite.internal.processors.service.GridServiceProcessor;
+import org.apache.ignite.internal.processors.service.IgniteServiceProcessor;
import org.apache.ignite.internal.processors.session.GridTaskSessionProcessor;
import org.apache.ignite.internal.processors.subscription.GridInternalSubscriptionProcessor;
import org.apache.ignite.internal.processors.task.GridTaskProcessor;
@@ -224,6 +226,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_CONFIG_URL;
import static org.apache.ignite.IgniteSystemProperties.IGNITE_DAEMON;
+import static org.apache.ignite.IgniteSystemProperties.IGNITE_EVENT_DRIVEN_SERVICE_PROCESSOR_ENABLED;
import static org.apache.ignite.IgniteSystemProperties.IGNITE_NO_ASCII;
import static org.apache.ignite.IgniteSystemProperties.IGNITE_OPTIMIZED_MARSHALLER_USE_DEFAULT_SUID;
import static org.apache.ignite.IgniteSystemProperties.IGNITE_REST_START_ON_CLIENT;
@@ -251,6 +254,7 @@ import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_DATA_STORAGE_
import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_DATA_STREAMER_POOL_SIZE;
import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_DEPLOYMENT_MODE;
import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_DYNAMIC_CACHE_START_ROLLBACK_SUPPORTED;
+import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_EVENT_DRIVEN_SERVICE_PROCESSOR_ENABLED;
import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_IGNITE_FEATURES;
import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_IGNITE_INSTANCE_NAME;
import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_IPS;
@@ -1016,7 +1020,7 @@ public class IgniteKernal implements IgniteEx, IgniteMXBean, Externalizable {
startProcessor(new GridCacheProcessor(ctx));
startProcessor(new GridQueryProcessor(ctx));
startProcessor(new ClientListenerProcessor(ctx));
- startProcessor(new GridServiceProcessor(ctx));
+ startProcessor(createServiceProcessor());
startProcessor(new GridTaskSessionProcessor(ctx));
startProcessor(new GridJobProcessor(ctx));
startProcessor(new GridTaskProcessor(ctx));
@@ -1357,6 +1361,20 @@ public class IgniteKernal implements IgniteEx, IgniteMXBean, Externalizable {
}
/**
+ * Creates service processor depend on {@link IgniteSystemProperties#IGNITE_EVENT_DRIVEN_SERVICE_PROCESSOR_ENABLED}.
+ *
+ * @return Service processor.
+ */
+ private GridProcessorAdapter createServiceProcessor() {
+ final boolean srvcProcMode = getBoolean(IGNITE_EVENT_DRIVEN_SERVICE_PROCESSOR_ENABLED, true);
+
+ if (srvcProcMode)
+ return new IgniteServiceProcessor(ctx);
+
+ return new GridServiceProcessor(ctx);
+ }
+
+ /**
* Validates common configuration parameters.
*
* @param cfg Configuration.
@@ -1643,6 +1661,9 @@ public class IgniteKernal implements IgniteEx, IgniteMXBean, Externalizable {
ctx.addNodeAttribute(e.getKey(), e.getValue());
}
}
+
+ ctx.addNodeAttribute(ATTR_EVENT_DRIVEN_SERVICE_PROCESSOR_ENABLED,
+ ctx.service() instanceof IgniteServiceProcessor);
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/main/java/org/apache/ignite/internal/IgniteNodeAttributes.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgniteNodeAttributes.java b/modules/core/src/main/java/org/apache/ignite/internal/IgniteNodeAttributes.java
index 45ca234..1740072 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/IgniteNodeAttributes.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/IgniteNodeAttributes.java
@@ -208,6 +208,10 @@ public final class IgniteNodeAttributes {
/** Supported features. */
public static final String ATTR_IGNITE_FEATURES = ATTR_PREFIX + ".features";
+ /** Ignite services processor mode. */
+ public static final String ATTR_EVENT_DRIVEN_SERVICE_PROCESSOR_ENABLED = ATTR_PREFIX +
+ ".event.driven.service.processor.enabled";
+
/**
* Enforces singleton.
*/
http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
index b1c023a..5e7811b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
@@ -184,6 +184,9 @@ import org.apache.ignite.internal.processors.query.h2.twostep.messages.GridQuery
import org.apache.ignite.internal.processors.query.schema.message.SchemaOperationStatusMessage;
import org.apache.ignite.internal.processors.rest.handlers.task.GridTaskResultRequest;
import org.apache.ignite.internal.processors.rest.handlers.task.GridTaskResultResponse;
+import org.apache.ignite.internal.processors.service.ServiceSingleNodeDeploymentResult;
+import org.apache.ignite.internal.processors.service.ServiceDeploymentProcessId;
+import org.apache.ignite.internal.processors.service.ServiceSingleNodeDeploymentResultBatch;
import org.apache.ignite.internal.util.GridByteArrayList;
import org.apache.ignite.internal.util.GridIntList;
import org.apache.ignite.internal.util.GridLongList;
@@ -1120,6 +1123,21 @@ public class GridIoMessageFactory implements MessageFactory {
break;
+ case 167:
+ msg = new ServiceDeploymentProcessId();
+
+ break;
+
+ case 168:
+ msg = new ServiceSingleNodeDeploymentResultBatch();
+
+ break;
+
+ case 169:
+ msg = new ServiceSingleNodeDeploymentResult();
+
+ break;
+
// [-3..119] [124..129] [-23..-28] [-36..-55] - this
// [120..123] - DR
// [-4..-22, -30..-35] - SQL
http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
index 78319a0..556af19 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
@@ -143,6 +143,7 @@ import org.jetbrains.annotations.Nullable;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.apache.ignite.IgniteSystemProperties.IGNITE_BINARY_MARSHALLER_USE_STRING_SERIALIZATION_VER_2;
import static org.apache.ignite.IgniteSystemProperties.IGNITE_DISCOVERY_HISTORY_SIZE;
+import static org.apache.ignite.IgniteSystemProperties.IGNITE_EVENT_DRIVEN_SERVICE_PROCESSOR_ENABLED;
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.getInteger;
@@ -157,6 +158,7 @@ import static org.apache.ignite.failure.FailureType.CRITICAL_ERROR;
import static org.apache.ignite.failure.FailureType.SYSTEM_WORKER_TERMINATION;
import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_DATA_REGIONS_OFFHEAP_SIZE;
import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_DEPLOYMENT_MODE;
+import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_EVENT_DRIVEN_SERVICE_PROCESSOR_ENABLED;
import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_LATE_AFFINITY_ASSIGNMENT;
import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_MACS;
import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_MARSHALLER_USE_BINARY_STRING_SER_VER_2;
@@ -789,6 +791,8 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
ctx.cache().context().exchange().onLocalJoin(discoEvt, discoCache);
+ ctx.service().onLocalJoin(discoEvt, discoCache);
+
ctx.authentication().onLocalJoin();
ctx.encryption().onLocalJoin();
@@ -847,6 +851,8 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
ctx.cache().context().exchange().onLocalJoin(localJoinEvent(), discoCache);
+ ctx.service().onLocalJoin(localJoinEvent(), discoCache);
+
ctx.cluster().clientReconnectFuture().listen(new CI1<IgniteFuture<?>>() {
@Override public void apply(IgniteFuture<?> fut) {
try {
@@ -1223,6 +1229,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
boolean locDelayAssign = locNode.attribute(ATTR_LATE_AFFINITY_ASSIGNMENT);
+ Boolean locSrvcProcMode = locNode.attribute(ATTR_EVENT_DRIVEN_SERVICE_PROCESSOR_ENABLED);
Boolean locSecurityCompatibilityEnabled = locNode.attribute(ATTR_SECURITY_COMPATIBILITY_MODE);
for (ClusterNode n : nodes) {
@@ -1308,6 +1315,22 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
", rmtAddrs=" + U.addressesAsString(n) + ", rmtNode=" + U.toShortString(n) + "]");
}
+ Boolean rmtSrvcProcModeAttr = n.attribute(ATTR_EVENT_DRIVEN_SERVICE_PROCESSOR_ENABLED);
+
+ final boolean rmtSrvcProcMode = rmtSrvcProcModeAttr != null ? rmtSrvcProcModeAttr : false;
+
+ if (!F.eq(locSrvcProcMode, rmtSrvcProcMode)) {
+ throw new IgniteCheckedException("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(n) +
+ ", locNodeId=" + locNode.id() + ", rmtNode=" + U.toShortString(n) + "]");
+ }
+
if (n.version().compareToIgnoreTimestamp(SERVICE_PERMISSIONS_SINCE) >= 0
&& ctx.security().enabled() // Matters only if security enabled.
) {
http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/DynamicCacheChangeBatch.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/DynamicCacheChangeBatch.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/DynamicCacheChangeBatch.java
index d85e29b..9dc1195 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/DynamicCacheChangeBatch.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/DynamicCacheChangeBatch.java
@@ -23,6 +23,8 @@ import org.apache.ignite.internal.managers.discovery.DiscoCache;
import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
import org.apache.ignite.internal.managers.discovery.GridDiscoveryManager;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.service.ServiceDeploymentActions;
+import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.S;
@@ -52,6 +54,10 @@ public class DynamicCacheChangeBatch implements DiscoveryCustomMessage {
/** Restarting caches. */
private Set<String> restartingCaches;
+ /** Affinity (cache related) services updates to be processed on services deployment process. */
+ @GridToStringExclude
+ @Nullable private transient ServiceDeploymentActions serviceDeploymentActions;
+
/**
* @param reqs Requests.
*/
@@ -118,6 +124,20 @@ public class DynamicCacheChangeBatch implements DiscoveryCustomMessage {
}
/**
+ * @return Services deployment actions to be processed on services deployment process.
+ */
+ @Nullable public ServiceDeploymentActions servicesDeploymentActions() {
+ return serviceDeploymentActions;
+ }
+
+ /**
+ * @param serviceDeploymentActions Services deployment actions to be processed on services deployment process.
+ */
+ public void servicesDeploymentActions(ServiceDeploymentActions serviceDeploymentActions) {
+ this.serviceDeploymentActions = serviceDeploymentActions;
+ }
+
+ /**
* @return {@code True} if required to start all caches on client node.
*/
public boolean startCaches() {
http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
index b919024..1c44eaf 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
@@ -137,6 +137,7 @@ import org.apache.ignite.internal.processors.query.schema.SchemaNodeLeaveExchang
import org.apache.ignite.internal.processors.query.schema.message.SchemaAbstractDiscoveryMessage;
import org.apache.ignite.internal.processors.query.schema.message.SchemaProposeDiscoveryMessage;
import org.apache.ignite.internal.processors.security.SecurityContext;
+import org.apache.ignite.internal.processors.service.GridServiceProcessor;
import org.apache.ignite.internal.processors.timeout.GridTimeoutObject;
import org.apache.ignite.internal.suggestions.GridPerformanceSuggestions;
import org.apache.ignite.internal.util.F0;
@@ -1019,7 +1020,8 @@ public class GridCacheProcessor extends GridProcessorAdapter {
if (!active)
return;
- ctx.service().onUtilityCacheStarted();
+ if (ctx.service() instanceof GridServiceProcessor)
+ ((GridServiceProcessor)ctx.service()).onUtilityCacheStarted();
final AffinityTopologyVersion startTopVer = ctx.discovery().localJoin().joinTopologyVersion();
@@ -3047,7 +3049,8 @@ public class GridCacheProcessor extends GridProcessorAdapter {
if (exchActions.systemCachesStarting() && exchActions.stateChangeRequest() == null) {
ctx.dataStructures().restoreStructuresState(ctx);
- ctx.service().updateUtilityCache();
+ if (ctx.service() instanceof GridServiceProcessor)
+ ((GridServiceProcessor)ctx.service()).updateUtilityCache();
}
if (err == null)
http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
index dbc51f7..d4d89bc 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
@@ -99,6 +99,7 @@ import org.apache.ignite.internal.processors.cluster.BaselineTopology;
import org.apache.ignite.internal.processors.cluster.ChangeGlobalStateFinishMessage;
import org.apache.ignite.internal.processors.cluster.ChangeGlobalStateMessage;
import org.apache.ignite.internal.processors.cluster.DiscoveryDataClusterState;
+import org.apache.ignite.internal.processors.service.GridServiceProcessor;
import org.apache.ignite.internal.util.IgniteUtils;
import org.apache.ignite.internal.util.TimeBag;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
@@ -1136,7 +1137,8 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
try {
cctx.kernalContext().dataStructures().onDeActivate(cctx.kernalContext());
- cctx.kernalContext().service().onDeActivate(cctx.kernalContext());
+ if (cctx.kernalContext().service() instanceof GridServiceProcessor)
+ ((GridServiceProcessor)cctx.kernalContext().service()).onDeActivate(cctx.kernalContext());
assert registerCachesFuture == null : "No caches registration should be scheduled before new caches have started.";
http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/ChangeGlobalStateMessage.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/ChangeGlobalStateMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/ChangeGlobalStateMessage.java
index 51a65bb..b1ff048 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/ChangeGlobalStateMessage.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/ChangeGlobalStateMessage.java
@@ -25,6 +25,7 @@ import org.apache.ignite.internal.managers.discovery.GridDiscoveryManager;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.ExchangeActions;
import org.apache.ignite.internal.processors.cache.StoredCacheData;
+import org.apache.ignite.internal.processors.service.ServiceDeploymentActions;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.lang.IgniteUuid;
@@ -65,6 +66,10 @@ public class ChangeGlobalStateMessage implements DiscoveryCustomMessage {
@GridToStringExclude
private transient ExchangeActions exchangeActions;
+ /** Services deployment actions to be processed on services deployment process. */
+ @GridToStringExclude
+ @Nullable private transient ServiceDeploymentActions serviceDeploymentActions;
+
/**
* @param reqId State change request ID.
* @param initiatingNodeId Node initiated state change.
@@ -117,6 +122,20 @@ public class ChangeGlobalStateMessage implements DiscoveryCustomMessage {
this.exchangeActions = exchangeActions;
}
+ /**
+ * @return Services deployment actions to be processed on services deployment process.
+ */
+ @Nullable public ServiceDeploymentActions servicesDeploymentActions() {
+ return serviceDeploymentActions;
+ }
+
+ /**
+ * @param serviceDeploymentActions Services deployment actions to be processed on services deployment process.
+ */
+ public void servicesDeploymentActions(ServiceDeploymentActions serviceDeploymentActions) {
+ this.serviceDeploymentActions = serviceDeploymentActions;
+ }
+
/** {@inheritDoc} */
@Override public IgniteUuid id() {
return id;
http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/GridClusterStateProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/GridClusterStateProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/GridClusterStateProcessor.java
index 9d2adae..b347d39 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/GridClusterStateProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/GridClusterStateProcessor.java
@@ -56,6 +56,7 @@ import org.apache.ignite.internal.processors.cache.StoredCacheData;
import org.apache.ignite.internal.processors.cache.persistence.metastorage.MetastorageLifecycleListener;
import org.apache.ignite.internal.processors.cache.persistence.metastorage.ReadOnlyMetastorage;
import org.apache.ignite.internal.processors.cache.persistence.metastorage.ReadWriteMetastorage;
+import org.apache.ignite.internal.processors.service.GridServiceProcessor;
import org.apache.ignite.internal.processors.task.GridInternal;
import org.apache.ignite.internal.util.future.GridFinishedFuture;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
@@ -1165,9 +1166,13 @@ public class GridClusterStateProcessor extends GridProcessorAdapter implements I
Exception e = null;
try {
- ctx.service().onUtilityCacheStarted();
+ if (ctx.service() instanceof GridServiceProcessor) {
+ GridServiceProcessor srvcProc = (GridServiceProcessor)ctx.service();
- ctx.service().onActivate(ctx);
+ srvcProc.onUtilityCacheStarted();
+
+ srvcProc.onActivate(ctx);
+ }
ctx.dataStructures().onActivate(ctx);
http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java
index 22d6997..2456efa 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java
@@ -69,6 +69,7 @@ import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheProcessor;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.CachePartitionPartialCountersMap;
import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryHandler;
+import org.apache.ignite.internal.processors.service.GridServiceProcessor;
import org.apache.ignite.internal.processors.timeout.GridTimeoutObject;
import org.apache.ignite.internal.util.future.GridFinishedFuture;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
@@ -295,7 +296,8 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
ctx.cacheObjects().onContinuousProcessorStarted(ctx);
- ctx.service().onContinuousProcessorStarted(ctx);
+ if (ctx.service() instanceof GridServiceProcessor)
+ ((GridServiceProcessor)ctx.service()).onContinuousProcessorStarted(ctx);
if (log.isDebugEnabled())
log.debug("Continuous processor started.");
http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceAssignments.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceAssignments.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceAssignments.java
index 66b2816..290fbf4 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceAssignments.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceAssignments.java
@@ -30,7 +30,10 @@ import org.apache.ignite.services.ServiceConfiguration;
/**
* Service per-node assignment.
+ *
+ * @deprecated Services internals use messages for deployment management instead of the utility cache, since Ignite 2.8.
*/
+@Deprecated
public class GridServiceAssignments implements Serializable, GridCacheInternal {
/** Serialization version. */
private static final long serialVersionUID = 0L;
http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceAssignmentsKey.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceAssignmentsKey.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceAssignmentsKey.java
index 5b26f9f..87cb0ba 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceAssignmentsKey.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceAssignmentsKey.java
@@ -22,7 +22,10 @@ import org.apache.ignite.internal.util.typedef.internal.S;
/**
* Service configuration key.
+ *
+ * @deprecated Services internals use messages for deployment management instead of the utility cache, since Ignite 2.8.
*/
+@Deprecated
public class GridServiceAssignmentsKey extends GridCacheUtilityKey<GridServiceAssignmentsKey> {
/** */
private static final long serialVersionUID = 0L;
http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceDeployment.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceDeployment.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceDeployment.java
index f25c38e..ae8670f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceDeployment.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceDeployment.java
@@ -25,7 +25,10 @@ import org.apache.ignite.services.ServiceConfiguration;
/**
* Service deployment.
+ *
+ * @deprecated Services internals use messages for deployment management instead of the utility cache, since Ignite 2.8.
*/
+@Deprecated
public class GridServiceDeployment implements GridCacheInternal, Serializable {
/** */
private static final long serialVersionUID = 0L;
http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceDeploymentCompoundFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceDeploymentCompoundFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceDeploymentCompoundFuture.java
index 45ccc24..f776285 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceDeploymentCompoundFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceDeploymentCompoundFuture.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.processors.service;
+import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -33,9 +34,9 @@ import org.jetbrains.annotations.Nullable;
* IgniteInternalFuture#get get()} method after all futures complete or fail. Inner exception will contain
* configurations of failed services.
*/
-public class GridServiceDeploymentCompoundFuture extends GridCompoundFuture<Object, Object> {
- /** Names of services written to cache during current deployment. */
- private Collection<String> svcsToRollback;
+public class GridServiceDeploymentCompoundFuture<T extends Serializable> extends GridCompoundFuture<Object, Object> {
+ /** Ids of services written to cache during current deployment. */
+ private Collection<T> svcsToRollback;
/** */
private volatile ServiceDeploymentException err;
@@ -71,21 +72,21 @@ public class GridServiceDeploymentCompoundFuture extends GridCompoundFuture<Obje
* @param fut Child future.
* @param own If {@code true}, then corresponding service will be cancelled on failure.
*/
- public void add(GridServiceDeploymentFuture fut, boolean own) {
+ public void add(GridServiceDeploymentFuture<T> fut, boolean own) {
super.add(fut);
if (own) {
if (svcsToRollback == null)
svcsToRollback = new ArrayList<>();
- svcsToRollback.add(fut.configuration().getName());
+ svcsToRollback.add(fut.serviceId());
}
}
/**
- * @return Collection of names of services that were written to cache during current deployment.
+ * @return Collection of ids of services that were written to cache during current deployment.
*/
- public Collection<String> servicesToRollback() {
+ public Collection<T> servicesToRollback() {
if (svcsToRollback != null)
return svcsToRollback;
else
http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceDeploymentFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceDeploymentFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceDeploymentFuture.java
index c87cc6e..ba4422e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceDeploymentFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceDeploymentFuture.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.processors.service;
+import java.io.Serializable;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.services.ServiceConfiguration;
@@ -24,15 +25,27 @@ import org.apache.ignite.services.ServiceConfiguration;
/**
* Service deployment future.
*/
-public class GridServiceDeploymentFuture extends GridFutureAdapter<Object> {
+public class GridServiceDeploymentFuture<T extends Serializable> extends GridFutureAdapter<Object> {
/** */
private final ServiceConfiguration cfg;
+ /** */
+ private final T srvcId;
+
/**
* @param cfg Configuration.
*/
public GridServiceDeploymentFuture(ServiceConfiguration cfg) {
+ this(cfg, null);
+ }
+
+ /**
+ * @param cfg Configuration.
+ * @param srvcId Service id.
+ */
+ public GridServiceDeploymentFuture(ServiceConfiguration cfg, T srvcId) {
this.cfg = cfg;
+ this.srvcId = srvcId;
}
/**
@@ -42,8 +55,15 @@ public class GridServiceDeploymentFuture extends GridFutureAdapter<Object> {
return cfg;
}
+ /**
+ * @return Service id.
+ */
+ public T serviceId() {
+ return srvcId;
+ }
+
/** {@inheritDoc} */
@Override public String toString() {
return S.toString(GridServiceDeploymentFuture.class, this);
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceDeploymentKey.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceDeploymentKey.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceDeploymentKey.java
index 80415bb..e5b85dc 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceDeploymentKey.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceDeploymentKey.java
@@ -22,7 +22,10 @@ import org.apache.ignite.internal.util.typedef.internal.S;
/**
* Service configuration key.
+ *
+ * @deprecated Services internals use messages for deployment management instead of the utility cache, since Ignite 2.8.
*/
+@Deprecated
public class GridServiceDeploymentKey extends GridCacheUtilityKey<GridServiceDeploymentKey> {
/** */
private static final long serialVersionUID = 0L;
http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceProcessor.java
index 3adc282..8fcbe4a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceProcessor.java
@@ -61,7 +61,6 @@ import org.apache.ignite.internal.events.DiscoveryCustomEvent;
import org.apache.ignite.internal.managers.discovery.DiscoCache;
import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
import org.apache.ignite.internal.managers.eventstorage.DiscoveryEventListener;
-import org.apache.ignite.internal.processors.GridProcessorAdapter;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.CacheAffinityChangeMessage;
import org.apache.ignite.internal.processors.cache.CacheIteratorConverter;
@@ -118,9 +117,18 @@ import static org.apache.ignite.transactions.TransactionIsolation.REPEATABLE_REA
/**
* Grid service processor.
+ * <p/>
+ * Obsolete implementation of service processor, based on replicated system cache.
+ * <p/>
+ * NOTE: if you fix a bug in this class, please take a look in {@link IgniteServiceProcessor}, perhaps the class
+ * contains a similar block of code which also should be updated.
+ *
+ * @see IgniteServiceProcessor
+ * @deprecated Here is improved, but uncompatible implementation {@link IgniteServiceProcessor}, see IEP-17 for details.
*/
+@Deprecated
@SuppressWarnings({"SynchronizationOnLocalVariableOrMethodParameter", "ConstantConditions"})
-public class GridServiceProcessor extends GridProcessorAdapter implements IgniteChangeGlobalStateSupport {
+public class GridServiceProcessor extends ServiceProcessorAdapter implements IgniteChangeGlobalStateSupport {
/** Time to wait before reassignment retries. */
private static final long RETRY_TIMEOUT = 1000;
@@ -136,7 +144,7 @@ public class GridServiceProcessor extends GridProcessorAdapter implements Ignite
private final Map<String, Collection<ServiceContextImpl>> locSvcs = new HashMap<>();
/** Deployment futures. */
- private final ConcurrentMap<String, GridServiceDeploymentFuture> depFuts = new ConcurrentHashMap<>();
+ private final ConcurrentMap<String, GridServiceDeploymentFuture<String>> depFuts = new ConcurrentHashMap<>();
/** Deployment futures. */
private final ConcurrentMap<String, GridFutureAdapter<?>> undepFuts = new ConcurrentHashMap<>();
@@ -456,58 +464,38 @@ public class GridServiceProcessor extends GridProcessorAdapter implements Ignite
throw new IgniteException("Service configuration check failed (" + desc + ")");
}
- /**
- * @param name Service name.
- * @param svc Service.
- * @return Future.
- */
- public IgniteInternalFuture<?> deployNodeSingleton(ClusterGroup prj, String name, Service svc) {
- return deployMultiple(prj, name, svc, 0, 1);
+ /** {@inheritDoc} */
+ @Override public IgniteInternalFuture<?> deployNodeSingleton(ClusterGroup prj, String name, Service srvc) {
+ return deployMultiple(prj, name, srvc, 0, 1);
}
- /**
- * @param name Service name.
- * @param svc Service.
- * @return Future.
- */
- public IgniteInternalFuture<?> deployClusterSingleton(ClusterGroup prj, String name, Service svc) {
- return deployMultiple(prj, name, svc, 1, 1);
+ /** {@inheritDoc} */
+ @Override public IgniteInternalFuture<?> deployClusterSingleton(ClusterGroup prj, String name, Service srvc) {
+ return deployMultiple(prj, name, srvc, 1, 1);
}
- /**
- * @param name Service name.
- * @param svc Service.
- * @param totalCnt Total count.
- * @param maxPerNodeCnt Max per-node count.
- * @return Future.
- */
- public IgniteInternalFuture<?> deployMultiple(ClusterGroup prj, String name, Service svc, int totalCnt,
+ /** {@inheritDoc} */
+ @Override public IgniteInternalFuture<?> deployMultiple(ClusterGroup prj, String name, Service srvc, int totalCnt,
int maxPerNodeCnt) {
ServiceConfiguration cfg = new ServiceConfiguration();
cfg.setName(name);
- cfg.setService(svc);
+ cfg.setService(srvc);
cfg.setTotalCount(totalCnt);
cfg.setMaxPerNodeCount(maxPerNodeCnt);
return deployAll(prj, Collections.singleton(cfg));
}
- /**
- * @param name Service name.
- * @param svc Service.
- * @param cacheName Cache name.
- * @param affKey Affinity key.
- * @return Future.
- */
- public IgniteInternalFuture<?> deployKeyAffinitySingleton(String name, Service svc, String cacheName,
+ /** {@inheritDoc} */
+ @Override public IgniteInternalFuture<?> deployKeyAffinitySingleton(String name, Service srvc, String cacheName,
Object affKey) {
A.notNull(affKey, "affKey");
ServiceConfiguration cfg = new ServiceConfiguration();
cfg.setName(name);
- cfg.setService(svc);
+ cfg.setService(srvc);
cfg.setCacheName(cacheName);
cfg.setAffinityKey(affKey);
cfg.setTotalCount(1);
@@ -522,13 +510,13 @@ public class GridServiceProcessor extends GridProcessorAdapter implements Ignite
* @param dfltNodeFilter Default NodeFilter.
* @return Configurations to deploy.
*/
- private PreparedConfigurations prepareServiceConfigurations(Collection<ServiceConfiguration> cfgs,
+ private PreparedConfigurations<String> prepareServiceConfigurations(Collection<ServiceConfiguration> cfgs,
IgnitePredicate<ClusterNode> dfltNodeFilter) {
List<ServiceConfiguration> cfgsCp = new ArrayList<>(cfgs.size());
Marshaller marsh = ctx.config().getMarshaller();
- List<GridServiceDeploymentFuture> failedFuts = null;
+ List<GridServiceDeploymentFuture<String>> failedFuts = null;
for (ServiceConfiguration cfg : cfgs) {
Exception err = null;
@@ -578,7 +566,7 @@ public class GridServiceProcessor extends GridProcessorAdapter implements Ignite
if (failedFuts == null)
failedFuts = new ArrayList<>();
- GridServiceDeploymentFuture fut = new GridServiceDeploymentFuture(cfg);
+ GridServiceDeploymentFuture<String> fut = new GridServiceDeploymentFuture<>(cfg);
fut.onDone(err);
@@ -586,15 +574,11 @@ public class GridServiceProcessor extends GridProcessorAdapter implements Ignite
}
}
- return new PreparedConfigurations(cfgsCp, failedFuts);
+ return new PreparedConfigurations<>(cfgsCp, failedFuts);
}
- /**
- * @param prj Grid projection.
- * @param cfgs Service configurations.
- * @return Future for deployment.
- */
- public IgniteInternalFuture<?> deployAll(ClusterGroup prj, Collection<ServiceConfiguration> cfgs) {
+ /** {@inheritDoc} */
+ @Override public IgniteInternalFuture<?> deployAll(ClusterGroup prj, Collection<ServiceConfiguration> cfgs) {
if (prj == null)
// Deploy to servers by default if no projection specified.
return deployAll(cfgs, ctx.cluster().get().forServers().predicate());
@@ -614,11 +598,11 @@ public class GridServiceProcessor extends GridProcessorAdapter implements Ignite
@Nullable IgnitePredicate<ClusterNode> dfltNodeFilter) {
assert cfgs != null;
- PreparedConfigurations srvCfg = prepareServiceConfigurations(cfgs, dfltNodeFilter);
+ PreparedConfigurations<String> srvCfg = prepareServiceConfigurations(cfgs, dfltNodeFilter);
List<ServiceConfiguration> cfgsCp = srvCfg.cfgs;
- List<GridServiceDeploymentFuture> failedFuts = srvCfg.failedFuts;
+ List<GridServiceDeploymentFuture<String>> failedFuts = srvCfg.failedFuts;
Collections.sort(cfgsCp, new Comparator<ServiceConfiguration>() {
@Override public int compare(ServiceConfiguration cfg1, ServiceConfiguration cfg2) {
@@ -626,10 +610,10 @@ public class GridServiceProcessor extends GridProcessorAdapter implements Ignite
}
});
- GridServiceDeploymentCompoundFuture res;
+ GridServiceDeploymentCompoundFuture<String> res;
while (true) {
- res = new GridServiceDeploymentCompoundFuture();
+ res = new GridServiceDeploymentCompoundFuture<>();
if (ctx.deploy().enabled())
ctx.cache().context().deploy().ignoreOwnership(true);
@@ -684,7 +668,7 @@ public class GridServiceProcessor extends GridProcessorAdapter implements Ignite
"Failed to deploy services, client node disconnected: " + cfgs);
for (String name : res.servicesToRollback()) {
- GridServiceDeploymentFuture fut = depFuts.remove(name);
+ GridServiceDeploymentFuture<String> fut = depFuts.remove(name);
if (fut != null)
fut.onDone(err);
@@ -694,7 +678,7 @@ public class GridServiceProcessor extends GridProcessorAdapter implements Ignite
}
if (failedFuts != null) {
- for (GridServiceDeploymentFuture fut : failedFuts)
+ for (GridServiceDeploymentFuture<String> fut : failedFuts)
res.add(fut, false);
}
@@ -708,13 +692,13 @@ public class GridServiceProcessor extends GridProcessorAdapter implements Ignite
* @param cfg Service configuration.
* @throws IgniteCheckedException If operation failed.
*/
- private void writeServiceToCache(GridServiceDeploymentCompoundFuture res, ServiceConfiguration cfg)
+ private void writeServiceToCache(GridServiceDeploymentCompoundFuture<String> res, ServiceConfiguration cfg)
throws IgniteCheckedException {
String name = cfg.getName();
- GridServiceDeploymentFuture fut = new GridServiceDeploymentFuture(cfg);
+ GridServiceDeploymentFuture<String> fut = new GridServiceDeploymentFuture<>(cfg);
- GridServiceDeploymentFuture old = depFuts.putIfAbsent(name, fut);
+ GridServiceDeploymentFuture<String> old = depFuts.putIfAbsent(name, fut);
try {
if (old != null) {
@@ -772,11 +756,8 @@ public class GridServiceProcessor extends GridProcessorAdapter implements Ignite
}
}
- /**
- * @param name Service name.
- * @return Future.
- */
- public IgniteInternalFuture<?> cancel(String name) {
+ /** {@inheritDoc} */
+ @Override public IgniteInternalFuture<?> cancel(String name) {
while (true) {
try {
return removeServiceFromCache(name).fut;
@@ -795,10 +776,8 @@ public class GridServiceProcessor extends GridProcessorAdapter implements Ignite
}
}
- /**
- * @return Future.
- */
- public IgniteInternalFuture<?> cancelAll() {
+ /** {@inheritDoc} */
+ @Override public IgniteInternalFuture<?> cancelAll() {
Iterator<Cache.Entry<Object, Object>> it = serviceEntries(ServiceDeploymentPredicate.INSTANCE);
List<String> svcNames = new ArrayList<>();
@@ -812,13 +791,10 @@ public class GridServiceProcessor extends GridProcessorAdapter implements Ignite
return cancelAll(svcNames);
}
- /**
- * @param svcNames Name of service to deploy.
- * @return Future.
- */
+ /** {@inheritDoc} */
@SuppressWarnings("unchecked")
- public IgniteInternalFuture<?> cancelAll(Collection<String> svcNames) {
- List<String> svcNamesCp = new ArrayList<>(svcNames);
+ @Override public IgniteInternalFuture<?> cancelAll(Collection<String> servicesNames) {
+ List<String> svcNamesCp = new ArrayList<>(servicesNames);
Collections.sort(svcNamesCp);
@@ -830,7 +806,7 @@ public class GridServiceProcessor extends GridProcessorAdapter implements Ignite
List<String> toRollback = new ArrayList<>();
try (Transaction tx = serviceCache().txStart(PESSIMISTIC, READ_COMMITTED)) {
- for (String name : svcNames) {
+ for (String name : servicesNames) {
if (res == null)
res = new GridCompoundFuture<>();
@@ -923,18 +899,13 @@ public class GridServiceProcessor extends GridProcessorAdapter implements Ignite
}
}
- /**
- * @param name Service name.
- * @param timeout If greater than 0 limits task execution time. Cannot be negative.
- * @return Service topology.
- * @throws IgniteCheckedException On error.
- */
- public Map<UUID, Integer> serviceTopology(String name, long timeout) throws IgniteCheckedException {
+ /** {@inheritDoc} */
+ @Override public Map<UUID, Integer> serviceTopology(String name, long timeout) throws IgniteCheckedException {
IgniteInternalCache<Object, Object> cache = serviceCache();
ClusterNode node = cache.affinity().mapKeyToNode(name);
- final ServiceTopologyCallable call = new ServiceTopologyCallable(name);
+ final ServiceTopologyCallable call = new ServiceTopologyCallable(name, pendingJobCtxs);
return ctx.closure().callAsyncNoFailover(
GridClosureCallMode.BROADCAST,
@@ -958,10 +929,8 @@ public class GridServiceProcessor extends GridProcessorAdapter implements Ignite
return val != null ? val.assigns() : null;
}
- /**
- * @return Collection of service descriptors.
- */
- public Collection<ServiceDescriptor> serviceDescriptors() {
+ /** {@inheritDoc} */
+ @Override public Collection<ServiceDescriptor> serviceDescriptors() {
Collection<ServiceDescriptor> descs = new ArrayList<>();
Iterator<Cache.Entry<Object, Object>> it = serviceEntries(ServiceDeploymentPredicate.INSTANCE);
@@ -992,12 +961,8 @@ public class GridServiceProcessor extends GridProcessorAdapter implements Ignite
return descs;
}
- /**
- * @param name Service name.
- * @param <T> Service type.
- * @return Service by specified service name.
- */
- public <T> T service(String name) {
+ /** {@inheritDoc} */
+ @Override public <T> T service(String name) {
ctx.security().authorize(name, SecurityPermission.SERVICE_INVOKE, null);
Collection<ServiceContextImpl> ctxs;
@@ -1024,11 +989,8 @@ public class GridServiceProcessor extends GridProcessorAdapter implements Ignite
}
}
- /**
- * @param name Service name.
- * @return Service by specified service name.
- */
- public ServiceContextImpl serviceContext(String name) {
+ /** {@inheritDoc} */
+ @Override public ServiceContextImpl serviceContext(String name) {
Collection<ServiceContextImpl> ctxs;
synchronized (locSvcs) {
@@ -1051,17 +1013,9 @@ public class GridServiceProcessor extends GridProcessorAdapter implements Ignite
}
}
- /**
- * @param prj Grid projection.
- * @param name Service name.
- * @param svcItf Service class.
- * @param sticky Whether multi-node request should be done.
- * @param timeout If greater than 0 limits service acquire time. Cannot be negative.
- * @param <T> Service interface type.
- * @return The proxy of a service by its name and class.
- * @throws IgniteException If failed to create proxy.
- */
- public <T> T serviceProxy(ClusterGroup prj, String name, Class<? super T> svcItf, boolean sticky, long timeout)
+ /** {@inheritDoc} */
+ @Override public <T> T serviceProxy(ClusterGroup prj, String name, Class<? super T> srvcCls, boolean sticky,
+ long timeout)
throws IgniteException {
ctx.security().authorize(name, SecurityPermission.SERVICE_INVOKE, null);
@@ -1072,16 +1026,16 @@ public class GridServiceProcessor extends GridProcessorAdapter implements Ignite
Service svc = ctx.service();
if (svc != null) {
- if (!svcItf.isAssignableFrom(svc.getClass()))
+ if (!srvcCls.isAssignableFrom(svc.getClass()))
throw new IgniteException("Service does not implement specified interface [svcItf=" +
- svcItf.getName() + ", svcCls=" + svc.getClass().getName() + ']');
+ srvcCls.getName() + ", svcCls=" + svc.getClass().getName() + ']');
return (T)svc;
}
}
}
- return new GridServiceProxy<T>(prj, name, svcItf, sticky, timeout, ctx).proxy();
+ return new GridServiceProxy<T>(prj, name, srvcCls, sticky, timeout, ctx).proxy();
}
/**
@@ -1097,12 +1051,8 @@ public class GridServiceProcessor extends GridProcessorAdapter implements Ignite
return false;
}
- /**
- * @param name Service name.
- * @param <T> Service type.
- * @return Services by specified service name.
- */
- public <T> Collection<T> services(String name) {
+ /** {@inheritDoc} */
+ @Override public <T> Collection<T> services(String name) {
ctx.security().authorize(name, SecurityPermission.SERVICE_INVOKE, null);
Collection<ServiceContextImpl> ctxs;
@@ -1973,7 +1923,7 @@ public class GridServiceProcessor extends GridProcessorAdapter implements Ignite
t = th;
}
- GridServiceDeploymentFuture fut = depFuts.get(assigns.name());
+ GridServiceDeploymentFuture<String> fut = depFuts.get(assigns.name());
if (fut != null && fut.configuration().equalsIgnoreNodeFilter(assigns.configuration())) {
depFuts.remove(assigns.name(), fut);
@@ -2129,11 +2079,16 @@ public class GridServiceProcessor extends GridProcessorAdapter implements Ignite
@LoggerResource
private transient IgniteLogger log;
+ /** */
+ private final List<ComputeJobContext> pendingCtxs;
+
/**
* @param svcName Service name.
+ * @param pendingCtxs Pending compute job contexts that waiting for utility cache initialization.
*/
- public ServiceTopologyCallable(String svcName) {
+ public ServiceTopologyCallable(String svcName, List<ComputeJobContext> pendingCtxs) {
this.svcName = svcName;
+ this.pendingCtxs = pendingCtxs;
}
/** {@inheritDoc} */
@@ -2141,8 +2096,6 @@ public class GridServiceProcessor extends GridProcessorAdapter implements Ignite
IgniteInternalCache<Object, Object> cache = ignite.context().cache().utilityCache();
if (cache == null) {
- List<ComputeJobContext> pendingCtxs = ignite.context().service().pendingJobCtxs;
-
synchronized (pendingCtxs) {
// Double check cache reference after lock acqusition.
cache = ignite.context().cache().utilityCache();
http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceProxy.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceProxy.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceProxy.java
index 4d98187..aa6ce44 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceProxy.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceProxy.java
@@ -197,19 +197,15 @@ public class GridServiceProxy<T> implements Serializable {
true).get();
}
}
- catch (GridServiceNotFoundException | ClusterTopologyCheckedException e) {
- if (log.isDebugEnabled())
- log.debug("Service was not found or topology changed (will retry): " + e.getMessage());
- }
catch (RuntimeException | Error e) {
throw e;
}
catch (IgniteCheckedException e) {
// Check if ignorable exceptions are in the cause chain.
- Throwable ignorableCause = X.cause(e, GridServiceNotFoundException.class);
+ Throwable ignorableCause = X.cause(e, ClusterTopologyCheckedException.class);
- if (ignorableCause == null)
- ignorableCause = X.cause(e, ClusterTopologyCheckedException.class);
+ if (ignorableCause == null && ctx.service() instanceof GridServiceProcessor)
+ ignorableCause = X.cause(e, GridServiceNotFoundException.class);
if (ignorableCause != null) {
if (log.isDebugEnabled())