IGNITE-5392 - Joining node must accept cluster active status

modules/core/src/main/java/org/apache/ignite/internal/GridComponent.java

@@ -69,7 +69,7 @@ enum DiscoveryDataExchangeType {
      *
      * @throws IgniteCheckedException Throws in case of any errors.
      */
-    public void start(boolean activeOnStart) throws IgniteCheckedException;
+    public void start() throws IgniteCheckedException;
 
     /**
      * Stops grid component.
@@ -86,7 +86,7 @@ enum DiscoveryDataExchangeType {
      *
      * @throws IgniteCheckedException Thrown in case of any errors.
      */
-    public void onKernalStart(boolean activeOnStart) throws IgniteCheckedException;
+    public void onKernalStart() throws IgniteCheckedException;
 
     /**
      * Callback to notify that kernal is about to stop.

modules/core/src/main/java/org/apache/ignite/internal/GridPluginComponent.java

@@ -51,7 +51,7 @@ public PluginProvider plugin() {
 
     /** {@inheritDoc} */
     @SuppressWarnings("unchecked")
-    @Override public void start(boolean activeOnStart) throws IgniteCheckedException {
+    @Override public void start() throws IgniteCheckedException {
         throw new UnsupportedOperationException();
     }
 
@@ -61,7 +61,7 @@ public PluginProvider plugin() {
     }
 
     /** {@inheritDoc} */
-    @Override public void onKernalStart(boolean activeOnStart) throws IgniteCheckedException {
+    @Override public void onKernalStart() throws IgniteCheckedException {
         plugin.onIgniteStart();
     }
 

modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java

@@ -926,7 +926,8 @@ public void start(
                 startProcessor(new  GridAffinityProcessor(ctx));
                 startProcessor(createComponent(GridSegmentationProcessor.class, ctx));
                 startProcessor(createComponent(IgniteCacheObjectProcessor.class, ctx));
-                startProcessor(new GridCacheProcessor(ctx));startProcessor(new GridClusterStateProcessor(ctx));
+                startProcessor(new GridClusterStateProcessor(ctx));
+                startProcessor(new GridCacheProcessor(ctx));
                 startProcessor(new GridQueryProcessor(ctx));
                 startProcessor(new SqlListenerProcessor(ctx));
                 startProcessor(new GridServiceProcessor(ctx));
@@ -991,10 +992,10 @@ public void start(
             ctx.performance().addAll(OsConfigurationSuggestions.getSuggestions());
 
             // Notify discovery manager the first to make sure that topology is discovered.
-            ctx.discovery().onKernalStart(activeOnStart);
+            ctx.discovery().onKernalStart();
 
             // Notify IO manager the second so further components can send and receive messages.
-            ctx.io().onKernalStart(activeOnStart);
+            ctx.io().onKernalStart();
 
             // Start plugins.
             for (PluginProvider provider : ctx.plugins().allProviders())
@@ -1017,7 +1018,7 @@ public void start(
 
                 if (!skipDaemon(comp)) {
                     try {
-                        comp.onKernalStart(activeOnStart);
+                        comp.onKernalStart();
                     }
                     catch (IgniteNeedReconnectException e) {
                         assert ctx.discovery().reconnectSupported();
@@ -1749,7 +1750,7 @@ private void startManager(GridManager mgr) throws IgniteCheckedException {
 
         try {
             if (!skipDaemon(mgr))
-                mgr.start(cfg.isActiveOnStart());
+                mgr.start();
         }
         catch (IgniteCheckedException e) {
             U.error(log, "Failed to start manager: " + mgr, e);
@@ -1767,7 +1768,7 @@ private void startProcessor(GridProcessor proc) throws IgniteCheckedException {
 
         try {
             if (!skipDaemon(proc))
-                proc.start(cfg.isActiveOnStart());
+                proc.start();
         }
         catch (IgniteCheckedException e) {
             throw new IgniteCheckedException("Failed to start processor: " + proc, e);

modules/core/src/main/java/org/apache/ignite/internal/managers/GridManagerAdapter.java

@@ -362,7 +362,7 @@ protected final String stopInfo() {
     }
 
     /** {@inheritDoc} */
-    @Override public final void onKernalStart(boolean activeOnStart) throws IgniteCheckedException {
+    @Override public final void onKernalStart() throws IgniteCheckedException {
         for (final IgniteSpi spi : spis) {
             try {
                 spi.onContextInitialized(new IgniteSpiContext() {

modules/core/src/main/java/org/apache/ignite/internal/managers/checkpoint/GridCheckpointManager.java

@@ -102,7 +102,7 @@ public GridCheckpointManager(GridKernalContext ctx) {
     }
 
     /** {@inheritDoc} */
-    @Override public void start(boolean activeOnStart) throws IgniteCheckedException {
+    @Override public void start() throws IgniteCheckedException {
         for (CheckpointSpi spi : getSpis()) {
             spi.setCheckpointListener(new CheckpointListener() {
                 @Override public void onCheckpointRemoved(String key) {

modules/core/src/main/java/org/apache/ignite/internal/managers/collision/GridCollisionManager.java

@@ -47,7 +47,7 @@ public GridCollisionManager(GridKernalContext ctx) {
     }
 
     /** {@inheritDoc} */
-    @Override public void start(boolean activeOnStart) throws IgniteCheckedException {
+    @Override public void start() throws IgniteCheckedException {
         startSpi();
 
         if (enabled()) {

modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java

@@ -261,7 +261,7 @@ public void resetMetrics() {
 
     /** {@inheritDoc} */
     @SuppressWarnings("deprecation")
-    @Override public void start(boolean activeOnStart) throws IgniteCheckedException {
+    @Override public void start() throws IgniteCheckedException {
         assertParameter(discoDelay > 0, "discoveryStartupDelay > 0");
 
         startSpi();

modules/core/src/main/java/org/apache/ignite/internal/managers/deployment/GridDeploymentManager.java

@@ -91,7 +91,7 @@ public GridDeploymentManager(GridKernalContext ctx) {
     }
 
     /** {@inheritDoc} */
-    @Override public void start(boolean activeOnStart) throws IgniteCheckedException {
+    @Override public void start() throws IgniteCheckedException {
         GridProtocolHandler.registerDeploymentManager(this);
 
         assertParameter(ctx.config().getDeploymentMode() != null, "ctx.config().getDeploymentMode() != null");

modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java

@@ -73,6 +73,7 @@
 import org.apache.ignite.internal.processors.cache.CacheGroupDescriptor;
 import org.apache.ignite.internal.processors.cache.DynamicCacheDescriptor;
 import org.apache.ignite.internal.processors.cache.GridCacheAdapter;
+import org.apache.ignite.internal.processors.cluster.GridClusterStateProcessor;
 import org.apache.ignite.internal.processors.jobmetrics.GridJobMetrics;
 import org.apache.ignite.internal.processors.security.SecurityContext;
 import org.apache.ignite.internal.processors.timeout.GridTimeoutProcessor;
@@ -463,7 +464,7 @@ private void updateClientNodes(UUID leftNodeId) {
     }
 
     /** {@inheritDoc} */
-    @Override public void start(boolean activeOnStart) throws IgniteCheckedException {
+    @Override public void start() throws IgniteCheckedException {
         long totSysMemory = -1;
 
         try {
@@ -741,19 +742,36 @@ else if (type == EVT_CLIENT_NODE_RECONNECTED) {
             }
 
             @Override public void onExchange(DiscoveryDataBag dataBag) {
+                assert dataBag != null;
+                assert dataBag.joiningNodeId() != null;
+
                 if (ctx.localNodeId().equals(dataBag.joiningNodeId())) {
-                    //NodeAdded msg reached joining node after round-trip over the ring
+                    // NodeAdded msg reached joining node after round-trip over the ring.
+                    GridClusterStateProcessor stateProc = ctx.state();
+
+                    stateProc.onGridDataReceived(dataBag.gridDiscoveryData(
+                        stateProc.discoveryDataType().ordinal()));
+
                     for (GridComponent c : ctx.components()) {
-                        if (c.discoveryDataType() != null)
+                        if (c.discoveryDataType() != null && c != stateProc)
                             c.onGridDataReceived(dataBag.gridDiscoveryData(c.discoveryDataType().ordinal()));
                     }
                 }
                 else {
-                    //discovery data from newly joined node has to be applied to the current old node
+                    // Discovery data from newly joined node has to be applied to the current old node.
+                    GridClusterStateProcessor stateProc = ctx.state();
+
+                    JoiningNodeDiscoveryData data0 = dataBag.newJoinerDiscoveryData(
+                        stateProc.discoveryDataType().ordinal());
+
+                    assert data0 != null;
+
+                    stateProc.onJoiningNodeDataReceived(data0);
+
                     for (GridComponent c : ctx.components()) {
-                        if (c.discoveryDataType() != null) {
-                            JoiningNodeDiscoveryData data =
-                                    dataBag.newJoinerDiscoveryData(c.discoveryDataType().ordinal());
+                        if (c.discoveryDataType() != null && c != stateProc) {
+                            JoiningNodeDiscoveryData data = dataBag.newJoinerDiscoveryData(
+                                c.discoveryDataType().ordinal());
 
                             if (data != null)
                                 c.onJoiningNodeDataReceived(data);
@@ -1222,17 +1240,6 @@ private void checkAttributes(Iterable<ClusterNode> nodes) throws IgniteCheckedEx
                     ", rmtAddrs=" + U.addressesAsString(n) + ']');
             }
 
-            boolean rmtActiveOnStart = n.attribute(ATTR_ACTIVE_ON_START);
-
-            if (locActiveOnStart != rmtActiveOnStart) {
-                throw new IgniteCheckedException("Remote node has active on start flag different from local " +
-                    "[locId8=" + U.id8(locNode.id()) +
-                    ", locActiveOnStart=" + locActiveOnStart +
-                    ", rmtId8=" + U.id8(n.id()) +
-                    ", rmtActiveOnStart=" + rmtActiveOnStart +
-                    ", rmtAddrs=" + U.addressesAsString(n) + ']');
-            }
-
             Boolean rmtSrvcCompatibilityEnabled = n.attribute(ATTR_SERVICES_COMPATIBILITY_MODE);
 
             if (!F.eq(locSrvcCompatibilityEnabled, rmtSrvcCompatibilityEnabled)) {

modules/core/src/main/java/org/apache/ignite/internal/managers/eventstorage/GridEventStorageManager.java

@@ -273,7 +273,7 @@ private void leaveBusy() {
     }
 
     /** {@inheritDoc} */
-    @Override public void start(boolean activeOnStart) throws IgniteCheckedException {
+    @Override public void start() throws IgniteCheckedException {
         Map<IgnitePredicate<? extends Event>, int[]> evtLsnrs = ctx.config().getLocalEventListeners();
 
         if (evtLsnrs != null) {

modules/core/src/main/java/org/apache/ignite/internal/managers/failover/GridFailoverManager.java

@@ -40,7 +40,7 @@ public GridFailoverManager(GridKernalContext ctx) {
     }
 
     /** {@inheritDoc} */
-    @Override public void start(boolean activeOnStart) throws IgniteCheckedException {
+    @Override public void start() throws IgniteCheckedException {
         startSpi();
 
         if (log.isDebugEnabled())

modules/core/src/main/java/org/apache/ignite/internal/managers/indexing/GridIndexingManager.java

@@ -48,7 +48,7 @@ public GridIndexingManager(GridKernalContext ctx) {
     /**
      * @throws IgniteCheckedException Thrown in case of any errors.
      */
-    @Override public void start(boolean activeOnStart) throws IgniteCheckedException {
+    @Override public void start() throws IgniteCheckedException {
         startSpi();
 
         if (log.isDebugEnabled())

modules/core/src/main/java/org/apache/ignite/internal/managers/loadbalancer/GridLoadBalancerManager.java

@@ -45,7 +45,7 @@ public GridLoadBalancerManager(GridKernalContext ctx) {
     }
 
     /** {@inheritDoc} */
-    @Override public void start(boolean activeOnStart) throws IgniteCheckedException {
+    @Override public void start() throws IgniteCheckedException {
         startSpi();
 
         if (log.isDebugEnabled())

modules/core/src/main/java/org/apache/ignite/internal/pagemem/store/IgnitePageStoreManager.java

@@ -20,6 +20,7 @@
 import java.nio.ByteBuffer;
 import java.util.Map;
 import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.configuration.CacheConfiguration;
 import org.apache.ignite.internal.pagemem.PageMemory;
 import org.apache.ignite.internal.processors.cache.CacheGroupContext;
 import org.apache.ignite.internal.processors.cache.CacheGroupDescriptor;
@@ -187,7 +188,6 @@ public void initializeForCache(CacheGroupDescriptor grpDesc,
      * @throws IgniteCheckedException If failed.
      */
     public void storeCacheData(CacheGroupDescriptor grpDesc, StoredCacheData cacheData) throws IgniteCheckedException;
-
     /**
      * @param grpId Cache group ID.
      * @return {@code True} if index store for given cache group existed before node started.

modules/core/src/main/java/org/apache/ignite/internal/processors/GridProcessorAdapter.java

@@ -55,7 +55,7 @@ protected GridProcessorAdapter(GridKernalContext ctx) {
     }
 
     /** {@inheritDoc} */
-    @Override public void onKernalStart(boolean activeOnStart) throws IgniteCheckedException {
+    @Override public void onKernalStart() throws IgniteCheckedException {
         // No-op.
     }
 
@@ -100,7 +100,7 @@ protected GridProcessorAdapter(GridKernalContext ctx) {
     }
 
     /** {@inheritDoc} */
-    @Override public void start(boolean activeOnStart) throws IgniteCheckedException {
+    @Override public void start() throws IgniteCheckedException {
         // No-op.
     }
 

modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityProcessor.java

@@ -133,7 +133,7 @@ public GridAffinityProcessor(GridKernalContext ctx) {
     }
 
     /** {@inheritDoc} */
-    @Override public void start(boolean activeOnStart) throws IgniteCheckedException {
+    @Override public void start() throws IgniteCheckedException {
         ctx.event().addLocalEventListener(lsnr, EVT_NODE_FAILED, EVT_NODE_LEFT, EVT_NODE_JOINED);
     }
 

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheAffinitySharedManager.java

@@ -349,11 +349,11 @@ private void updateCachesInfo(ExchangeActions exchActions) {
      * @throws IgniteCheckedException If failed.
      * @return {@code True} if client-only exchange is needed.
      */
-    public boolean onCacheChangeRequest(final GridDhtPartitionsExchangeFuture fut,
+    public boolean onCacheChangeRequest(
+        final GridDhtPartitionsExchangeFuture fut,
         boolean crd,
-        final ExchangeActions exchActions)
-        throws IgniteCheckedException
-    {
+        final ExchangeActions exchActions
+    ) throws IgniteCheckedException {
         assert exchActions != null && !exchActions.empty() : exchActions;
 
         updateCachesInfo(exchActions);
@@ -377,7 +377,19 @@ public boolean onCacheChangeRequest(final GridDhtPartitionsExchangeFuture fut,
 
             NearCacheConfiguration nearCfg = null;
 
-            if (cctx.localNodeId().equals(req.initiatingNodeId())) {
+            if (exchActions.newClusterState() == ClusterState.ACTIVE) {
+                if (CU.isSystemCache(req.cacheName()))
+                    startCache = true;
+                else if (!cctx.localNode().isClient()) {
+                    startCache = cctx.cacheContext(action.descriptor().cacheId()) == null &&
+                        CU.affinityNode(cctx.localNode(), req.startCacheConfiguration().getNodeFilter());
+
+                    nearCfg = req.nearCacheConfiguration();
+                }
+                else // Only static cache configured on client must be started.
+                    startCache = cctx.kernalContext().state().isLocalConfigure(req.cacheName());
+            }
+            else if (cctx.localNodeId().equals(req.initiatingNodeId())) {
                 startCache = true;
 
                 nearCfg = req.nearCacheConfiguration();
@@ -388,9 +400,19 @@ public boolean onCacheChangeRequest(final GridDhtPartitionsExchangeFuture fut,
             }
 
             try {
+                // Save configuration before cache started.
+                if (cctx.pageStore() != null && !cctx.localNode().isClient())
+                    cctx.pageStore().storeCacheData(
+                        cacheDesc.groupDescriptor(),
+                        new StoredCacheData(req.startCacheConfiguration())
+                    );
+
                 if (startCache) {
                     cctx.cache().prepareCacheStart(cacheDesc, nearCfg, fut.topologyVersion());
 
+                    if (exchActions.newClusterState() == null)
+                        cctx.kernalContext().state().onCacheStart(req);
+
                     if (fut.cacheAddedOnExchange(cacheDesc.cacheId(), cacheDesc.receivedFrom())) {
                         if (fut.discoCache().cacheGroupAffinityNodes(cacheDesc.groupId()).isEmpty())
                             U.quietAndWarn(log, "No server nodes found for cache client: " + req.cacheName());
@@ -845,9 +867,11 @@ else if (grpHolder.client() && grp != null) {
      * @param descs Cache descriptors.
      * @throws IgniteCheckedException If failed.
      */
-    public void initStartedCaches(boolean crd,
+    public void initStartedCaches(
+        boolean crd,
         final GridDhtPartitionsExchangeFuture fut,
-        Collection<DynamicCacheDescriptor> descs) throws IgniteCheckedException {
+        Collection<DynamicCacheDescriptor> descs
+    ) throws IgniteCheckedException {
         for (DynamicCacheDescriptor desc : descs) {
             CacheGroupDescriptor grpDesc = desc.groupDescriptor();
 

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheClientReconnectDiscoveryData.java

@@ -56,7 +56,7 @@ Map<Integer, CacheGroupInfo> clientCacheGroups() {
     /**
      * @return Information about caches started on re-joining client node.
      */
-    Map<String, CacheInfo> clientCaches() {
+   public Map<String, CacheInfo> clientCaches() {
         return clientCaches;
     }
 

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheData.java

@@ -177,6 +177,13 @@ public UUID receivedFrom() {
         return rcvdFrom;
     }
 
+    /**
+     * @return Flags.
+     */
+    public long flags() {
+        return flags;
+    }
+
     /** {@inheritDoc} */
     @Override public String toString() {
         return S.toString(CacheData.class, this, "cacheName", cacheCfg.getName());