You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ag...@apache.org on 2018/01/17 10:35:18 UTC
[06/17] ignite git commit: IEP-4 Baseline topology for persistent
caches (Phase 1) Contributed by: Dmitriy Govorukhin
Dmitry Pavlov Eduard
Shangareev Ily
http://git-wip-us.apache.org/repos/asf/ignite/blob/6f7aba85/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 927fd90..54e0f56 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
@@ -17,22 +17,29 @@
package org.apache.ignite.internal.processors.cluster;
+import java.io.Serializable;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteCompute;
+import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.cluster.BaselineNode;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.events.DiscoveryEvent;
import org.apache.ignite.events.Event;
import org.apache.ignite.internal.GridKernalContext;
+import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.cluster.ClusterGroupAdapter;
import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
@@ -46,6 +53,9 @@ import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.cache.GridChangeGlobalStateMessageResponse;
import org.apache.ignite.internal.processors.cache.StateChangeRequest;
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.util.future.GridFinishedFuture;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
@@ -58,8 +68,12 @@ import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteCallable;
import org.apache.ignite.lang.IgniteFuture;
+import org.apache.ignite.lang.IgniteInClosure;
import org.apache.ignite.lang.IgniteRunnable;
+import org.apache.ignite.lang.IgniteUuid;
+import org.apache.ignite.marshaller.jdk.JdkMarshaller;
import org.apache.ignite.resources.IgniteInstanceResource;
+import org.apache.ignite.spi.IgniteNodeValidationResult;
import org.apache.ignite.spi.discovery.DiscoveryDataBag;
import org.jetbrains.annotations.Nullable;
@@ -71,13 +85,31 @@ import static org.apache.ignite.internal.managers.communication.GridIoPolicy.SYS
/**
*
*/
-public class GridClusterStateProcessor extends GridProcessorAdapter {
+public class GridClusterStateProcessor extends GridProcessorAdapter implements IGridClusterStateProcessor, MetastorageLifecycleListener {
+ /** */
+ private static final String METASTORE_CURR_BLT_KEY = "metastoreBltKey";
+
+ /** */
+ private boolean inMemoryMode;
+
+ /**
+ * Compatibility mode flag. When node detects it runs in heterogeneous cluster (nodes of different versions),
+ * it should skip baseline topology operations.
+ */
+ private volatile boolean compatibilityMode;
+
/** */
private volatile DiscoveryDataClusterState globalState;
+ /** */
+ private final BaselineTopologyHistory bltHist = new BaselineTopologyHistory();
+
/** Local action future. */
private final AtomicReference<GridChangeGlobalStateFuture> stateChangeFut = new AtomicReference<>();
+ /** */
+ private final ConcurrentMap<UUID, GridFutureAdapter<Void>> transitionFuts = new ConcurrentHashMap<>();
+
/** Future initialized if node joins when cluster state change is in progress. */
private TransitionOnJoinWaitFuture joinFut;
@@ -89,6 +121,13 @@ public class GridClusterStateProcessor extends GridProcessorAdapter {
@GridToStringExclude
private GridCacheSharedContext<?, ?> sharedCtx;
+ /** Fully initialized metastorage. */
+ @GridToStringExclude
+ private ReadWriteMetastorage metastorage;
+
+ /** */
+ private final JdkMarshaller marsh = new JdkMarshaller();
+
/** Listener. */
private final GridLocalEventListener lsr = new GridLocalEventListener() {
@Override public void onEvent(Event evt) {
@@ -115,12 +154,20 @@ public class GridClusterStateProcessor extends GridProcessorAdapter {
*/
public GridClusterStateProcessor(GridKernalContext ctx) {
super(ctx);
+
+ ctx.internalSubscriptionProcessor().registerMetastorageListener(this);
}
/**
- * @return Cluster state to be used on public API.
+ * @return {@code True} if {@link IGridClusterStateProcessor} has detected that cluster is working
+ * in compatibility mode (nodes of different versions are joined to the cluster).
*/
- public boolean publicApiActiveState() {
+ public boolean compatibilityMode() {
+ return compatibilityMode;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean publicApiActiveState(boolean waitForTransition) {
if (ctx.isDaemon())
return sendComputeCheckGlobalState();
@@ -133,19 +180,120 @@ public class GridClusterStateProcessor extends GridProcessorAdapter {
if (transitionRes != null)
return transitionRes;
- else
- return false;
+ else {
+ if (waitForTransition) {
+ GridFutureAdapter<Void> fut = transitionFuts.get(globalState.transitionRequestId());
+
+ if (fut != null) {
+ try {
+ fut.get();
+ }
+ catch (IgniteCheckedException ex) {
+ throw new IgniteException(ex);
+ }
+ }
+
+ transitionRes = globalState.transitionResult();
+
+ assert transitionRes != null;
+
+ return transitionRes;
+ }
+ else
+ return false;
+ }
}
else
return globalState.active();
}
/** {@inheritDoc} */
+ @Override public void onReadyForRead(ReadOnlyMetastorage metastorage) throws IgniteCheckedException {
+ BaselineTopology blt = (BaselineTopology) metastorage.read(METASTORE_CURR_BLT_KEY);
+
+ if (blt != null) {
+ if (log.isInfoEnabled())
+ U.log(log, "Restoring history for BaselineTopology[id=" + blt.id() + "]");
+
+ bltHist.restoreHistory(metastorage, blt.id());
+ }
+
+ onStateRestored(blt);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onReadyForReadWrite(ReadWriteMetastorage metastorage) throws IgniteCheckedException {
+ this.metastorage = metastorage;
+
+ if (compatibilityMode) {
+ if (log.isInfoEnabled())
+ log.info("BaselineTopology won't be stored as this node is running in compatibility mode");
+
+ return;
+ }
+
+ writeBaselineTopology(globalState.baselineTopology(), null);
+
+ bltHist.flushHistoryItems(metastorage);
+ }
+
+ /**
+ * Resets branching history on current BaselineTopology.
+ *
+ * @throws IgniteCheckedException If write to metastore has failed.
+ */
+ public void resetBranchingHistory(long newBranchingHash) throws IgniteCheckedException {
+ globalState.baselineTopology().resetBranchingHistory(newBranchingHash);
+
+ writeBaselineTopology(globalState.baselineTopology(), null);
+
+ U.log(log,
+ String.format("Branching history of current BaselineTopology is reset to the value %d", newBranchingHash));
+ }
+
+ /**
+ * @param blt Blt.
+ */
+ private void writeBaselineTopology(BaselineTopology blt, BaselineTopologyHistoryItem prevBltHistItem) throws IgniteCheckedException {
+ assert metastorage != null;
+
+ sharedCtx.database().checkpointReadLock();
+
+ try {
+ if (blt != null) {
+ if (log.isInfoEnabled()) {
+ U.log(log, "Writing BaselineTopology[id=" + blt.id() + "]");
+
+ if (prevBltHistItem != null)
+ U.log(log, "Writing BaselineTopologyHistoryItem[id=" + prevBltHistItem.id() + "]");
+ }
+
+ bltHist.writeHistoryItem(metastorage, prevBltHistItem);
+
+ metastorage.write(METASTORE_CURR_BLT_KEY, blt);
+ }
+ else {
+ if (log.isInfoEnabled())
+ U.log(log, "Removing BaselineTopology and history");
+
+ metastorage.remove(METASTORE_CURR_BLT_KEY);
+
+ bltHist.removeHistory(metastorage);
+ }
+ }
+ finally {
+ sharedCtx.database().checkpointReadUnlock();
+ }
+ }
+
+ /** {@inheritDoc} */
@Override public void start() throws IgniteCheckedException {
+ inMemoryMode = !CU.isPersistenceEnabled(ctx.config());
+
// Start first node as inactive if persistence is enabled.
- boolean activeOnStart = !CU.isPersistenceEnabled(ctx.config()) && ctx.config().isActiveOnStart();
+ boolean activeOnStart = inMemoryMode && ctx.config().isActiveOnStart();
- globalState = DiscoveryDataClusterState.createState(activeOnStart);
+ globalState = DiscoveryDataClusterState.createState(activeOnStart, null);
ctx.event().addLocalEventListener(lsr, EVT_NODE_LEFT, EVT_NODE_FAILED);
}
@@ -160,25 +308,46 @@ public class GridClusterStateProcessor extends GridProcessorAdapter {
super.onKernalStop(cancel);
}
- /**
- * @param discoCache Discovery data cache.
- * @return If transition is in progress returns future which is completed when transition finishes.
- */
- @Nullable public IgniteInternalFuture<Boolean> onLocalJoin(DiscoCache discoCache) {
- if (globalState.transition()) {
- joinFut = new TransitionOnJoinWaitFuture(globalState, discoCache);
+ /** {@inheritDoc} */
+ @Override @Nullable public IgniteInternalFuture<Boolean> onLocalJoin(DiscoCache discoCache) {
+ final DiscoveryDataClusterState state = globalState;
+
+ if (state.transition()) {
+ joinFut = new TransitionOnJoinWaitFuture(state, discoCache);
return joinFut;
}
+ else if (!ctx.clientNode()
+ && !ctx.isDaemon()
+ && ctx.config().isAutoActivationEnabled()
+ && !state.active()
+ && isBaselineSatisfied(state.baselineTopology(), discoCache.serverNodes()))
+ changeGlobalState0(true, state.baselineTopology(), false);
return null;
}
/**
- * @param node Failed node.
- * @return Message if cluster state changed.
+ * Checks whether all conditions to meet BaselineTopology are satisfied.
*/
- @Nullable public ChangeGlobalStateFinishMessage onNodeLeft(ClusterNode node) {
+ private boolean isBaselineSatisfied(BaselineTopology blt, List<ClusterNode> serverNodes) {
+ if (blt == null)
+ return false;
+
+ if (blt.consistentIds() == null)
+ return false;
+
+ if (//only node participating in BaselineTopology is allowed to send activation command...
+ blt.consistentIds().contains(ctx.discovery().localNode().consistentId())
+ //...and with this node BaselineTopology is reached
+ && blt.isSatisfied(serverNodes))
+ return true;
+
+ return false;
+ }
+
+ /** {@inheritDoc} */
+ @Override @Nullable public ChangeGlobalStateFinishMessage onNodeLeft(ClusterNode node) {
if (globalState.transition()) {
Set<UUID> nodes = globalState.transitionNodes();
@@ -187,7 +356,7 @@ public class GridClusterStateProcessor extends GridProcessorAdapter {
"Switching to inactive state.");
ChangeGlobalStateFinishMessage msg =
- new ChangeGlobalStateFinishMessage(globalState.transitionRequestId(), false);
+ new ChangeGlobalStateFinishMessage(globalState.transitionRequestId(), false, false);
onStateFinishMessage(msg);
@@ -198,14 +367,16 @@ public class GridClusterStateProcessor extends GridProcessorAdapter {
return null;
}
- /**
- * @param msg Message.
- */
- public void onStateFinishMessage(ChangeGlobalStateFinishMessage msg) {
- if (msg.requestId().equals(globalState.transitionRequestId())) {
+ /** {@inheritDoc} */
+ @Override public void onStateFinishMessage(ChangeGlobalStateFinishMessage msg) {
+ DiscoveryDataClusterState state = globalState;
+
+ if (msg.requestId().equals(state.transitionRequestId())) {
log.info("Received state change finish message: " + msg.clusterActive());
- globalState = DiscoveryDataClusterState.createState(msg.clusterActive());
+ globalState = globalState.finish(msg.success());
+
+ afterStateChangeFinished(msg.id(), msg.success());
ctx.cache().onStateChangeFinish(msg);
@@ -213,22 +384,39 @@ public class GridClusterStateProcessor extends GridProcessorAdapter {
if (joinFut != null)
joinFut.onDone(false);
+
+ GridFutureAdapter<Void> transitionFut = transitionFuts.remove(state.transitionRequestId());
+
+ if (transitionFut != null) {
+ state.setTransitionResult(msg.requestId(), msg.clusterActive());
+
+ transitionFut.onDone();
+ }
}
else
U.warn(log, "Received state finish message with unexpected ID: " + msg);
}
- /**
- * @param topVer Current topology version.
- * @param msg Message.
- * @param discoCache Current nodes.
- * @return {@code True} if need start state change process.
- */
- public boolean onStateChangeMessage(AffinityTopologyVersion topVer,
+ /** */
+ protected void afterStateChangeFinished(IgniteUuid msgId, boolean success) {
+ // no-op
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean onStateChangeMessage(
+ AffinityTopologyVersion topVer,
ChangeGlobalStateMessage msg,
- DiscoCache discoCache) {
- if (globalState.transition()) {
- if (globalState.active() != msg.activate()) {
+ DiscoCache discoCache
+ ) {
+ DiscoveryDataClusterState state = globalState;
+
+ if (log.isInfoEnabled())
+ U.log(log, "Received " + prettyStr(msg.activate()) + " request with BaselineTopology" +
+ (msg.baselineTopology() == null ? ": null"
+ : "[id=" + msg.baselineTopology().id() + "]"));
+
+ if (state.transition()) {
+ if (isApplicable(msg, state)) {
GridChangeGlobalStateFuture fut = changeStateFuture(msg);
if (fut != null)
@@ -237,27 +425,31 @@ public class GridClusterStateProcessor extends GridProcessorAdapter {
else {
final GridChangeGlobalStateFuture stateFut = changeStateFuture(msg);
- if (stateFut != null) {
- IgniteInternalFuture<?> exchFut = ctx.cache().context().exchange().affinityReadyFuture(
- globalState.transitionTopologyVersion());
+ GridFutureAdapter<Void> transitionFut = transitionFuts.get(state.transitionRequestId());
- if (exchFut == null)
- exchFut = new GridFinishedFuture<>();
+ if (stateFut != null && transitionFut != null) {
+ transitionFut.listen(new IgniteInClosure<IgniteInternalFuture<Void>>() {
+ @Override public void apply(IgniteInternalFuture<Void> fut) {
+ try {
+ fut.get();
+
+ stateFut.onDone();
+ }
+ catch (Exception ex) {
+ stateFut.onDone(ex);
+ }
- exchFut.listen(new CI1<IgniteInternalFuture<?>>() {
- @Override public void apply(IgniteInternalFuture<?> exchFut) {
- stateFut.onDone();
}
});
}
}
}
else {
- if (globalState.active() != msg.activate()) {
+ if (isApplicable(msg, state)) {
ExchangeActions exchangeActions;
try {
- exchangeActions = ctx.cache().onStateChangeRequest(msg, topVer);
+ exchangeActions = ctx.cache().onStateChangeRequest(msg, topVer, state);
}
catch (IgniteCheckedException e) {
GridChangeGlobalStateFuture fut = changeStateFuture(msg);
@@ -281,14 +473,25 @@ public class GridClusterStateProcessor extends GridProcessorAdapter {
if (log.isInfoEnabled())
log.info("Started state transition: " + msg.activate());
- globalState = DiscoveryDataClusterState.createTransitionState(msg.activate(),
+ BaselineTopologyHistoryItem bltHistItem = BaselineTopologyHistoryItem.fromBaseline(
+ globalState.baselineTopology());
+
+ transitionFuts.put(msg.requestId(), new GridFutureAdapter<Void>());
+
+ globalState = DiscoveryDataClusterState.createTransitionState(
+ globalState,
+ msg.activate(),
+ msg.activate() ? msg.baselineTopology() : globalState.baselineTopology(),
msg.requestId(),
topVer,
nodeIds);
+ if (msg.forceChangeBaselineTopology())
+ globalState.setTransitionResult(msg.requestId(), msg.activate());
+
AffinityTopologyVersion stateChangeTopVer = topVer.nextMinorVersion();
- StateChangeRequest req = new StateChangeRequest(msg, stateChangeTopVer);
+ StateChangeRequest req = new StateChangeRequest(msg, bltHistItem, msg.activate() != state.active(), stateChangeTopVer);
exchangeActions.stateChangeRequest(req);
@@ -309,12 +512,34 @@ public class GridClusterStateProcessor extends GridProcessorAdapter {
}
/**
- * @return Current cluster state, should be called only from discovery thread.
+ * @param msg State change message.
+ * @param state Current cluster state.
+ * @return {@code True} if state change from message can be applied to the current state.
+ */
+ protected boolean isApplicable(ChangeGlobalStateMessage msg, DiscoveryDataClusterState state) {
+ return !isEquivalent(msg, state);
+ }
+
+ /**
+ * @param msg State change message.
+ * @param state Current cluster state.
+ * @return {@code True} if states are equivalent.
*/
- public DiscoveryDataClusterState clusterState() {
+ protected static boolean isEquivalent(ChangeGlobalStateMessage msg, DiscoveryDataClusterState state) {
+ return (msg.activate() == state.active() && BaselineTopology.equals(msg.baselineTopology(), state.baselineTopology()));
+ }
+
+ /** {@inheritDoc} */
+ @Override public DiscoveryDataClusterState clusterState() {
return globalState;
}
+ /** {@inheritDoc} */
+ @Override public DiscoveryDataClusterState pendingState(ChangeGlobalStateMessage stateMsg) {
+ return DiscoveryDataClusterState.createState(stateMsg.activate() || stateMsg.forceChangeBaselineTopology(),
+ stateMsg.baselineTopology());
+ }
+
/**
* @param msg State change message.
* @return Local future for state change process.
@@ -346,15 +571,13 @@ public class GridClusterStateProcessor extends GridProcessorAdapter {
* @param activate New state.
* @return State change error.
*/
- private IgniteCheckedException concurrentStateChangeError(boolean activate) {
+ protected IgniteCheckedException concurrentStateChangeError(boolean activate) {
return new IgniteCheckedException("Failed to " + prettyStr(activate) +
", because another state change operation is currently in progress: " + prettyStr(!activate));
}
- /**
- *
- */
- public void cacheProcessorStarted() {
+ /** {@inheritDoc} */
+ @Override public void cacheProcessorStarted() {
cacheProc = ctx.cache();
sharedCtx = cacheProc.context();
@@ -391,40 +614,192 @@ public class GridClusterStateProcessor extends GridProcessorAdapter {
}
/** {@inheritDoc} */
+ @Override public void collectJoiningNodeData(DiscoveryDataBag dataBag) {
+ try {
+ byte[] marshalledState = marsh.marshal(globalState);
+
+ dataBag.addJoiningNodeData(discoveryDataType().ordinal(), marshalledState);
+ }
+ catch (IgniteCheckedException e) {
+ throw new IgniteException(e);
+ }
+ }
+
+ /** {@inheritDoc} */
@Override public void collectGridNodeData(DiscoveryDataBag dataBag) {
- if (!dataBag.commonDataCollectedFor(STATE_PROC.ordinal()))
- dataBag.addGridCommonData(STATE_PROC.ordinal(), globalState);
+ if (!dataBag.commonDataCollectedFor(STATE_PROC.ordinal())) {
+ DiscoveryDataBag.JoiningNodeDiscoveryData joiningNodeData = dataBag.newJoinerDiscoveryData(STATE_PROC.ordinal());
+
+ BaselineTopologyHistory historyToSend = null;
+
+ if (joiningNodeData != null) {
+ if (!joiningNodeData.hasJoiningNodeData()) {
+ //compatibility mode: old nodes don't send any data on join, so coordinator of new version
+ //doesn't send BaselineTopology history, only its current globalState
+ dataBag.addGridCommonData(STATE_PROC.ordinal(), globalState);
+
+ return;
+ }
+
+ DiscoveryDataClusterState joiningNodeState = null;
+
+ try {
+ if (joiningNodeData.joiningNodeData() != null)
+ joiningNodeState = marsh.unmarshal(
+ (byte[]) joiningNodeData.joiningNodeData(),
+ U.resolveClassLoader(ctx.config()));
+ } catch (IgniteCheckedException e) {
+ U.error(log, "Failed to unmarshal disco data from joining node: " + joiningNodeData.joiningNodeId());
+
+ return;
+ }
+
+ if (!bltHist.isEmpty()) {
+ if (joiningNodeState != null && joiningNodeState.baselineTopology() != null) {
+ int lastId = joiningNodeState.baselineTopology().id();
+
+ historyToSend = bltHist.tailFrom(lastId);
+ }
+ else
+ historyToSend = bltHist;
+ }
+
+ dataBag.addGridCommonData(STATE_PROC.ordinal(), new BaselineStateAndHistoryData(globalState, historyToSend));
+ }
+ }
}
/** {@inheritDoc} */
@Override public void onGridDataReceived(DiscoveryDataBag.GridDiscoveryData data) {
- DiscoveryDataClusterState state = (DiscoveryDataClusterState)data.commonData();
+ if (data.commonData() instanceof DiscoveryDataClusterState) {
+ if (globalState != null && globalState.baselineTopology() != null)
+ //node with BaselineTopology is not allowed to join mixed cluster
+ // (where some nodes don't support BaselineTopology)
+ throw new IgniteException("Node with BaselineTopology cannot join" +
+ " mixed cluster running in compatibility mode");
+
+ globalState = (DiscoveryDataClusterState) data.commonData();
+
+ compatibilityMode = true;
+
+ return;
+ }
+
+ BaselineStateAndHistoryData stateDiscoData = (BaselineStateAndHistoryData)data.commonData();
+
+ if (stateDiscoData != null) {
+ DiscoveryDataClusterState state = stateDiscoData.globalState;
+
+ if (state.transition())
+ transitionFuts.put(state.transitionRequestId(), new GridFutureAdapter<Void>());
- if (state != null)
globalState = state;
+
+ if (stateDiscoData.recentHistory != null) {
+ for (BaselineTopologyHistoryItem item : stateDiscoData.recentHistory.history())
+ bltHist.bufferHistoryItemForStore(item);
+ }
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteInternalFuture<?> changeGlobalState(
+ final boolean activate,
+ Collection<? extends BaselineNode> baselineNodes,
+ boolean forceChangeBaselineTopology
+ ) {
+ if (inMemoryMode)
+ return changeGlobalState0(activate, null, false);
+
+ BaselineTopology newBlt = compatibilityMode ? null :
+ calculateNewBaselineTopology(activate, baselineNodes, forceChangeBaselineTopology);
+
+ return changeGlobalState0(activate, newBlt, forceChangeBaselineTopology);
}
/**
- * @param activate New cluster state.
- * @return State change future.
+ *
*/
- public IgniteInternalFuture<?> changeGlobalState(final boolean activate) {
+ private BaselineTopology calculateNewBaselineTopology(final boolean activate,
+ Collection<? extends BaselineNode> baselineNodes,
+ boolean forceChangeBaselineTopology) {
+ BaselineTopology newBlt;
+
+ BaselineTopology currentBlt = globalState.baselineTopology();
+
+ int newBltId = 0;
+
+ if (currentBlt != null)
+ newBltId = activate ? currentBlt.id() + 1 : currentBlt.id();
+
+ if (baselineNodes != null && !baselineNodes.isEmpty()) {
+ List<BaselineNode> baselineNodes0 = new ArrayList<>();
+
+ for (BaselineNode node : baselineNodes) {
+ if (node instanceof ClusterNode) {
+ ClusterNode clusterNode = (ClusterNode) node;
+
+ if (!clusterNode.isClient() && !clusterNode.isDaemon())
+ baselineNodes0.add(node);
+ }
+ else
+ baselineNodes0.add(node);
+ }
+
+ baselineNodes = baselineNodes0;
+ }
+
+ if (forceChangeBaselineTopology)
+ newBlt = BaselineTopology.build(baselineNodes, newBltId);
+ else if (activate) {
+ if (baselineNodes == null)
+ baselineNodes = baselineNodes();
+
+ if (currentBlt == null)
+ newBlt = BaselineTopology.build(baselineNodes, newBltId);
+ else {
+ newBlt = currentBlt;
+
+ newBlt.updateHistory(baselineNodes);
+ }
+ }
+ else
+ newBlt = null;
+
+ return newBlt;
+ }
+
+ /** */
+ private Collection<BaselineNode> baselineNodes() {
+ List<ClusterNode> clNodes = ctx.discovery().serverNodes(AffinityTopologyVersion.NONE);
+
+ ArrayList<BaselineNode> bltNodes = new ArrayList<>(clNodes.size());
+
+ for (ClusterNode clNode : clNodes)
+ bltNodes.add(clNode);
+
+ return bltNodes;
+ }
+
+ /** */
+ private IgniteInternalFuture<?> changeGlobalState0(final boolean activate,
+ BaselineTopology blt, boolean forceChangeBaselineTopology) {
if (ctx.isDaemon() || ctx.clientNode()) {
GridFutureAdapter<Void> fut = new GridFutureAdapter<>();
- sendComputeChangeGlobalState(activate, fut);
+ sendComputeChangeGlobalState(activate, blt, forceChangeBaselineTopology, fut);
return fut;
}
if (cacheProc.transactions().tx() != null || sharedCtx.lockedTopologyVersion(null) != null) {
return new GridFinishedFuture<>(new IgniteCheckedException("Failed to " + prettyStr(activate) +
- " cluster (must invoke the method outside of an active transaction or lock)."));
+ " cluster (must invoke the method outside of an active transaction)."));
}
DiscoveryDataClusterState curState = globalState;
- if (!curState.transition() && curState.active() == activate)
+ if (!curState.transition() && curState.active() == activate && BaselineTopology.equals(curState.baselineTopology(), blt))
return new GridFinishedFuture<>();
GridChangeGlobalStateFuture startedFut = null;
@@ -473,9 +848,15 @@ public class GridClusterStateProcessor extends GridProcessorAdapter {
ChangeGlobalStateMessage msg = new ChangeGlobalStateMessage(startedFut.requestId,
ctx.localNodeId(),
storedCfgs,
- activate);
+ activate,
+ blt,
+ forceChangeBaselineTopology,
+ System.currentTimeMillis());
try {
+ if (log.isInfoEnabled())
+ U.log(log, "Sending " + prettyStr(activate) + " request with BaselineTopology " + blt);
+
ctx.discovery().sendCustomEvent(msg);
if (ctx.isStopping())
@@ -488,14 +869,115 @@ public class GridClusterStateProcessor extends GridProcessorAdapter {
startedFut.onDone(e);
}
- return startedFut;
+ return wrapStateChangeFuture(startedFut, msg);
+ }
+
+ /** {@inheritDoc} */
+ @Nullable @Override public IgniteNodeValidationResult validateNode(ClusterNode node, DiscoveryDataBag.JoiningNodeDiscoveryData discoData) {
+ if (node.isClient() || node.isDaemon())
+ return null;
+
+ if (discoData.joiningNodeData() == null) {
+ if (globalState.baselineTopology() != null) {
+ String msg = "Node not supporting BaselineTopology" +
+ " is not allowed to join the cluster with BaselineTopology";
+
+ return new IgniteNodeValidationResult(node.id(), msg, msg);
+ }
+
+ return null;
+ }
+
+ DiscoveryDataClusterState joiningNodeState;
+
+ try {
+ joiningNodeState = marsh.unmarshal((byte[]) discoData.joiningNodeData(), Thread.currentThread().getContextClassLoader());
+ } catch (IgniteCheckedException e) {
+ String msg = "Error on unmarshalling discovery data " +
+ "from node " + node.consistentId() + ": " + e.getMessage() +
+ "; node is not allowed to join";
+
+ return new IgniteNodeValidationResult(node.id(), msg , msg);
+ }
+
+ if (joiningNodeState == null || joiningNodeState.baselineTopology() == null)
+ return null;
+
+ if (globalState == null || globalState.baselineTopology() == null) {
+ if (joiningNodeState != null && joiningNodeState.baselineTopology() != null) {
+ String msg = "Node with set up BaselineTopology is not allowed to join cluster without one: " + node.consistentId();
+
+ return new IgniteNodeValidationResult(node.id(), msg, msg);
+ }
+ }
+
+ BaselineTopology joiningNodeBlt = joiningNodeState.baselineTopology();
+ BaselineTopology clusterBlt = globalState.baselineTopology();
+
+ String recommendation = " Consider cleaning persistent storage of the node and adding it to the cluster again.";
+
+ if (joiningNodeBlt.id() > clusterBlt.id()) {
+ String msg = "BaselineTopology of joining node ("
+ + node.consistentId()
+ + ") is not compatible with BaselineTopology in the cluster."
+ + " Joining node BlT id (" + joiningNodeBlt.id()
+ + ") is greater than cluster BlT id (" + clusterBlt.id() + ")."
+ + " New BaselineTopology was set on joining node with set-baseline command."
+ + recommendation;
+
+ return new IgniteNodeValidationResult(node.id(), msg, msg);
+ }
+
+ if (joiningNodeBlt.id() == clusterBlt.id()) {
+ if (!clusterBlt.isCompatibleWith(joiningNodeBlt)) {
+ String msg = "BaselineTopology of joining node ("
+ + node.consistentId()
+ + " ) is not compatible with BaselineTopology in the cluster."
+ + " Branching history of cluster BlT (" + clusterBlt.branchingHistory()
+ + ") doesn't contain branching point hash of joining node BlT ("
+ + joiningNodeBlt.branchingPointHash()
+ + ")." + recommendation;
+
+ return new IgniteNodeValidationResult(node.id(), msg, msg);
+ }
+ }
+ else if (joiningNodeBlt.id() < clusterBlt.id()) {
+ if (!bltHist.isCompatibleWith(joiningNodeBlt)) {
+ String msg = "BaselineTopology of joining node ("
+ + node.consistentId()
+ + ") is not compatible with BaselineTopology in the cluster."
+ + " BlT id of joining node (" + joiningNodeBlt.id()
+ + ") less than BlT id of cluster (" + clusterBlt.id()
+ + ") but cluster's BaselineHistory doesn't contain branching point hash of joining node BlT ("
+ + joiningNodeBlt.branchingPointHash()
+ + ")." + recommendation;
+
+ return new IgniteNodeValidationResult(node.id(), msg, msg);
+ }
+ }
+
+ return null;
+ }
+
+ /**
+ * @param fut Original state change future.
+ * @param msg State change message.
+ * @return Wrapped state change future.
+ */
+ protected IgniteInternalFuture<?> wrapStateChangeFuture(IgniteInternalFuture fut, ChangeGlobalStateMessage msg) {
+ return fut;
}
/**
* @param activate New cluster state.
* @param resFut State change future.
*/
- private void sendComputeChangeGlobalState(boolean activate, final GridFutureAdapter<Void> resFut) {
+ private void sendComputeChangeGlobalState(
+ boolean activate,
+ BaselineTopology blt,
+ boolean forceBlt,
+ final GridFutureAdapter<Void> resFut
+ ) {
AffinityTopologyVersion topVer = ctx.discovery().topologyVersionEx();
if (log.isInfoEnabled()) {
@@ -507,7 +989,7 @@ public class GridClusterStateProcessor extends GridProcessorAdapter {
IgniteCompute comp = ((ClusterGroupAdapter)ctx.cluster().get().forServers()).compute();
- IgniteFuture<Void> fut = comp.runAsync(new ClientChangeGlobalStateComputeRequest(activate));
+ IgniteFuture<Void> fut = comp.runAsync(new ClientChangeGlobalStateComputeRequest(activate, blt, forceBlt));
fut.listen(new CI1<IgniteFuture>() {
@Override public void apply(IgniteFuture fut) {
@@ -549,32 +1031,31 @@ public class GridClusterStateProcessor extends GridProcessorAdapter {
});
}
- /**
- * @param errs Errors.
- * @param req State change request.
- */
- public void onStateChangeError(Map<UUID, Exception> errs, StateChangeRequest req) {
+ /** {@inheritDoc} */
+ @Override public void onStateChangeError(Map<UUID, Exception> errs, StateChangeRequest req) {
assert !F.isEmpty(errs);
// Revert caches start if activation request fail.
- if (req.activate()) {
- try {
- cacheProc.onKernalStopCaches(true);
+ if (req.activeChanged()) {
+ if (req.activate()) {
+ try {
+ cacheProc.onKernalStopCaches(true);
- cacheProc.stopCaches(true);
+ cacheProc.stopCaches(true);
- sharedCtx.affinity().removeAllCacheInfo();
+ sharedCtx.affinity().removeAllCacheInfo();
- if (!ctx.clientNode())
- sharedCtx.deactivate();
+ if (!ctx.clientNode())
+ sharedCtx.deactivate();
+ }
+ catch (Exception e) {
+ U.error(log, "Failed to revert activation request changes", e);
+ }
}
- catch (Exception e) {
- U.error(log, "Failed to revert activation request changes", e);
+ else {
+ //todo https://issues.apache.org/jira/browse/IGNITE-5480
}
}
- else {
- //todo https://issues.apache.org/jira/browse/IGNITE-5480
- }
GridChangeGlobalStateFuture fut = changeStateFuture(req.initiatorNodeId(), req.requestId());
@@ -596,6 +1077,8 @@ public class GridClusterStateProcessor extends GridProcessorAdapter {
* @param req State change request.
*/
private void onFinalActivate(final StateChangeRequest req) {
+ ctx.dataStructures().onBeforeActivate();
+
ctx.closure().runLocalSafe(new Runnable() {
@Override public void run() {
boolean client = ctx.clientNode();
@@ -618,37 +1101,44 @@ public class GridClusterStateProcessor extends GridProcessorAdapter {
+ ctx.localNodeId() + ", client=" + client + ", topVer=" + req.topologyVersion() + "]");
}
catch (Exception ex) {
- e = new IgniteCheckedException("Failed to perform final activation steps", ex);
-
- U.error(log, "Failed to perform final activation steps [nodeId=" + ctx.localNodeId() +
- ", client=" + client + ", topVer=" + req.topologyVersion() + "]", ex);
- }
- finally {
- globalState.setTransitionResult(req.requestId(), true);
-
- sendChangeGlobalStateResponse(req.requestId(), req.initiatorNodeId(), e);
+ throw new IgniteException(ex);
}
}
});
}
- /**
- * @param req State change request.
- */
- private void onFinalDeActivate(final StateChangeRequest req) {
- globalState.setTransitionResult(req.requestId(), false);
+ /** {@inheritDoc} */
+ @Override public void onStateChangeExchangeDone(StateChangeRequest req) {
+ try {
+ if (req.activeChanged()) {
+ if (req.activate())
+ onFinalActivate(req);
+
+ globalState.setTransitionResult(req.requestId(), req.activate());
+ }
+
+ sendChangeGlobalStateResponse(req.requestId(), req.initiatorNodeId(), null);
+ }
+ catch (Exception ex) {
+ Exception e = new IgniteCheckedException("Failed to perform final activation steps", ex);
- sendChangeGlobalStateResponse(req.requestId(), req.initiatorNodeId(), null);
+ U.error(log, "Failed to perform final activation steps [nodeId=" + ctx.localNodeId() +
+ ", client=" + ctx.clientNode() + ", topVer=" + req.topologyVersion() + "]", ex);
+
+ sendChangeGlobalStateResponse(req.requestId(), req.initiatorNodeId(), e);
+ }
}
- /**
- * @param req State change request.
- */
- public void onStateChangeExchangeDone(StateChangeRequest req) {
- if (req.activate())
- onFinalActivate(req);
- else
- onFinalDeActivate(req);
+ /** {@inheritDoc} */
+ @Override public void onBaselineTopologyChanged(BaselineTopology blt, BaselineTopologyHistoryItem prevBltHistItem) throws IgniteCheckedException {
+ if (compatibilityMode) {
+ if (log.isDebugEnabled())
+ log.info("BaselineTopology won't be stored as this node is running in compatibility mode");
+
+ return;
+ }
+
+ writeBaselineTopology(blt, prevBltHistItem);
}
/**
@@ -718,6 +1208,27 @@ public class GridClusterStateProcessor extends GridProcessorAdapter {
}
}
+ /** */
+ private void onStateRestored(BaselineTopology blt) {
+ DiscoveryDataClusterState state = globalState;
+
+ if (!state.active() && !state.transition() && state.baselineTopology() == null) {
+ DiscoveryDataClusterState newState = DiscoveryDataClusterState.createState(false, blt);
+
+ globalState = newState;
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onExchangeFinishedOnCoordinator(IgniteInternalFuture exchangeFuture, boolean hasMovingPartitions) {
+ // no-op
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean evictionsAllowed() {
+ return true;
+ }
+
/**
* @param activate Activate.
* @return Activate flag string.
@@ -734,7 +1245,7 @@ public class GridClusterStateProcessor extends GridProcessorAdapter {
/**
*
*/
- private static class GridChangeGlobalStateFuture extends GridFutureAdapter<Void> {
+ private class GridChangeGlobalStateFuture extends GridFutureAdapter<Void> {
/** Request id. */
@GridToStringInclude
private final UUID requestId;
@@ -868,7 +1379,7 @@ public class GridClusterStateProcessor extends GridProcessorAdapter {
/** {@inheritDoc} */
@Override public boolean onDone(@Nullable Void res, @Nullable Throwable err) {
if (super.onDone(res, err)) {
- ctx.state().stateChangeFut.compareAndSet(this, null);
+ stateChangeFut.compareAndSet(this, null);
return true;
}
@@ -892,20 +1403,37 @@ public class GridClusterStateProcessor extends GridProcessorAdapter {
/** */
private final boolean activate;
+ /** */
+ private final BaselineTopology baselineTopology;
+
+ /** */
+ private final boolean forceChangeBaselineTopology;
+
/** Ignite. */
@IgniteInstanceResource
- private Ignite ig;
+ private IgniteEx ig;
/**
* @param activate New cluster state.
*/
- private ClientChangeGlobalStateComputeRequest(boolean activate) {
+ private ClientChangeGlobalStateComputeRequest(boolean activate, BaselineTopology blt, boolean forceBlt) {
this.activate = activate;
+ this.baselineTopology = blt;
+ this.forceChangeBaselineTopology = forceBlt;
}
/** {@inheritDoc} */
@Override public void run() {
- ig.active(activate);
+ try {
+ ig.context().state().changeGlobalState(
+ activate,
+ baselineTopology != null ? baselineTopology.currentBaseline() : null,
+ forceChangeBaselineTopology
+ ).get();
+ }
+ catch (IgniteCheckedException ex) {
+ throw new IgniteException(ex);
+ }
}
}
@@ -961,4 +1489,22 @@ public class GridClusterStateProcessor extends GridProcessorAdapter {
return false;
}
}
+
+ /** */
+ private static class BaselineStateAndHistoryData implements Serializable {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** */
+ private final DiscoveryDataClusterState globalState;
+
+ /** */
+ private final BaselineTopologyHistory recentHistory;
+
+ /** */
+ BaselineStateAndHistoryData(DiscoveryDataClusterState globalState, BaselineTopologyHistory recentHistory) {
+ this.globalState = globalState;
+ this.recentHistory = recentHistory;
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6f7aba85/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/IGridClusterStateProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/IGridClusterStateProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/IGridClusterStateProcessor.java
new file mode 100644
index 0000000..bc72a51
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/IGridClusterStateProcessor.java
@@ -0,0 +1,122 @@
+/*
+ * 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.cluster;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.UUID;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.cluster.BaselineNode;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.managers.discovery.DiscoCache;
+import org.apache.ignite.internal.processors.GridProcessor;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.cache.StateChangeRequest;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ *
+ */
+public interface IGridClusterStateProcessor extends GridProcessor {
+ /**
+ * @return Cluster state to be used on public API.
+ */
+ boolean publicApiActiveState(boolean waitForTransition);
+
+ /**
+ * @param discoCache Discovery data cache.
+ * @return If transition is in progress returns future which is completed when transition finishes.
+ */
+ @Nullable IgniteInternalFuture<Boolean> onLocalJoin(DiscoCache discoCache);
+
+ /**
+ * @param node Failed node.
+ * @return Message if cluster state changed.
+ */
+ @Nullable ChangeGlobalStateFinishMessage onNodeLeft(ClusterNode node);
+
+ /**
+ * @param msg Message.
+ */
+ void onStateFinishMessage(ChangeGlobalStateFinishMessage msg);
+
+ /**
+ * @param topVer Current topology version.
+ * @param msg Message.
+ * @param discoCache Current nodes.
+ * @return {@code True} if need start state change process.
+ */
+ boolean onStateChangeMessage(AffinityTopologyVersion topVer,
+ ChangeGlobalStateMessage msg,
+ DiscoCache discoCache);
+
+ /**
+ * @return Current cluster state, should be called only from discovery thread.
+ */
+ DiscoveryDataClusterState clusterState();
+
+ /**
+ * @return Pending cluster state which will be used when state transition is finished.
+ */
+ DiscoveryDataClusterState pendingState(ChangeGlobalStateMessage stateMsg);
+
+ /**
+ *
+ */
+ void cacheProcessorStarted();
+
+ /**
+ * @param activate New cluster state.
+ * @return State change future.
+ */
+ IgniteInternalFuture<?> changeGlobalState(
+ boolean activate,
+ Collection<? extends BaselineNode> baselineNodes,
+ boolean forceChangeBaselineTopology
+ );
+
+ /**
+ * @param errs Errors.
+ * @param req State change request.
+ */
+ void onStateChangeError(Map<UUID, Exception> errs, StateChangeRequest req);
+
+ /**
+ * @param req State change request.
+ */
+ void onStateChangeExchangeDone(StateChangeRequest req);
+
+ /**
+ * @param blt New baseline topology.
+ * @param prevBltHistItem Previous baseline history item.
+ */
+ void onBaselineTopologyChanged(BaselineTopology blt, BaselineTopologyHistoryItem prevBltHistItem) throws IgniteCheckedException;
+
+ /**
+ * @param exchangeFuture Exchange future.
+ * @param hasMovingPartitions {@code True} if there are moving partitions.
+ */
+ void onExchangeFinishedOnCoordinator(IgniteInternalFuture exchangeFuture, boolean hasMovingPartitions);
+
+ /**
+ * @return {@code True} if partition evictions are allowed in current state.
+ */
+ boolean evictionsAllowed();
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6f7aba85/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/AtomicDataStructureProxy.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/AtomicDataStructureProxy.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/AtomicDataStructureProxy.java
index 8355210..2c336a0 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/AtomicDataStructureProxy.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/AtomicDataStructureProxy.java
@@ -89,8 +89,8 @@ public abstract class AtomicDataStructureProxy<V extends AtomicDataStructureValu
* @return Datastructure name.
*/
public String name() {
- return name;
- }
+ return name;
+ }
/**
* @return Key value.
http://git-wip-us.apache.org/repos/asf/ignite/blob/6f7aba85/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/DataStructuresProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/DataStructuresProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/DataStructuresProcessor.java
index acd8c11..99b2fa8 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/DataStructuresProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/DataStructuresProcessor.java
@@ -17,7 +17,6 @@
package org.apache.ignite.internal.processors.datastructures;
-import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
@@ -74,7 +73,6 @@ import org.apache.ignite.internal.util.typedef.CX1;
import org.apache.ignite.internal.util.typedef.T2;
import org.apache.ignite.internal.util.typedef.X;
import org.apache.ignite.internal.util.typedef.internal.A;
-import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.internal.util.typedef.internal.GPR;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
@@ -105,10 +103,10 @@ import static org.apache.ignite.transactions.TransactionIsolation.REPEATABLE_REA
*/
public final class DataStructuresProcessor extends GridProcessorAdapter implements IgniteChangeGlobalStateSupport {
/** */
- private static final String DEFAULT_DS_GROUP_NAME = "default-ds-group";
+ public static final String DEFAULT_VOLATILE_DS_GROUP_NAME = "default-volatile-ds-group";
/** */
- private static final String DEFAULT_VOLATILE_DS_GROUP_NAME = "default-volatile-ds-group";
+ private static final String DEFAULT_DS_GROUP_NAME = "default-ds-group";
/** */
private static final String DS_CACHE_NAME_PREFIX = "datastructures_";
@@ -172,13 +170,13 @@ public final class DataStructuresProcessor extends GridProcessorAdapter implemen
}
/** {@inheritDoc} */
- @Override public void start() throws IgniteCheckedException {
+ @Override public void start() {
ctx.event().addLocalEventListener(lsnr, EVT_NODE_LEFT, EVT_NODE_FAILED);
}
/** {@inheritDoc} */
@SuppressWarnings("unchecked")
- @Override public void onKernalStart(boolean active) throws IgniteCheckedException {
+ @Override public void onKernalStart(boolean active) {
if (ctx.config().isDaemon() || !active)
return;
@@ -188,7 +186,14 @@ public final class DataStructuresProcessor extends GridProcessorAdapter implemen
/**
*
*/
- private void onKernalStart0(){
+ public void onBeforeActivate() {
+ initLatch = new CountDownLatch(1);
+ }
+
+ /**
+ *
+ */
+ private void onKernalStart0() {
initLatch.countDown();
}
@@ -225,10 +230,14 @@ public final class DataStructuresProcessor extends GridProcessorAdapter implemen
((GridCacheLockEx)ds).onStop();
}
- if (initLatch.getCount() > 0) {
+ CountDownLatch init0 = initLatch;
+
+ if (init0 != null && init0.getCount() > 0) {
initFailed = true;
- initLatch.countDown();
+ init0.countDown();
+
+ initLatch = null;
}
Iterator<Map.Entry<Integer, UUID>> iter = qryIdMap.entrySet().iterator();
@@ -245,22 +254,20 @@ public final class DataStructuresProcessor extends GridProcessorAdapter implemen
}
/** {@inheritDoc} */
- @Override public void onActivate(GridKernalContext ctx) throws IgniteCheckedException {
+ @Override public void onActivate(GridKernalContext ctx) {
if (log.isDebugEnabled())
- log.debug("Activate data structure processor [nodeId=" + ctx.localNodeId() +
+ log.debug("Activating data structure processor [nodeId=" + ctx.localNodeId() +
" topVer=" + ctx.discovery().topologyVersionEx() + " ]");
initFailed = false;
- initLatch = new CountDownLatch(1);
-
qryIdMap.clear();
ctx.event().addLocalEventListener(lsnr, EVT_NODE_LEFT, EVT_NODE_FAILED);
- onKernalStart0();
-
restoreStructuresState(ctx);
+
+ onKernalStart0();
}
/**
@@ -284,12 +291,14 @@ public final class DataStructuresProcessor extends GridProcessorAdapter implemen
@Override public void onDeActivate(GridKernalContext ctx) {
if (log.isDebugEnabled())
log.debug("DeActivate data structure processor [nodeId=" + ctx.localNodeId() +
- " topVer=" + ctx.discovery().topologyVersionEx() + " ]");
+ ", topVer=" + ctx.discovery().topologyVersionEx() + "]");
ctx.event().removeLocalEventListener(lsnr, EVT_NODE_LEFT, EVT_NODE_FAILED);
onKernalStop(false);
+ initLatch = null;
+
for (GridCacheRemovable v : dsMap.values()) {
if (v instanceof IgniteChangeGlobalStateSupport)
((IgniteChangeGlobalStateSupport)v).onDeActivate(ctx);
@@ -1070,9 +1079,14 @@ public final class DataStructuresProcessor extends GridProcessorAdapter implemen
* Awaits for processor initialization.
*/
private void awaitInitialization() {
- if (initLatch.getCount() > 0) {
+ CountDownLatch latch0 = initLatch;
+
+ if (latch0 == null)
+ throw new IllegalStateException("Ignite cluster is not active");
+
+ if (latch0.getCount() > 0) {
try {
- U.await(initLatch);
+ U.await(latch0);
if (initFailed)
throw new IllegalStateException("Failed to initialize data structures processor.");
http://git-wip-us.apache.org/repos/asf/ignite/blob/6f7aba85/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java
index f53d4e2..1fd14f8 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java
@@ -2071,7 +2071,7 @@ public class GridQueryProcessor extends GridProcessorAdapter {
if (qry.isLocal())
throw new IgniteException("Local query is not supported without specific cache.");
- if (!ctx.state().publicApiActiveState()) {
+ if (!ctx.state().publicApiActiveState(true)) {
throw new IgniteException("Can not perform the operation because the cluster is inactive. Note, that " +
"the cluster is considered inactive by default if Ignite Persistent Store is used to let all the nodes " +
"join the cluster. To activate the cluster call Ignite.active(true).");
http://git-wip-us.apache.org/repos/asf/ignite/blob/6f7aba85/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/handlers/cluster/GridChangeStateCommandHandler.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/handlers/cluster/GridChangeStateCommandHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/handlers/cluster/GridChangeStateCommandHandler.java
index 6236026..7bb13d9 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/handlers/cluster/GridChangeStateCommandHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/handlers/cluster/GridChangeStateCommandHandler.java
@@ -64,12 +64,12 @@ public class GridChangeStateCommandHandler extends GridRestCommandHandlerAdapter
try {
if (req.command().equals(CLUSTER_CURRENT_STATE)) {
- Boolean currentState = ctx.state().publicApiActiveState();
+ Boolean currentState = ctx.state().publicApiActiveState(false);
res.setResponse(currentState);
}
else
- ctx.state().changeGlobalState(req.active()).get();
+ ctx.grid().active(req.active());
fut.onDone(res);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6f7aba85/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 6df107b..f2b2781 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
@@ -238,9 +238,7 @@ public class GridServiceProcessor extends GridProcessorAdapter implements Ignite
if (ctx.deploy().enabled())
ctx.cache().context().deploy().ignoreOwnership(true);
- if (!ctx.clientNode()) {
- assert serviceCache.context().affinityNode();
-
+ if (!ctx.clientNode() && serviceCache.context().affinityNode()) {
serviceCache.context().continuousQueries().executeInternalQuery(
new ServiceEntriesListener(), null, true, true, false
);
@@ -1790,8 +1788,7 @@ public class GridServiceProcessor extends GridProcessorAdapter implements Ignite
try {
svcName.set(dep.configuration().getName());
- ctx.cache().internalCache(UTILITY_CACHE_NAME).context().affinity().
- affinityReadyFuture(topVer).get();
+ ctx.cache().context().exchange().affinityReadyFuture(topVer).get();
reassign(dep, topVer);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6f7aba85/modules/core/src/main/java/org/apache/ignite/internal/processors/subscription/GridInternalSubscriptionProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/subscription/GridInternalSubscriptionProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/subscription/GridInternalSubscriptionProcessor.java
new file mode 100644
index 0000000..6c0942a
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/subscription/GridInternalSubscriptionProcessor.java
@@ -0,0 +1,57 @@
+/*
+ * 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.subscription;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.ignite.internal.GridKernalContext;
+import org.apache.ignite.internal.processors.GridProcessorAdapter;
+import org.apache.ignite.internal.processors.cache.persistence.metastorage.MetastorageLifecycleListener;
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Processor enables grid components to register listeners for events
+ * generated by other components on local node.
+ *
+ * It starts very first during node startup procedure so any components could use it.
+ *
+ * As only local events are supported, no network communication occurs during notification process.
+ */
+public class GridInternalSubscriptionProcessor extends GridProcessorAdapter {
+ /** */
+ private List<MetastorageLifecycleListener> metastorageListeners = new ArrayList<>();
+
+ /**
+ * @param ctx Kernal context.
+ */
+ public GridInternalSubscriptionProcessor(GridKernalContext ctx) {
+ super(ctx);
+ }
+
+ /** */
+ public void registerMetastorageListener(@NotNull MetastorageLifecycleListener metastorageListener) {
+ if (metastorageListener == null)
+ throw new NullPointerException("Metastorage subscriber should be not-null.");
+
+ metastorageListeners.add(metastorageListener);
+ }
+
+ /** */
+ public List<MetastorageLifecycleListener> getMetastorageSubscribers() {
+ return metastorageListeners;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6f7aba85/modules/core/src/main/java/org/apache/ignite/internal/util/lang/GridFilteredClosableIterator.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/lang/GridFilteredClosableIterator.java b/modules/core/src/main/java/org/apache/ignite/internal/util/lang/GridFilteredClosableIterator.java
new file mode 100644
index 0000000..0ca795c
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/lang/GridFilteredClosableIterator.java
@@ -0,0 +1,37 @@
+/*
+ * 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.util.lang;
+
+import org.apache.ignite.IgniteCheckedException;
+
+/**
+ *
+ */
+public abstract class GridFilteredClosableIterator<T> extends GridFilteredIterator<T> implements AutoCloseable {
+ /**
+ * @param it Closeable iterator.
+ */
+ protected GridFilteredClosableIterator(GridCloseableIterator<? extends T> it) {
+ super(it);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void close() throws IgniteCheckedException {
+ ((GridCloseableIterator)it).close();
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6f7aba85/modules/core/src/main/java/org/apache/ignite/internal/util/lang/GridFilteredIterator.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/lang/GridFilteredIterator.java b/modules/core/src/main/java/org/apache/ignite/internal/util/lang/GridFilteredIterator.java
index 8f074d3..452bdd3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/lang/GridFilteredIterator.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/lang/GridFilteredIterator.java
@@ -25,7 +25,7 @@ import java.util.NoSuchElementException;
*/
public abstract class GridFilteredIterator<T> implements Iterator<T> {
/** */
- private final Iterator<? extends T> it;
+ protected final Iterator<? extends T> it;
/** */
private boolean hasNext;
http://git-wip-us.apache.org/repos/asf/ignite/blob/6f7aba85/modules/core/src/main/java/org/apache/ignite/internal/util/lang/GridFunc.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/lang/GridFunc.java b/modules/core/src/main/java/org/apache/ignite/internal/util/lang/GridFunc.java
index 24b9d48..1c5a69a 100755
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/lang/GridFunc.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/lang/GridFunc.java
@@ -168,6 +168,13 @@ public class GridFunc {
/** */
private static final IgniteClosure<ClusterNode, UUID> NODE2ID = new ClusterNodeGetIdClosure();
+ /** */
+ private static final IgniteClosure<ClusterNode, Object> NODE2CONSISTENTID = new IgniteClosure<ClusterNode, Object>() {
+ @Override public Object apply(ClusterNode node) {
+ return node.consistentId();
+ }
+ };
+
/**
* Gets predicate that evaluates to {@code true} only for given local node ID.
*
@@ -324,6 +331,23 @@ public class GridFunc {
}
/**
+ * Convenient utility method that returns collection of node consistent IDs for a given
+ * collection of grid nodes.
+ * <p>
+ * Note that this method doesn't create a new collection but simply iterates
+ * over the input one.
+ *
+ * @param nodes Collection of grid nodes.
+ * @return Collection of node consistent IDs for given collection of grid nodes.
+ */
+ public static Collection<Object> nodeConsistentIds(@Nullable Collection<? extends ClusterNode> nodes) {
+ if (nodes == null || nodes.isEmpty())
+ return Collections.emptyList();
+
+ return F.viewReadOnly(nodes, NODE2CONSISTENTID);
+ }
+
+ /**
* Gets random value from given collection.
*
* @param c Input collection (no {@code null} and not emtpy).
http://git-wip-us.apache.org/repos/asf/ignite/blob/6f7aba85/modules/core/src/main/java/org/apache/ignite/internal/visor/baseline/VisorBaselineNode.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/visor/baseline/VisorBaselineNode.java b/modules/core/src/main/java/org/apache/ignite/internal/visor/baseline/VisorBaselineNode.java
new file mode 100644
index 0000000..b37a364
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/visor/baseline/VisorBaselineNode.java
@@ -0,0 +1,89 @@
+/*
+ * 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.visor.baseline;
+
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
+import java.util.Map;
+import org.apache.ignite.cluster.BaselineNode;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.internal.visor.VisorDataTransferObject;
+
+/**
+ * Data transfer object for {@link BaselineNode}.
+ */
+public class VisorBaselineNode extends VisorDataTransferObject {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** */
+ private String consistentId;
+
+ /** */
+ private Map<String, Object> attrs;
+
+ /**
+ * Default constructor.
+ */
+ public VisorBaselineNode() {
+ // No-op.
+ }
+
+ /**
+ * Create data transfer object for baseline node.
+ *
+ * @param node Baseline node.
+ */
+ public VisorBaselineNode(BaselineNode node) {
+ consistentId = String.valueOf(node.consistentId());
+ attrs = node.attributes();
+ }
+
+ /**
+ * @return Node consistent ID.
+ */
+ public String getConsistentId() {
+ return consistentId;
+ }
+
+ /**
+ * @return Node attributes.
+ */
+ public Map<String, Object> getAttributes() {
+ return attrs;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void writeExternalData(ObjectOutput out) throws IOException {
+ U.writeString(out, consistentId);
+ U.writeMap(out, attrs);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void readExternalData(byte protoVer, ObjectInput in) throws IOException, ClassNotFoundException {
+ consistentId = U.readString(in);
+ attrs = U.readMap(in);
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(VisorBaselineNode.class, this);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6f7aba85/modules/core/src/main/java/org/apache/ignite/internal/visor/baseline/VisorBaselineOperation.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/visor/baseline/VisorBaselineOperation.java b/modules/core/src/main/java/org/apache/ignite/internal/visor/baseline/VisorBaselineOperation.java
new file mode 100644
index 0000000..b422417
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/visor/baseline/VisorBaselineOperation.java
@@ -0,0 +1,49 @@
+/*
+ * 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.visor.baseline;
+
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Baseline operation types.
+ **/
+public enum VisorBaselineOperation {
+ /** Collect baseline information. */
+ COLLECT,
+ /** Add nodes to baseline. */
+ ADD,
+ /** Remove nodes from baseline. */
+ REMOVE,
+ /** Set new baseline. */
+ SET,
+ /** Set baseline for topology version. */
+ VERSION;
+
+ /** Enumerated values. */
+ private static final VisorBaselineOperation[] VALS = values();
+
+ /**
+ * Efficiently gets enumerated value from its ordinal.
+ *
+ * @param ord Ordinal value.
+ * @return Enumerated value or {@code null} if ordinal out of range.
+ */
+ @Nullable public static VisorBaselineOperation fromOrdinal(int ord) {
+ return ord >= 0 && ord < VALS.length ? VALS[ord] : null;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6f7aba85/modules/core/src/main/java/org/apache/ignite/internal/visor/baseline/VisorBaselineTask.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/visor/baseline/VisorBaselineTask.java b/modules/core/src/main/java/org/apache/ignite/internal/visor/baseline/VisorBaselineTask.java
new file mode 100644
index 0000000..56c2dd9
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/visor/baseline/VisorBaselineTask.java
@@ -0,0 +1,224 @@
+/*
+ * 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.visor.baseline;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.cluster.BaselineNode;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.internal.cluster.IgniteClusterEx;
+import org.apache.ignite.internal.processors.task.GridInternal;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.internal.visor.VisorJob;
+import org.apache.ignite.internal.visor.VisorOneNodeTask;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Task that will collect baseline topology information.
+ */
+@GridInternal
+public class VisorBaselineTask extends VisorOneNodeTask<VisorBaselineTaskArg, VisorBaselineTaskResult> {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** {@inheritDoc} */
+ @Override protected VisorBaselineJob job(VisorBaselineTaskArg arg) {
+ return new VisorBaselineJob(arg, debug);
+ }
+
+ /**
+ * Job that will collect baseline topology information.
+ */
+ private static class VisorBaselineJob extends VisorJob<VisorBaselineTaskArg, VisorBaselineTaskResult> {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /**
+ * @param arg Formal job argument.
+ * @param debug Debug flag.
+ */
+ private VisorBaselineJob(VisorBaselineTaskArg arg, boolean debug) {
+ super(arg, debug);
+ }
+
+ /**
+ * Collect baseline and server nodes.
+ *
+ * @return Baseline descriptor.
+ */
+ private VisorBaselineTaskResult collect() {
+ IgniteClusterEx cluster = ignite.cluster();
+
+ Collection<? extends BaselineNode> baseline = cluster.currentBaselineTopology();
+
+ Collection<? extends BaselineNode> servers = cluster.forServers().nodes();
+
+ return new VisorBaselineTaskResult(ignite.active(), cluster.topologyVersion(),
+ F.isEmpty(baseline) ? null : baseline, servers);
+ }
+
+ /**
+ * Set new baseline.
+ *
+ * @param baselineTop Collection of baseline node.
+ * @return Baseline descriptor.
+ */
+ private VisorBaselineTaskResult set0(Collection<BaselineNode> baselineTop) {
+ ignite.cluster().setBaselineTopology(baselineTop);
+
+ return collect();
+ }
+
+ /**
+ * @return Current baseline.
+ */
+ private Map<String, BaselineNode> currentBaseLine() {
+ Collection<BaselineNode> baseline = ignite.cluster().currentBaselineTopology();
+
+ Map<String, BaselineNode> nodes = new HashMap<>();
+
+ for (BaselineNode node : baseline)
+ nodes.put(node.consistentId().toString(), node);
+
+ return nodes;
+ }
+
+ /**
+ * @return Current server nodes.
+ */
+ private Map<String, BaselineNode> currentServers() {
+ Map<String, BaselineNode> nodes = new HashMap<>();
+
+ for (ClusterNode node : ignite.cluster().forServers().nodes())
+ nodes.put(node.consistentId().toString(), node);
+
+ return nodes;
+ }
+
+ /**
+ * Set new baseline.
+ *
+ * @param consistentIds Collection of consistent IDs to set.
+ * @return New baseline.
+ */
+ private VisorBaselineTaskResult set(List<String> consistentIds) {
+ Map<String, BaselineNode> servers = currentServers();
+
+ Collection<BaselineNode> baselineTop = new ArrayList<>();
+
+ for (String consistentId : consistentIds) {
+ BaselineNode node = servers.get(consistentId);
+
+ if (node == null)
+ throw new IllegalStateException("Node not found for consistent ID: " + consistentId);
+
+ baselineTop.add(node);
+ }
+
+ return set0(baselineTop);
+ }
+
+ /**
+ * Add new nodes to baseline.
+ *
+ * @param consistentIds Collection of consistent IDs to add.
+ * @return New baseline.
+ */
+ private VisorBaselineTaskResult add(List<String> consistentIds) {
+ Map<String, BaselineNode> baseline = currentBaseLine();
+ Map<String, BaselineNode> servers = currentServers();
+
+ for (String consistentId : consistentIds) {
+ BaselineNode node = servers.get(consistentId);
+
+ if (node == null)
+ throw new IllegalStateException("Node not found for consistent ID: " + consistentId);
+
+ baseline.put(consistentId, node);
+ }
+
+ return set0(baseline.values());
+ }
+
+ /**
+ * Remove nodes from baseline.
+ *
+ * @param consistentIds Collection of consistent IDs to remove.
+ * @return New baseline.
+ */
+ private VisorBaselineTaskResult remove(List<String> consistentIds) {
+ Map<String, BaselineNode> baseline = currentBaseLine();
+
+ for (String consistentId : consistentIds) {
+ BaselineNode node = baseline.remove(consistentId);
+
+ if (node == null)
+ throw new IllegalStateException("Node not found for consistent ID: " + consistentId);
+ }
+
+ return set0(baseline.values());
+ }
+
+ /**
+ * Set baseline by topology version.
+ *
+ * @param targetVer Target topology version.
+ * @return New baseline.
+ */
+ private VisorBaselineTaskResult version(long targetVer) {
+ IgniteClusterEx cluster = ignite.cluster();
+
+ if (targetVer > cluster.topologyVersion())
+ throw new IllegalArgumentException("Topology version is ahead of time: " + targetVer);
+
+ cluster.setBaselineTopology(targetVer);
+
+ return collect();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected VisorBaselineTaskResult run(@Nullable VisorBaselineTaskArg arg) throws IgniteException {
+ switch (arg.getOperation()) {
+ case ADD:
+ return add(arg.getConsistentIds());
+
+ case REMOVE:
+ return remove(arg.getConsistentIds());
+
+ case SET:
+ return set(arg.getConsistentIds());
+
+ case VERSION:
+ return version(arg.getTopologyVersion());
+
+ default:
+ return collect();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(VisorBaselineJob.class, this);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6f7aba85/modules/core/src/main/java/org/apache/ignite/internal/visor/baseline/VisorBaselineTaskArg.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/visor/baseline/VisorBaselineTaskArg.java b/modules/core/src/main/java/org/apache/ignite/internal/visor/baseline/VisorBaselineTaskArg.java
new file mode 100644
index 0000000..092161e
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/visor/baseline/VisorBaselineTaskArg.java
@@ -0,0 +1,99 @@
+/*
+ * 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.visor.baseline;
+
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
+import java.util.List;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.internal.visor.VisorDataTransferObject;
+
+/**
+ * Argument for {@link VisorBaselineTask}.
+ */
+public class VisorBaselineTaskArg extends VisorDataTransferObject {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** */
+ private VisorBaselineOperation op;
+
+ /** */
+ private long topVer;
+
+ /** */
+ private List<String> consistentIds;
+
+ /**
+ * Default constructor.
+ */
+ public VisorBaselineTaskArg() {
+ // No-op.
+ }
+
+ /**
+ * @param topVer Topology version.
+ */
+ public VisorBaselineTaskArg(VisorBaselineOperation op, long topVer, List<String> consistentIds) {
+ this.op = op;
+ this.topVer = topVer;
+ this.consistentIds = consistentIds;
+ }
+
+ /**
+ * @return Base line operation.
+ */
+ public VisorBaselineOperation getOperation() {
+ return op;
+ }
+
+ /**
+ * @return Topology version.
+ */
+ public long getTopologyVersion() {
+ return topVer;
+ }
+
+ /**
+ * @return Consistent IDs.
+ */
+ public List<String> getConsistentIds() {
+ return consistentIds;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void writeExternalData(ObjectOutput out) throws IOException {
+ U.writeEnum(out, op);
+ out.writeLong(topVer);
+ U.writeCollection(out, consistentIds);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void readExternalData(byte protoVer, ObjectInput in) throws IOException, ClassNotFoundException {
+ op = VisorBaselineOperation.fromOrdinal(in.readByte());
+ topVer = in.readLong();
+ consistentIds = U.readList(in);
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(VisorBaselineTaskArg.class, this);
+ }
+}