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/04/13 09:33:22 UTC
[10/54] [abbrv] ignite git commit: IGNITE-7222 Added ZooKeeper
discovery SPI
http://git-wip-us.apache.org/repos/asf/ignite/blob/a64b941d/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java
index 01a5a71..cebe4b1 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java
@@ -56,6 +56,8 @@ import org.apache.ignite.internal.managers.deployment.GridDeployment;
import org.apache.ignite.internal.managers.deployment.GridDeploymentInfo;
import org.apache.ignite.internal.managers.deployment.GridDeploymentInfoBean;
import org.apache.ignite.internal.managers.discovery.CustomEventListener;
+import org.apache.ignite.internal.managers.discovery.DiscoCache;
+import org.apache.ignite.internal.managers.discovery.DiscoveryMessageResultsCollector;
import org.apache.ignite.internal.managers.eventstorage.GridLocalEventListener;
import org.apache.ignite.internal.managers.eventstorage.HighPriorityListener;
import org.apache.ignite.internal.processors.GridProcessorAdapter;
@@ -63,6 +65,7 @@ import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.GridCacheAdapter;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheProcessor;
+import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.CachePartitionPartialCountersMap;
import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryHandler;
import org.apache.ignite.internal.processors.timeout.GridTimeoutObject;
import org.apache.ignite.internal.util.future.GridFinishedFuture;
@@ -147,6 +150,12 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
/** Query sequence number for message topic. */
private final AtomicLong seq = new AtomicLong();
+ /** */
+ private ContinuousRoutinesInfo routinesInfo;
+
+ /** */
+ private int discoProtoVer;
+
/**
* @param ctx Kernal context.
*/
@@ -156,6 +165,11 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
/** {@inheritDoc} */
@Override public void start() throws IgniteCheckedException {
+ discoProtoVer = ctx.discovery().mutableCustomMessages() ? 1 : 2;
+
+ if (discoProtoVer == 2)
+ routinesInfo = new ContinuousRoutinesInfo();
+
if (ctx.config().isDaemon())
return;
@@ -177,6 +191,8 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
@Override public void onCustomEvent(AffinityTopologyVersion topVer,
ClusterNode snd,
StartRoutineDiscoveryMessage msg) {
+ assert discoProtoVer == 1 : discoProtoVer;
+
if (ctx.isStopping())
return;
@@ -184,6 +200,20 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
}
});
+ ctx.discovery().setCustomEventListener(StartRoutineDiscoveryMessageV2.class,
+ new CustomEventListener<StartRoutineDiscoveryMessageV2>() {
+ @Override public void onCustomEvent(AffinityTopologyVersion topVer,
+ ClusterNode snd,
+ StartRoutineDiscoveryMessageV2 msg) {
+ assert discoProtoVer == 2 : discoProtoVer;
+
+ if (ctx.isStopping())
+ return;
+
+ processStartRequestV2(topVer, snd, msg);
+ }
+ });
+
ctx.discovery().setCustomEventListener(StartRoutineAckDiscoveryMessage.class,
new CustomEventListener<StartRoutineAckDiscoveryMessage>() {
@Override public void onCustomEvent(AffinityTopologyVersion topVer,
@@ -201,6 +231,9 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
@Override public void onCustomEvent(AffinityTopologyVersion topVer,
ClusterNode snd,
StopRoutineDiscoveryMessage msg) {
+ if (discoProtoVer == 2)
+ routinesInfo.removeRoutine(msg.routineId);
+
if (ctx.isStopping())
return;
@@ -222,32 +255,36 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
ctx.io().addMessageListener(TOPIC_CONTINUOUS, new GridMessageListener() {
@Override public void onMessage(UUID nodeId, Object obj, byte plc) {
- GridContinuousMessage msg = (GridContinuousMessage)obj;
+ if (obj instanceof ContinuousRoutineStartResultMessage)
+ processRoutineStartResultMessage(nodeId, (ContinuousRoutineStartResultMessage)obj);
+ else {
+ GridContinuousMessage msg = (GridContinuousMessage)obj;
- if (msg.data() == null && msg.dataBytes() != null) {
- try {
- msg.data(U.unmarshal(marsh, msg.dataBytes(), U.resolveClassLoader(ctx.config())));
- }
- catch (IgniteCheckedException e) {
- U.error(log, "Failed to process message (ignoring): " + msg, e);
+ if (msg.data() == null && msg.dataBytes() != null) {
+ try {
+ msg.data(U.unmarshal(marsh, msg.dataBytes(), U.resolveClassLoader(ctx.config())));
+ }
+ catch (IgniteCheckedException e) {
+ U.error(log, "Failed to process message (ignoring): " + msg, e);
- return;
+ return;
+ }
}
- }
- switch (msg.type()) {
- case MSG_EVT_NOTIFICATION:
- processNotification(nodeId, msg);
+ switch (msg.type()) {
+ case MSG_EVT_NOTIFICATION:
+ processNotification(nodeId, msg);
- break;
+ break;
- case MSG_EVT_ACK:
- processMessageAck(msg);
+ case MSG_EVT_ACK:
+ processMessageAck(msg);
- break;
+ break;
- default:
- assert false : "Unexpected message received: " + msg.type();
+ default:
+ assert false : "Unexpected message received: " + msg.type();
+ }
}
}
});
@@ -341,6 +378,15 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
/** {@inheritDoc} */
@Override public void collectJoiningNodeData(DiscoveryDataBag dataBag) {
+ if (ctx.isDaemon())
+ return;
+
+ if (discoProtoVer == 2) {
+ routinesInfo.collectJoiningNodeData(dataBag);
+
+ return;
+ }
+
Serializable data = getDiscoveryData(dataBag.joiningNodeId());
if (data != null)
@@ -349,6 +395,15 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
/** {@inheritDoc} */
@Override public void collectGridNodeData(DiscoveryDataBag dataBag) {
+ if (ctx.isDaemon())
+ return;
+
+ if (discoProtoVer == 2) {
+ routinesInfo.collectGridNodeData(dataBag);
+
+ return;
+ }
+
Serializable data = getDiscoveryData(dataBag.joiningNodeId());
if (data != null)
@@ -393,6 +448,7 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
return data;
}
+
return null;
}
@@ -430,22 +486,118 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
@Override public void onJoiningNodeDataReceived(JoiningNodeDiscoveryData data) {
if (log.isDebugEnabled()) {
log.debug("onJoiningNodeDataReceived [joining=" + data.joiningNodeId() +
- ", loc=" + ctx.localNodeId() +
- ", data=" + data.joiningNodeData() +
- ']');
+ ", loc=" + ctx.localNodeId() +
+ ", data=" + data.joiningNodeData() +
+ ']');
}
- if (data.hasJoiningNodeData())
- onDiscoDataReceived((DiscoveryData) data.joiningNodeData());
+ if (discoProtoVer == 2) {
+ if (data.hasJoiningNodeData()) {
+ ContinuousRoutinesJoiningNodeDiscoveryData nodeData = (ContinuousRoutinesJoiningNodeDiscoveryData)
+ data.joiningNodeData();
+
+ for (ContinuousRoutineInfo routineInfo : nodeData.startedRoutines) {
+ routinesInfo.addRoutineInfo(routineInfo);
+
+ startDiscoveryDataRoutine(routineInfo);
+ }
+ }
+ }
+ else {
+ if (data.hasJoiningNodeData())
+ onDiscoDataReceived((DiscoveryData) data.joiningNodeData());
+ }
}
/** {@inheritDoc} */
@Override public void onGridDataReceived(GridDiscoveryData data) {
- Map<UUID, Serializable> nodeSpecData = data.nodeSpecificData();
+ if (discoProtoVer == 2) {
+ if (ctx.isDaemon())
+ return;
+
+ if (data.commonData() != null) {
+ ContinuousRoutinesCommonDiscoveryData commonData =
+ (ContinuousRoutinesCommonDiscoveryData)data.commonData();
+
+ for (ContinuousRoutineInfo routineInfo : commonData.startedRoutines) {
+ if (routinesInfo.routineExists(routineInfo.routineId))
+ continue;
+
+ routinesInfo.addRoutineInfo(routineInfo);
+
+ startDiscoveryDataRoutine(routineInfo);
+ }
+ }
+ }
+ else {
+ Map<UUID, Serializable> nodeSpecData = data.nodeSpecificData();
+
+ if (nodeSpecData != null) {
+ for (Map.Entry<UUID, Serializable> e : nodeSpecData.entrySet())
+ onDiscoDataReceived((DiscoveryData) e.getValue());
+ }
+ }
+ }
+
+ /**
+ * @param routineInfo Routine info.
+ */
+ private void startDiscoveryDataRoutine(ContinuousRoutineInfo routineInfo) {
+ IgnitePredicate<ClusterNode> nodeFilter = null;
+
+ try {
+ if (routineInfo.nodeFilter != null) {
+ nodeFilter = U.unmarshal(marsh, routineInfo.nodeFilter, U.resolveClassLoader(ctx.config()));
+
+ ctx.resource().injectGeneric(nodeFilter);
+ }
+ }
+ catch (IgniteCheckedException e) {
+ U.error(log, "Failed to unmarshal continuous routine filter, ignore routine [" +
+ "routineId=" + routineInfo.routineId +
+ ", srcNodeId=" + routineInfo.srcNodeId + ']', e);
+
+ return;
+ }
+
+ if (nodeFilter == null || nodeFilter.apply(ctx.discovery().localNode())) {
+ GridContinuousHandler hnd;
- if (nodeSpecData != null) {
- for (Map.Entry<UUID, Serializable> e : nodeSpecData.entrySet())
- onDiscoDataReceived((DiscoveryData) e.getValue());
+ try {
+ hnd = U.unmarshal(marsh, routineInfo.hnd, U.resolveClassLoader(ctx.config()));
+
+ if (ctx.config().isPeerClassLoadingEnabled())
+ hnd.p2pUnmarshal(routineInfo.srcNodeId, ctx);
+ }
+ catch (IgniteCheckedException e) {
+ U.error(log, "Failed to unmarshal continuous routine handler, ignore routine [" +
+ "routineId=" + routineInfo.routineId +
+ ", srcNodeId=" + routineInfo.srcNodeId + ']', e);
+
+ return;
+ }
+
+ try {
+ registerHandler(routineInfo.srcNodeId,
+ routineInfo.routineId,
+ hnd,
+ routineInfo.bufSize,
+ routineInfo.interval,
+ routineInfo.autoUnsubscribe,
+ false);
+ }
+ catch (IgniteCheckedException e) {
+ U.error(log, "Failed to register continuous routine handler, ignore routine [" +
+ "routineId=" + routineInfo.routineId +
+ ", srcNodeId=" + routineInfo.srcNodeId + ']', e);
+ }
+ }
+ else {
+ if (log.isDebugEnabled()) {
+ log.debug("Do not register continuous routine, rejected by node filter [" +
+ "routineId=" + routineInfo.routineId +
+ ", srcNodeId=" + routineInfo.srcNodeId + ']');
+ }
}
}
@@ -564,13 +716,14 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
* @param rmtFilter Remote filter.
* @param prjPred Projection predicate.
* @return Routine ID.
+ * @throws IgniteCheckedException If failed.
*/
@SuppressWarnings("unchecked")
public UUID registerStaticRoutine(
String cacheName,
CacheEntryUpdatedListener<?, ?> locLsnr,
CacheEntryEventSerializableFilter rmtFilter,
- @Nullable IgnitePredicate<ClusterNode> prjPred) {
+ @Nullable IgnitePredicate<ClusterNode> prjPred) throws IgniteCheckedException {
String topicPrefix = "CONTINUOUS_QUERY_STATIC" + "_" + cacheName;
CacheContinuousQueryHandler hnd = new CacheContinuousQueryHandler(
@@ -589,6 +742,17 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
LocalRoutineInfo routineInfo = new LocalRoutineInfo(prjPred, hnd, 1, 0, true);
+ if (discoProtoVer == 2) {
+ routinesInfo.addRoutineInfo(createRoutineInfo(
+ ctx.localNodeId(),
+ routineId,
+ hnd,
+ prjPred,
+ routineInfo.bufSize,
+ routineInfo.interval,
+ routineInfo.autoUnsubscribe));
+ }
+
locInfos.put(routineId, routineInfo);
registerMessageListener(hnd);
@@ -597,6 +761,40 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
}
/**
+ * @param srcNodeId Source node ID.
+ * @param routineId Routine ID.
+ * @param hnd Handler.
+ * @param nodeFilter Node filter.
+ * @param bufSize Handler buffer size.
+ * @param interval Time interval.
+ * @param autoUnsubscribe Auto unsubscribe flag.
+ * @return Routine info instance.
+ * @throws IgniteCheckedException If failed.
+ */
+ private ContinuousRoutineInfo createRoutineInfo(
+ UUID srcNodeId,
+ UUID routineId,
+ GridContinuousHandler hnd,
+ @Nullable IgnitePredicate<ClusterNode> nodeFilter,
+ int bufSize,
+ long interval,
+ boolean autoUnsubscribe)
+ throws IgniteCheckedException {
+ byte[] hndBytes = marsh.marshal(hnd);
+
+ byte[] filterBytes = nodeFilter != null ? marsh.marshal(nodeFilter) : null;
+
+ return new ContinuousRoutineInfo(
+ srcNodeId,
+ routineId,
+ hndBytes,
+ filterBytes,
+ bufSize,
+ interval,
+ autoUnsubscribe);
+ }
+
+ /**
* @param hnd Handler.
* @param bufSize Buffer size.
* @param interval Time interval.
@@ -638,30 +836,10 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
// Whether local node is included in routine.
boolean locIncluded = prjPred == null || prjPred.apply(ctx.discovery().localNode());
- StartRequestData reqData = new StartRequestData(prjPred, hnd.clone(), bufSize, interval, autoUnsubscribe);
+ AbstractContinuousMessage msg;
try {
- if (ctx.config().isPeerClassLoadingEnabled()) {
- // Handle peer deployment for projection predicate.
- if (prjPred != null && !U.isGrid(prjPred.getClass())) {
- Class cls = U.detectClass(prjPred);
-
- String clsName = cls.getName();
-
- GridDeployment dep = ctx.deploy().deploy(cls, U.detectClassLoader(cls));
-
- if (dep == null)
- throw new IgniteDeploymentCheckedException("Failed to deploy projection predicate: " + prjPred);
-
- reqData.className(clsName);
- reqData.deploymentInfo(new GridDeploymentInfoBean(dep));
-
- reqData.p2pMarshal(marsh);
- }
-
- // Handle peer deployment for other handler-specific objects.
- reqData.handler().p2pMarshal(ctx);
- }
+ msg = createStartMessage(routineId, hnd, bufSize, interval, autoUnsubscribe, prjPred);
}
catch (IgniteCheckedException e) {
return new GridFinishedFuture<>(e);
@@ -674,20 +852,26 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
return new GridFinishedFuture<>(new NodeStoppingException("Failed to start continuous query (node is stopping)"));
try {
- StartFuture fut = new StartFuture(ctx, routineId);
+ StartFuture fut = new StartFuture(routineId);
startFuts.put(routineId, fut);
try {
- if (locIncluded || hnd.isQuery())
- registerHandler(ctx.localNodeId(), routineId, hnd, bufSize, interval, autoUnsubscribe, true);
+ if (locIncluded || hnd.isQuery()) {
+ registerHandler(ctx.localNodeId(),
+ routineId,
+ hnd,
+ bufSize,
+ interval,
+ autoUnsubscribe,
+ true);
+ }
- ctx.discovery().sendCustomEvent(new StartRoutineDiscoveryMessage(routineId, reqData,
- reqData.handler().keepBinary()));
- }
- catch (IgniteCheckedException e) {
- startFuts.remove(routineId);
- locInfos.remove(routineId);
+ ctx.discovery().sendCustomEvent(msg);
+ }
+ catch (IgniteCheckedException e) {
+ startFuts.remove(routineId);
+ locInfos.remove(routineId);
unregisterHandler(routineId, hnd, true);
@@ -707,6 +891,92 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
}
/**
+ * @param routineId Routine ID.
+ * @param hnd Handler.
+ * @param bufSize Buffer size.
+ * @param interval Interval.
+ * @param autoUnsubscribe Auto unsubscribe flag.
+ * @param nodeFilter Node filter.
+ * @return Routine start message.
+ * @throws IgniteCheckedException If failed.
+ */
+ private AbstractContinuousMessage createStartMessage(UUID routineId,
+ GridContinuousHandler hnd,
+ int bufSize,
+ long interval,
+ boolean autoUnsubscribe,
+ @Nullable IgnitePredicate<ClusterNode> nodeFilter)
+ throws IgniteCheckedException
+ {
+ hnd = hnd.clone();
+
+ String clsName = null;
+ GridDeploymentInfoBean dep = null;
+
+ if (ctx.config().isPeerClassLoadingEnabled()) {
+ // Handle peer deployment for projection predicate.
+ if (nodeFilter != null && !U.isGrid(nodeFilter.getClass())) {
+ Class cls = U.detectClass(nodeFilter);
+
+ clsName = cls.getName();
+
+ GridDeployment dep0 = ctx.deploy().deploy(cls, U.detectClassLoader(cls));
+
+ if (dep0 == null)
+ throw new IgniteDeploymentCheckedException("Failed to deploy projection predicate: " + nodeFilter);
+
+ dep = new GridDeploymentInfoBean(dep0);
+ }
+
+ // Handle peer deployment for other handler-specific objects.
+ hnd.p2pMarshal(ctx);
+ }
+
+ if (discoProtoVer == 1) {
+ StartRequestData reqData = new StartRequestData(
+ nodeFilter,
+ hnd,
+ bufSize,
+ interval,
+ autoUnsubscribe);
+
+ if (clsName != null) {
+ reqData.className(clsName);
+ reqData.deploymentInfo(dep);
+
+ reqData.p2pMarshal(marsh);
+ }
+
+ return new StartRoutineDiscoveryMessage(
+ routineId,
+ reqData,
+ reqData.handler().keepBinary());
+ }
+ else {
+ assert discoProtoVer == 2 : discoProtoVer;
+
+ byte[] nodeFilterBytes = nodeFilter != null ? U.marshal(marsh, nodeFilter) : null;
+ byte[] hndBytes = U.marshal(marsh, hnd);
+
+ StartRequestDataV2 reqData = new StartRequestDataV2(nodeFilterBytes,
+ hndBytes,
+ bufSize,
+ interval,
+ autoUnsubscribe);
+
+ if (clsName != null) {
+ reqData.className(clsName);
+ reqData.deploymentInfo(dep);
+ }
+
+ return new StartRoutineDiscoveryMessageV2(
+ routineId,
+ reqData,
+ hnd.keepBinary());
+ }
+ }
+
+ /**
* @param hnd Handler.
*/
private void registerMessageListener(GridContinuousHandler hnd) {
@@ -760,29 +1030,38 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
doStop = true;
}
- if (doStop) {
- // Unregister routine locally.
- LocalRoutineInfo routine = locInfos.remove(routineId);
-
- // Finish if routine is not found (wrong ID is provided).
- if (routine == null) {
- stopFuts.remove(routineId);
+ if (doStop) {
+ boolean stop = false;
- fut.onDone();
+ // Unregister routine locally.
+ LocalRoutineInfo routine = locInfos.remove(routineId);
- return fut;
- }
+ if (routine != null) {
+ stop = true;
// Unregister handler locally.
unregisterHandler(routineId, routine.hnd, true);
+ }
- try {
- ctx.discovery().sendCustomEvent(new StopRoutineDiscoveryMessage(routineId));
- }
- catch (IgniteCheckedException e) {
- fut.onDone(e);
+ if (!stop && discoProtoVer == 2)
+ stop = routinesInfo.routineExists(routineId);
+
+ // Finish if routine is not found (wrong ID is provided).
+ if (!stop) {
+ stopFuts.remove(routineId);
+
+ fut.onDone();
+
+ return fut;
}
+ try {
+ ctx.discovery().sendCustomEvent(new StopRoutineDiscoveryMessage(routineId));
+ }
+ catch (IgniteCheckedException e) {
+ fut.onDone(e);
+ }
+
if (ctx.isStopping())
fut.onDone();
}
@@ -924,6 +1203,9 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
clientInfos.clear();
+ if (discoProtoVer == 2)
+ routinesInfo.onClientDisconnected(locInfos.keySet());
+
if (log.isDebugEnabled()) {
log.debug("after onDisconnected [rmtInfos=" + rmtInfos +
", locInfos=" + locInfos +
@@ -996,35 +1278,11 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
StartFuture fut = startFuts.remove(msg.routineId());
if (fut != null) {
- if (msg.errs().isEmpty()) {
- LocalRoutineInfo routine = locInfos.get(msg.routineId());
-
- // Update partition counters.
- if (routine != null && routine.handler().isQuery()) {
- Map<UUID, Map<Integer, T2<Long, Long>>> cntrsPerNode = msg.updateCountersPerNode();
- Map<Integer, T2<Long, Long>> cntrs = msg.updateCounters();
-
- GridCacheAdapter<Object, Object> interCache =
- ctx.cache().internalCache(routine.handler().cacheName());
-
- GridCacheContext cctx = interCache != null ? interCache.context() : null;
-
- if (cctx != null && cntrsPerNode != null && !cctx.isLocal() && cctx.affinityNode())
- cntrsPerNode.put(ctx.localNodeId(),
- toCountersMap(cctx.topology().localUpdateCounters(false)));
-
- routine.handler().updateCounters(topVer, cntrsPerNode, cntrs);
- }
-
- fut.onRemoteRegistered();
- }
- else {
- IgniteCheckedException firstEx = F.first(msg.errs().values());
-
- fut.onDone(firstEx);
-
- stopRoutine(msg.routineId());
- }
+ fut.onAllRemoteRegistered(
+ topVer,
+ msg.errs(),
+ msg.updateCountersPerNode(),
+ msg.updateCounters());
}
}
@@ -1138,6 +1396,199 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
}
/**
+ * @param sndId Sender node ID.
+ * @param msg Message.
+ */
+ private void processRoutineStartResultMessage(UUID sndId, ContinuousRoutineStartResultMessage msg) {
+ StartFuture fut = startFuts.get(msg.routineId());
+
+ if (fut != null)
+ fut.onResult(sndId, msg);
+ }
+
+ /**
+ * @param topVer Current topology version.
+ * @param snd Sender.
+ * @param msg Start request.
+ */
+ private void processStartRequestV2(final AffinityTopologyVersion topVer,
+ final ClusterNode snd,
+ final StartRoutineDiscoveryMessageV2 msg) {
+ StartRequestDataV2 reqData = msg.startRequestData();
+
+ ContinuousRoutineInfo routineInfo = new ContinuousRoutineInfo(snd.id(),
+ msg.routineId(),
+ reqData.handlerBytes(),
+ reqData.nodeFilterBytes(),
+ reqData.bufferSize(),
+ reqData.interval(),
+ reqData.autoUnsubscribe());
+
+ routinesInfo.addRoutineInfo(routineInfo);
+
+ final DiscoCache discoCache = ctx.discovery().discoCache(topVer);
+
+ // Should not use marshaller and send messages from discovery thread.
+ ctx.getSystemExecutorService().execute(new Runnable() {
+ @Override public void run() {
+ if (snd.id().equals(ctx.localNodeId())) {
+ StartFuture fut = startFuts.get(msg.routineId());
+
+ if (fut != null)
+ fut.initRemoteNodes(discoCache);
+
+ return;
+ }
+
+ StartRequestDataV2 reqData = msg.startRequestData();
+
+ Exception err = null;
+
+ IgnitePredicate<ClusterNode> nodeFilter = null;
+
+ byte[] cntrs = null;
+
+ if (reqData.nodeFilterBytes() != null) {
+ try {
+ if (ctx.config().isPeerClassLoadingEnabled() && reqData.className() != null) {
+ String clsName = reqData.className();
+ GridDeploymentInfo depInfo = reqData.deploymentInfo();
+
+ GridDeployment dep = ctx.deploy().getGlobalDeployment(depInfo.deployMode(),
+ clsName,
+ clsName,
+ depInfo.userVersion(),
+ snd.id(),
+ depInfo.classLoaderId(),
+ depInfo.participants(),
+ null);
+
+ if (dep == null) {
+ throw new IgniteDeploymentCheckedException("Failed to obtain deployment " +
+ "for class: " + clsName);
+ }
+
+ nodeFilter = U.unmarshal(marsh,
+ reqData.nodeFilterBytes(),
+ U.resolveClassLoader(dep.classLoader(), ctx.config()));
+ }
+ else {
+ nodeFilter = U.unmarshal(marsh,
+ reqData.nodeFilterBytes(),
+ U.resolveClassLoader(ctx.config()));
+ }
+
+ if (nodeFilter != null)
+ ctx.resource().injectGeneric(nodeFilter);
+ }
+ catch (Exception e) {
+ err = e;
+
+ U.error(log, "Failed to unmarshal continuous routine filter [" +
+ "routineId=" + msg.routineId +
+ ", srcNodeId=" + snd.id() + ']', e);
+ }
+ }
+
+ boolean register = err == null &&
+ (nodeFilter == null || nodeFilter.apply(ctx.discovery().localNode()));
+
+ if (register) {
+ try {
+ GridContinuousHandler hnd = U.unmarshal(marsh,
+ reqData.handlerBytes(),
+ U.resolveClassLoader(ctx.config()));
+
+ if (ctx.config().isPeerClassLoadingEnabled())
+ hnd.p2pUnmarshal(snd.id(), ctx);
+
+ if (msg.keepBinary()) {
+ assert hnd instanceof CacheContinuousQueryHandler : hnd;
+
+ ((CacheContinuousQueryHandler)hnd).keepBinary(true);
+ }
+
+ GridContinuousHandler hnd0 = hnd instanceof GridMessageListenHandler ?
+ new GridMessageListenHandler((GridMessageListenHandler)hnd) :
+ hnd;
+
+ registerHandler(snd.id(),
+ msg.routineId,
+ hnd0,
+ reqData.bufferSize(),
+ reqData.interval(),
+ reqData.autoUnsubscribe(),
+ false);
+
+ if (hnd0.isQuery()) {
+ GridCacheProcessor proc = ctx.cache();
+
+ if (proc != null) {
+ GridCacheAdapter cache = ctx.cache().internalCache(hnd0.cacheName());
+
+ if (cache != null && !cache.isLocal() && cache.context().userCache()) {
+ CachePartitionPartialCountersMap cntrsMap =
+ cache.context().topology().localUpdateCounters(false);
+
+ cntrs = U.marshal(marsh, cntrsMap);
+ }
+ }
+ }
+ }
+ catch (Exception e) {
+ err = e;
+
+ U.error(log, "Failed to register continuous routine handler [" +
+ "routineId=" + msg.routineId +
+ ", srcNodeId=" + snd.id() + ']', e);
+ }
+ }
+
+ sendMessageStartResult(snd, msg.routineId(), cntrs, err);
+ }
+ });
+ }
+
+ /**
+ * @param node Target node.
+ * @param routineId Routine ID.
+ * @param cntrsMapBytes Marshalled {@link CachePartitionPartialCountersMap}.
+ * @param err Start error if any.
+ */
+ private void sendMessageStartResult(final ClusterNode node,
+ final UUID routineId,
+ byte[] cntrsMapBytes,
+ final @Nullable Exception err)
+ {
+ byte[] errBytes = null;
+
+ if (err != null) {
+ try {
+ errBytes = U.marshal(marsh, err);
+ }
+ catch (Exception e) {
+ U.error(log, "Failed to marshal routine start error: " + e, e);
+ }
+ }
+
+ ContinuousRoutineStartResultMessage msg = new ContinuousRoutineStartResultMessage(routineId,
+ cntrsMapBytes,
+ errBytes,
+ err != null);
+
+ try {
+ ctx.io().sendToGridTopic(node, TOPIC_CONTINUOUS, msg, SYSTEM_POOL, null);
+ }
+ catch (ClusterTopologyCheckedException e) {
+ if (log.isDebugEnabled())
+ log.debug("Failed to send routine start result, node failed: " + e);
+ }
+ catch (IgniteCheckedException e) {
+ U.error(log, "Failed to send routine start result: " + e, e);
+ }
+ }
+
+ /**
* @param msg Message.
*/
private void processMessageAck(GridContinuousMessage msg) {
@@ -1455,6 +1906,13 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
UUID nodeId = ((DiscoveryEvent)evt).eventNode().id();
+ if (discoProtoVer == 2) {
+ routinesInfo.onNodeFail(nodeId);
+
+ for (StartFuture fut : startFuts.values())
+ fut.onNodeFail(nodeId);
+ }
+
clientInfos.remove(nodeId);
// Unregister handlers created by left node.
@@ -1894,10 +2352,7 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
/**
* Future for start routine.
*/
- private static class StartFuture extends GridFutureAdapter<UUID> {
- /** */
- private GridKernalContext ctx;
-
+ private class StartFuture extends GridFutureAdapter<UUID> {
/** Consume ID. */
private UUID routineId;
@@ -1907,56 +2362,170 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
/** All remote listeners are registered. */
private volatile boolean rmt;
- /** Timeout object. */
- private volatile GridTimeoutObject timeoutObj;
+ /** */
+ private final DiscoveryMessageResultsCollector<ContinuousRoutineStartResultMessage, RoutineRegisterResults>
+ resCollect;
/**
- * @param ctx Kernal context.
* @param routineId Consume ID.
*/
- StartFuture(GridKernalContext ctx, UUID routineId) {
- this.ctx = ctx;
-
+ StartFuture(UUID routineId) {
this.routineId = routineId;
+
+ resCollect = new DiscoveryMessageResultsCollector<ContinuousRoutineStartResultMessage, RoutineRegisterResults>(ctx) {
+ @Override protected RoutineRegisterResults createResult(Map<UUID, NodeMessage<ContinuousRoutineStartResultMessage>> rcvd) {
+ Map<UUID, Exception> errs = null;
+ Map<UUID, Map<Integer, T2<Long, Long>>> cntrsPerNode = null;
+
+ for (Map.Entry<UUID, NodeMessage<ContinuousRoutineStartResultMessage>> entry : rcvd.entrySet()) {
+ ContinuousRoutineStartResultMessage msg = entry.getValue().message();
+
+ if (msg == null)
+ continue;
+
+ if (msg.error()) {
+ byte[] errBytes = msg.errorBytes();
+
+ Exception err = null;
+
+ if (errBytes != null) {
+ try {
+ err = U.unmarshal(marsh, errBytes, U.resolveClassLoader(ctx.config()));
+ }
+ catch (Exception e) {
+ U.warn(log, "Failed to unmarhal continuous routine start error: " + e);
+ }
+ }
+
+ if (err == null) {
+ err = new IgniteCheckedException("Failed to start continuous " +
+ "routine on node: " + entry.getKey());
+ }
+
+ if (errs == null)
+ errs = new HashMap<>();
+
+ errs.put(entry.getKey(), err);
+ }
+ else {
+ byte[] cntrsMapBytes = msg.countersMapBytes();
+
+ if (cntrsMapBytes != null) {
+ try {
+ CachePartitionPartialCountersMap cntrsMap = U.unmarshal(
+ marsh,
+ cntrsMapBytes,
+ U.resolveClassLoader(ctx.config()));
+
+ if (cntrsPerNode == null)
+ cntrsPerNode = new HashMap<>();
+
+ cntrsPerNode.put(entry.getKey(), CachePartitionPartialCountersMap.toCountersMap(cntrsMap));
+ }
+ catch (Exception e) {
+ U.warn(log, "Failed to unmarhal continuous query update counters: " + e);
+ }
+ }
+ }
+ }
+
+ return new RoutineRegisterResults(discoCache.version(), errs, cntrsPerNode);
+ }
+
+ @Override protected void onResultsCollected(RoutineRegisterResults res0) {
+ onAllRemoteRegistered(res0.topVer, res0.errs, res0.cntrsPerNode, null);
+ }
+
+ @Override protected boolean waitForNode(DiscoCache discoCache, ClusterNode node) {
+ return !ctx.localNodeId().equals(node.id());
+ }
+ };
}
/**
- * Called when local listener is registered.
+ * @param topVer Topology version.
+ * @param errs Errors.
+ * @param cntrsPerNode Update counters.
+ * @param cntrs Update counters.
*/
- public void onLocalRegistered() {
- loc = true;
+ private void onAllRemoteRegistered(
+ AffinityTopologyVersion topVer,
+ @Nullable Map<UUID, ? extends Exception> errs,
+ Map<UUID, Map<Integer, T2<Long, Long>>> cntrsPerNode,
+ Map<Integer, T2<Long, Long>> cntrs) {
+ try {
+ if (errs == null || errs.isEmpty()) {
+ LocalRoutineInfo routine = locInfos.get(routineId);
- if (rmt && !isDone())
- onDone(routineId);
+ // Update partition counters.
+ if (routine != null && routine.handler().isQuery()) {
+ GridCacheAdapter<Object, Object> interCache =
+ ctx.cache().internalCache(routine.handler().cacheName());
+
+ GridCacheContext cctx = interCache != null ? interCache.context() : null;
+
+ if (cctx != null && cntrsPerNode != null && !cctx.isLocal() && cctx.affinityNode())
+ cntrsPerNode.put(ctx.localNodeId(),
+ toCountersMap(cctx.topology().localUpdateCounters(false)));
+
+ routine.handler().updateCounters(topVer, cntrsPerNode, cntrs);
+ }
+
+ onRemoteRegistered();
+ }
+ else {
+ Exception firstEx = F.first(errs.values());
+
+ onDone(firstEx);
+
+ stopRoutine(routineId);
+ }
+ }
+ finally {
+ startFuts.remove(routineId, this);
+ }
}
/**
- * Called when all remote listeners are registered.
+ * @param discoCache Discovery state.
*/
- public void onRemoteRegistered() {
- rmt = true;
+ void initRemoteNodes(DiscoCache discoCache) {
+ resCollect.init(discoCache);
+ }
- if (loc && !isDone())
- onDone(routineId);
+ /**
+ * @param nodeId Node ID.
+ * @param msg Message.
+ */
+ void onResult(UUID nodeId, ContinuousRoutineStartResultMessage msg) {
+ resCollect.onMessage(nodeId, msg);
}
/**
- * @param timeoutObj Timeout object.
+ * @param nodeId Failed node ID.
*/
- public void addTimeoutObject(GridTimeoutObject timeoutObj) {
- assert timeoutObj != null;
+ void onNodeFail(UUID nodeId) {
+ resCollect.onNodeFail(nodeId);
+ }
- this.timeoutObj = timeoutObj;
+ /**
+ * Called when local listener is registered.
+ */
+ void onLocalRegistered() {
+ loc = true;
- ctx.timeout().addTimeoutObject(timeoutObj);
+ if (rmt && !isDone())
+ onDone(routineId);
}
- /** {@inheritDoc} */
- @Override public boolean onDone(@Nullable UUID res, @Nullable Throwable err) {
- if (timeoutObj != null)
- ctx.timeout().removeTimeoutObject(timeoutObj);
+ /**
+ * Called when all remote listeners are registered.
+ */
+ void onRemoteRegistered() {
+ rmt = true;
- return super.onDone(res, err);
+ if (loc && !isDone())
+ onDone(routineId);
}
/** {@inheritDoc} */
@@ -1966,6 +2535,33 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
}
/**
+ *
+ */
+ private static class RoutineRegisterResults {
+ /** */
+ private final AffinityTopologyVersion topVer;
+
+ /** */
+ private final Map<UUID, ? extends Exception> errs;
+
+ /** */
+ private final Map<UUID, Map<Integer, T2<Long, Long>>> cntrsPerNode;
+
+ /**
+ * @param topVer Topology version.
+ * @param errs Errors.
+ * @param cntrsPerNode Update counters.
+ */
+ RoutineRegisterResults(AffinityTopologyVersion topVer,
+ Map<UUID, ? extends Exception> errs,
+ Map<UUID, Map<Integer, T2<Long, Long>>> cntrsPerNode) {
+ this.topVer = topVer;
+ this.errs = errs;
+ this.cntrsPerNode = cntrsPerNode;
+ }
+ }
+
+ /**
* Future for stop routine.
*/
private static class StopFuture extends GridFutureAdapter<Object> {
http://git-wip-us.apache.org/repos/asf/ignite/blob/a64b941d/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/StartRequestDataV2.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/StartRequestDataV2.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/StartRequestDataV2.java
new file mode 100644
index 0000000..c001616
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/StartRequestDataV2.java
@@ -0,0 +1,164 @@
+/*
+ * 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.continuous;
+
+import java.io.Serializable;
+import org.apache.ignite.internal.managers.deployment.GridDeploymentInfo;
+import org.apache.ignite.internal.util.typedef.internal.S;
+
+/**
+ * Start request data.
+ */
+class StartRequestDataV2 implements Serializable {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** Serialized node filter. */
+ private byte[] nodeFilterBytes;
+
+ /** Deployment class name. */
+ private String clsName;
+
+ /** Deployment info. */
+ private GridDeploymentInfo depInfo;
+
+ /** Serialized handler. */
+ private byte[] hndBytes;
+
+ /** Buffer size. */
+ private int bufSize;
+
+ /** Time interval. */
+ private long interval;
+
+ /** Automatic unsubscribe flag. */
+ private boolean autoUnsubscribe;
+
+ /**
+ * @param nodeFilterBytes Serialized node filter.
+ * @param hndBytes Serialized handler.
+ * @param bufSize Buffer size.
+ * @param interval Time interval.
+ * @param autoUnsubscribe Automatic unsubscribe flag.
+ */
+ StartRequestDataV2(
+ byte[] nodeFilterBytes,
+ byte[] hndBytes,
+ int bufSize,
+ long interval,
+ boolean autoUnsubscribe) {
+ assert hndBytes != null;
+ assert bufSize > 0;
+ assert interval >= 0;
+
+ this.nodeFilterBytes = nodeFilterBytes;
+ this.hndBytes = hndBytes;
+ this.bufSize = bufSize;
+ this.interval = interval;
+ this.autoUnsubscribe = autoUnsubscribe;
+ }
+
+ /**
+ * @return Serialized node filter.
+ */
+ public byte[] nodeFilterBytes() {
+ return nodeFilterBytes;
+ }
+
+ /**
+ * @return Deployment class name.
+ */
+ public String className() {
+ return clsName;
+ }
+
+ /**
+ * @param clsName New deployment class name.
+ */
+ public void className(String clsName) {
+ this.clsName = clsName;
+ }
+
+ /**
+ * @return Deployment info.
+ */
+ public GridDeploymentInfo deploymentInfo() {
+ return depInfo;
+ }
+
+ /**
+ * @param depInfo New deployment info.
+ */
+ public void deploymentInfo(GridDeploymentInfo depInfo) {
+ this.depInfo = depInfo;
+ }
+
+ /**
+ * @return Handler.
+ */
+ public byte[] handlerBytes() {
+ return hndBytes;
+ }
+
+ /**
+ * @return Buffer size.
+ */
+ public int bufferSize() {
+ return bufSize;
+ }
+
+ /**
+ * @param bufSize New buffer size.
+ */
+ public void bufferSize(int bufSize) {
+ this.bufSize = bufSize;
+ }
+
+ /**
+ * @return Time interval.
+ */
+ public long interval() {
+ return interval;
+ }
+
+ /**
+ * @param interval New time interval.
+ */
+ public void interval(long interval) {
+ this.interval = interval;
+ }
+
+ /**
+ * @return Automatic unsubscribe flag.
+ */
+ public boolean autoUnsubscribe() {
+ return autoUnsubscribe;
+ }
+
+ /**
+ * @param autoUnsubscribe New automatic unsubscribe flag.
+ */
+ public void autoUnsubscribe(boolean autoUnsubscribe) {
+ this.autoUnsubscribe = autoUnsubscribe;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(StartRequestDataV2.class, this);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/a64b941d/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/StartRoutineDiscoveryMessageV2.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/StartRoutineDiscoveryMessageV2.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/StartRoutineDiscoveryMessageV2.java
new file mode 100644
index 0000000..275765d
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/StartRoutineDiscoveryMessageV2.java
@@ -0,0 +1,77 @@
+/*
+ * 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.continuous;
+
+import java.util.UUID;
+import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
+import org.apache.ignite.internal.util.typedef.internal.S;
+
+/**
+ *
+ */
+public class StartRoutineDiscoveryMessageV2 extends AbstractContinuousMessage {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** */
+ private static final int KEEP_BINARY_FLAG = 0x01;
+
+ /** */
+ private final StartRequestDataV2 startReqData;
+
+ /** Flags. */
+ private int flags;
+
+ /**
+ * @param routineId Routine id.
+ * @param startReqData Start request data.
+ * @param keepBinary Keep binary flag.
+ */
+ StartRoutineDiscoveryMessageV2(UUID routineId, StartRequestDataV2 startReqData, boolean keepBinary) {
+ super(routineId);
+
+ this.startReqData = startReqData;
+
+ if (keepBinary)
+ flags |= KEEP_BINARY_FLAG;
+ }
+
+ /**
+ * @return Start request data.
+ */
+ public StartRequestDataV2 startRequestData() {
+ return startReqData;
+ }
+
+ /**
+ * @return {@code True} if keep binary flag was set on continuous handler.
+ */
+ public boolean keepBinary() {
+ return (flags & KEEP_BINARY_FLAG) != 0;
+ }
+
+ /** {@inheritDoc} */
+ @Override public DiscoveryCustomMessage ackMessage() {
+ return null;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(StartRoutineDiscoveryMessageV2.class, this, "routineId", routineId());
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/a64b941d/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/StopRoutineAckDiscoveryMessage.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/StopRoutineAckDiscoveryMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/StopRoutineAckDiscoveryMessage.java
index 79d8b29..dfba0e7 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/StopRoutineAckDiscoveryMessage.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/StopRoutineAckDiscoveryMessage.java
@@ -42,6 +42,11 @@ public class StopRoutineAckDiscoveryMessage extends AbstractContinuousMessage {
}
/** {@inheritDoc} */
+ @Override public boolean stopProcess() {
+ return false;
+ }
+
+ /** {@inheritDoc} */
@Override public String toString() {
return S.toString(StopRoutineAckDiscoveryMessage.class, this, "routineId", routineId());
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/a64b941d/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java
index 4a893f4..8cad342 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java
@@ -47,6 +47,7 @@ import javax.cache.expiry.ExpiryPolicy;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteClientDisconnectedException;
import org.apache.ignite.IgniteDataStreamer;
import org.apache.ignite.IgniteDataStreamerTimeoutException;
import org.apache.ignite.IgniteException;
@@ -100,6 +101,7 @@ import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
import org.apache.ignite.internal.util.typedef.CI1;
import org.apache.ignite.internal.util.typedef.F;
+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.GPC;
@@ -1059,6 +1061,9 @@ public class DataStreamerImpl<K, V> implements IgniteDataStreamer<K, V>, Delayed
return;
while (true) {
+ if (disconnectErr != null)
+ throw disconnectErr;
+
Queue<IgniteInternalFuture<?>> q = null;
for (Buffer buf : bufMappings.values()) {
@@ -1826,15 +1831,19 @@ public class DataStreamerImpl<K, V> implements IgniteDataStreamer<K, V>, Delayed
catch (IgniteCheckedException e) {
GridFutureAdapter<Object> fut0 = ((GridFutureAdapter<Object>)fut);
- try {
- if (ctx.discovery().alive(node) && ctx.discovery().pingNode(node.id()))
- fut0.onDone(e);
- else
- fut0.onDone(new ClusterTopologyCheckedException("Failed to send request (node has left): "
- + node.id()));
- }
- catch (IgniteClientDisconnectedCheckedException e0) {
- fut0.onDone(e0);
+ if (X.hasCause(e, IgniteClientDisconnectedCheckedException.class, IgniteClientDisconnectedException.class))
+ fut0.onDone(e);
+ else {
+ try {
+ if (ctx.discovery().alive(node) && ctx.discovery().pingNode(node.id()))
+ fut0.onDone(e);
+ else
+ fut0.onDone(new ClusterTopologyCheckedException("Failed to send request (node has left): "
+ + node.id()));
+ }
+ catch (IgniteClientDisconnectedCheckedException e0) {
+ fut0.onDone(e0);
+ }
}
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/a64b941d/modules/core/src/main/java/org/apache/ignite/internal/processors/marshaller/MappingAcceptedMessage.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/marshaller/MappingAcceptedMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/marshaller/MappingAcceptedMessage.java
index 7af0559..80e3f7d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/marshaller/MappingAcceptedMessage.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/marshaller/MappingAcceptedMessage.java
@@ -63,6 +63,11 @@ public class MappingAcceptedMessage implements DiscoveryCustomMessage {
}
/** {@inheritDoc} */
+ @Override public boolean stopProcess() {
+ return false;
+ }
+
+ /** {@inheritDoc} */
@Nullable @Override public DiscoCache createDiscoCache(GridDiscoveryManager mgr,
AffinityTopologyVersion topVer, DiscoCache discoCache) {
throw new UnsupportedOperationException();
http://git-wip-us.apache.org/repos/asf/ignite/blob/a64b941d/modules/core/src/main/java/org/apache/ignite/internal/processors/marshaller/MappingProposedMessage.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/marshaller/MappingProposedMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/marshaller/MappingProposedMessage.java
index b4e13fb..9358585 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/marshaller/MappingProposedMessage.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/marshaller/MappingProposedMessage.java
@@ -98,6 +98,11 @@ public class MappingProposedMessage implements DiscoveryCustomMessage {
}
/** {@inheritDoc} */
+ @Override public boolean stopProcess() {
+ return false;
+ }
+
+ /** {@inheritDoc} */
@Nullable @Override public DiscoCache createDiscoCache(GridDiscoveryManager mgr,
AffinityTopologyVersion topVer, DiscoCache discoCache) {
throw new UnsupportedOperationException();
http://git-wip-us.apache.org/repos/asf/ignite/blob/a64b941d/modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/message/SchemaFinishDiscoveryMessage.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/message/SchemaFinishDiscoveryMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/message/SchemaFinishDiscoveryMessage.java
index 2245b24..f802e09 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/message/SchemaFinishDiscoveryMessage.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/message/SchemaFinishDiscoveryMessage.java
@@ -59,6 +59,11 @@ public class SchemaFinishDiscoveryMessage extends SchemaAbstractDiscoveryMessage
}
/** {@inheritDoc} */
+ @Override public boolean stopProcess() {
+ return false;
+ }
+
+ /** {@inheritDoc} */
@Override public boolean exchange() {
return false;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/a64b941d/modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/message/SchemaProposeDiscoveryMessage.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/message/SchemaProposeDiscoveryMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/message/SchemaProposeDiscoveryMessage.java
index 0e1270b..62b6d6a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/message/SchemaProposeDiscoveryMessage.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/message/SchemaProposeDiscoveryMessage.java
@@ -60,6 +60,11 @@ public class SchemaProposeDiscoveryMessage extends SchemaAbstractDiscoveryMessag
}
/** {@inheritDoc} */
+ @Override public boolean stopProcess() {
+ return false;
+ }
+
+ /** {@inheritDoc} */
@Override public boolean exchange() {
return exchange;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/a64b941d/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioServer.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioServer.java b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioServer.java
index e0ec8d1..0fcde0e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioServer.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioServer.java
@@ -491,6 +491,17 @@ public class GridNioServer<T> {
/**
* @param ses Session.
+ */
+ public void closeFromWorkerThread(GridNioSession ses) {
+ assert ses instanceof GridSelectorNioSessionImpl : ses;
+
+ GridSelectorNioSessionImpl ses0 = (GridSelectorNioSessionImpl)ses;
+
+ ((AbstractNioClientWorker)ses0.worker()).close((GridSelectorNioSessionImpl)ses, null);
+ }
+
+ /**
+ * @param ses Session.
* @param msg Message.
* @param createFut {@code True} if future should be created.
* @param ackC Closure invoked when message ACK is received.
@@ -2170,7 +2181,12 @@ public class GridNioServer<T> {
dumpSelectorInfo(sb, keys);
for (SelectionKey key : keys) {
- GridSelectorNioSessionImpl ses = (GridSelectorNioSessionImpl)key.attachment();
+ GridNioKeyAttachment attach = (GridNioKeyAttachment)key.attachment();
+
+ if (!attach.hasSession())
+ continue;
+
+ GridSelectorNioSessionImpl ses = attach.session();
boolean sesInfo = p == null || p.apply(ses);
http://git-wip-us.apache.org/repos/asf/ignite/blob/a64b941d/modules/core/src/main/java/org/apache/ignite/spi/IgniteSpiAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/IgniteSpiAdapter.java b/modules/core/src/main/java/org/apache/ignite/spi/IgniteSpiAdapter.java
index 1754cc8..e8c27d2 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/IgniteSpiAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/IgniteSpiAdapter.java
@@ -963,5 +963,15 @@ public abstract class IgniteSpiAdapter implements IgniteSpi {
@Override public Map<String, Object> nodeAttributes() {
return Collections.emptyMap();
}
+
+ /** {@inheritDoc} */
+ @Override public boolean communicationFailureResolveSupported() {
+ return false;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void resolveCommunicationFailure(ClusterNode node, Exception err) {
+ throw new UnsupportedOperationException();
+ }
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/a64b941d/modules/core/src/main/java/org/apache/ignite/spi/IgniteSpiContext.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/IgniteSpiContext.java b/modules/core/src/main/java/org/apache/ignite/spi/IgniteSpiContext.java
index 108c4d4..d4402f4 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/IgniteSpiContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/IgniteSpiContext.java
@@ -365,4 +365,15 @@ public interface IgniteSpiContext {
* @return Current node attributes.
*/
public Map<String, Object> nodeAttributes();
+
+ /**
+ * @return {@code True} if cluster supports communication error resolving.
+ */
+ public boolean communicationFailureResolveSupported();
+
+ /**
+ * @param node Problem node.
+ * @param err Error.
+ */
+ public void resolveCommunicationFailure(ClusterNode node, Exception err);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/a64b941d/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
index e1addd8..4a0710e 100755
--- a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
@@ -32,6 +32,7 @@ import java.nio.channels.SocketChannel;
import java.nio.channels.spi.AbstractInterruptibleChannel;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.BitSet;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
@@ -66,12 +67,14 @@ import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.IgniteInterruptedCheckedException;
import org.apache.ignite.internal.IgniteKernal;
import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
+import org.apache.ignite.internal.managers.discovery.IgniteDiscoverySpi;
import org.apache.ignite.internal.managers.eventstorage.GridLocalEventListener;
import org.apache.ignite.internal.managers.eventstorage.HighPriorityListener;
import org.apache.ignite.internal.util.GridConcurrentFactory;
import org.apache.ignite.internal.util.GridSpinReadWriteLock;
import org.apache.ignite.internal.util.future.GridFinishedFuture;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
+import org.apache.ignite.internal.util.future.IgniteFutureImpl;
import org.apache.ignite.internal.util.ipc.IpcEndpoint;
import org.apache.ignite.internal.util.ipc.IpcToNioAdapter;
import org.apache.ignite.internal.util.ipc.shmem.IpcOutOfSystemResourcesException;
@@ -134,6 +137,9 @@ import org.apache.ignite.spi.IgniteSpiThread;
import org.apache.ignite.spi.IgniteSpiTimeoutObject;
import org.apache.ignite.spi.communication.CommunicationListener;
import org.apache.ignite.spi.communication.CommunicationSpi;
+import org.apache.ignite.spi.communication.tcp.internal.ConnectionKey;
+import org.apache.ignite.spi.communication.tcp.internal.TcpCommunicationConnectionCheckFuture;
+import org.apache.ignite.spi.communication.tcp.internal.TcpCommunicationNodeConnectionCheckFuture;
import org.apache.ignite.spi.communication.tcp.messages.HandshakeMessage;
import org.apache.ignite.spi.communication.tcp.messages.HandshakeMessage2;
import org.apache.ignite.spi.communication.tcp.messages.NodeIdMessage;
@@ -146,6 +152,7 @@ import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.events.EventType.EVT_NODE_FAILED;
import static org.apache.ignite.events.EventType.EVT_NODE_LEFT;
import static org.apache.ignite.internal.util.nio.GridNioSessionMetaKey.SSL_META;
+import static org.apache.ignite.spi.communication.tcp.internal.TcpCommunicationConnectionCheckFuture.SES_FUT_META;
import static org.apache.ignite.spi.communication.tcp.messages.RecoveryLastReceivedMessage.ALREADY_CONNECTED;
import static org.apache.ignite.spi.communication.tcp.messages.RecoveryLastReceivedMessage.NEED_WAIT;
import static org.apache.ignite.spi.communication.tcp.messages.RecoveryLastReceivedMessage.NODE_STOPPING;
@@ -310,7 +317,7 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter implements Communicati
private static final IgniteProductVersion VERSION_SINCE_CLIENT_COULD_WAIT_TO_CONNECT = IgniteProductVersion.fromString("2.1.4");
/** Connection index meta for session. */
- private static final int CONN_IDX_META = GridNioSessionMetaKey.nextUniqueKey();
+ public static final int CONN_IDX_META = GridNioSessionMetaKey.nextUniqueKey();
/** Message tracker meta for session. */
private static final int TRACKER_META = GridNioSessionMetaKey.nextUniqueKey();
@@ -408,6 +415,9 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter implements Communicati
ConnectionKey connId = ses.meta(CONN_IDX_META);
if (connId != null) {
+ if (connId.dummy())
+ return;
+
UUID id = connId.nodeId();
GridCommunicationClient[] nodeClients = clients.get(id);
@@ -481,20 +491,22 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter implements Communicati
if (rmtNode == null) {
DiscoverySpi discoverySpi = ignite().configuration().getDiscoverySpi();
- assert discoverySpi instanceof TcpDiscoverySpi;
-
- TcpDiscoverySpi tcpDiscoverySpi = (TcpDiscoverySpi) discoverySpi;
+ boolean unknownNode = true;
- ClusterNode node0 = tcpDiscoverySpi.getNode0(sndId);
+ if (discoverySpi instanceof TcpDiscoverySpi) {
+ TcpDiscoverySpi tcpDiscoverySpi = (TcpDiscoverySpi) discoverySpi;
- boolean unknownNode = true;
+ ClusterNode node0 = tcpDiscoverySpi.getNode0(sndId);
- if (node0 != null) {
- assert node0.isClient() : node0;
+ if (node0 != null) {
+ assert node0.isClient() : node0;
- if (node0.version().compareTo(VERSION_SINCE_CLIENT_COULD_WAIT_TO_CONNECT) >= 0)
- unknownNode = false;
+ if (node0.version().compareTo(VERSION_SINCE_CLIENT_COULD_WAIT_TO_CONNECT) >= 0)
+ unknownNode = false;
+ }
}
+ else if (discoverySpi instanceof IgniteDiscoverySpi)
+ unknownNode = !((IgniteDiscoverySpi) discoverySpi).knownNode(sndId);
if (unknownNode) {
U.warn(log, "Close incoming connection, unknown node [nodeId=" + sndId + ", ses=" + ses + ']');
@@ -709,9 +721,9 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter implements Communicati
}
}
else {
- metricsLsnr.onMessageReceived(msg, connKey.nodeId());
-
if (msg instanceof RecoveryLastReceivedMessage) {
+ metricsLsnr.onMessageReceived(msg, connKey.nodeId());
+
GridNioRecoveryDescriptor recovery = ses.outRecoveryDescriptor();
if (recovery != null) {
@@ -724,9 +736,9 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter implements Communicati
}
recovery.ackReceived(msg0.received());
-
- return;
}
+
+ return;
}
else {
GridNioRecoveryDescriptor recovery = ses.inRecoveryDescriptor();
@@ -746,8 +758,23 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter implements Communicati
recovery.lastAcknowledged(rcvCnt);
}
}
+ else if (connKey.dummy()) {
+ assert msg instanceof NodeIdMessage : msg;
+
+ TcpCommunicationNodeConnectionCheckFuture fut = ses.meta(SES_FUT_META);
+
+ assert fut != null : msg;
+
+ fut.onConnected(U.bytesToUuid(((NodeIdMessage)msg).nodeIdBytes(), 0));
+
+ nioSrvr.closeFromWorkerThread(ses);
+
+ return;
+ }
}
+ metricsLsnr.onMessageReceived(msg, connKey.nodeId());
+
IgniteRunnable c;
if (msgQueueLimit > 0) {
@@ -2112,6 +2139,13 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter implements Communicati
}
}
+ /**
+ * @return Bound TCP server port.
+ */
+ public int boundPort() {
+ return boundTcpPort;
+ }
+
/** {@inheritDoc} */
@Override public void spiStart(String igniteInstanceName) throws IgniteSpiException {
assert locHost != null;
@@ -2570,6 +2604,27 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter implements Communicati
}
/**
+ * @param nodes Nodes to check connection with.
+ * @return Result future (each bit in result BitSet contains connection status to corresponding node).
+ */
+ public IgniteFuture<BitSet> checkConnection(List<ClusterNode> nodes) {
+ TcpCommunicationConnectionCheckFuture fut = new TcpCommunicationConnectionCheckFuture(
+ this,
+ log.getLogger(TcpCommunicationConnectionCheckFuture.class),
+ nioSrvr,
+ nodes);
+
+ long timeout = failureDetectionTimeoutEnabled() ? failureDetectionTimeout() : connTimeout;
+
+ if (log.isInfoEnabled())
+ log.info("Start check connection process [nodeCnt=" + nodes.size() + ", timeout=" + timeout + ']');
+
+ fut.init(timeout);
+
+ return new IgniteFutureImpl<>(fut);
+ }
+
+ /**
* Sends given message to destination node. Note that characteristics of the
* exchange such as durability, guaranteed delivery or error notification is
* dependant on SPI implementation.
@@ -3010,7 +3065,7 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter implements Communicati
ConnectionKey id = ses.meta(CONN_IDX_META);
if (id != null) {
- ClusterNode node = getSpiContext().node(id.nodeId);
+ ClusterNode node = getSpiContext().node(id.nodeId());
if (node != null && node.isClient()) {
String msg = "Client node outbound message queue size exceeded slowClientQueueLimit, " +
@@ -3031,9 +3086,20 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter implements Communicati
/**
* @param node Node.
* @return Node addresses.
+ * @throws IgniteCheckedException If failed.
+ */
+ private Collection<InetSocketAddress> nodeAddresses(ClusterNode node) throws IgniteCheckedException {
+ return nodeAddresses(node, filterReachableAddresses);
+ }
+
+ /**
+ * @param node Node.
+ * @param filterReachableAddresses Filter addresses flag.
+ * @return Node addresses.
* @throws IgniteCheckedException If node does not have addresses.
*/
- private LinkedHashSet<InetSocketAddress> nodeAddresses(ClusterNode node) throws IgniteCheckedException {
+ public Collection<InetSocketAddress> nodeAddresses(ClusterNode node, boolean filterReachableAddresses)
+ throws IgniteCheckedException {
Collection<String> rmtAddrs0 = node.attribute(createSpiAttributeName(ATTR_ADDRS));
Collection<String> rmtHostNames0 = node.attribute(createSpiAttributeName(ATTR_HOST_NAMES));
Integer boundPort = node.attribute(createSpiAttributeName(ATTR_PORT));
@@ -3114,7 +3180,7 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter implements Communicati
* @throws IgniteCheckedException If failed.
*/
protected GridCommunicationClient createTcpClient(ClusterNode node, int connIdx) throws IgniteCheckedException {
- LinkedHashSet<InetSocketAddress> addrs = nodeAddresses(node);
+ Collection<InetSocketAddress> addrs = nodeAddresses(node);
GridCommunicationClient client = null;
IgniteCheckedException errs = null;
@@ -3132,6 +3198,9 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter implements Communicati
int lastWaitingTimeout = 1;
while (client == null) { // Reconnection on handshake timeout.
+ if (stopping)
+ throw new IgniteSpiException("Node is stopping.");
+
if (addr.getAddress().isLoopbackAddress() && addr.getPort() == boundTcpPort) {
if (log.isDebugEnabled())
log.debug("Skipping local address [addr=" + addr +
@@ -3372,8 +3441,18 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter implements Communicati
"operating system firewall is disabled on local and remote hosts) " +
"[addrs=" + addrs + ']');
- if (enableForcibleNodeKill) {
- if (getSpiContext().node(node.id()) != null
+ boolean commErrResolve = false;
+
+ IgniteSpiContext ctx = getSpiContext();
+
+ if (connectionError(errs) && ctx.communicationFailureResolveSupported()) {
+ commErrResolve = true;
+
+ ctx.resolveCommunicationFailure(node, errs);
+ }
+
+ if (!commErrResolve && enableForcibleNodeKill) {
+ if (ctx.node(node.id()) != null
&& (CU.clientNode(node) || !CU.clientNode(getLocalNode())) &&
connectionError(errs)) {
String msg = "TcpCommunicationSpi failed to establish connection to node, node will be dropped from " +
@@ -3384,7 +3463,7 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter implements Communicati
else
U.warn(log, msg);
- getSpiContext().failNode(node.id(), "TcpCommunicationSpi failed to establish connection to node [" +
+ ctx.failNode(node.id(), "TcpCommunicationSpi failed to establish connection to node [" +
"rmtNode=" + node +
", errs=" + errs +
", connectErrs=" + Arrays.toString(errs.getSuppressed()) + ']');
@@ -4590,77 +4669,6 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter implements Communicati
/**
*
*/
- private static class ConnectionKey {
- /** */
- private final UUID nodeId;
-
- /** */
- private final int idx;
-
- /** */
- private final long connCnt;
-
- /**
- * @param nodeId Node ID.
- * @param idx Connection index.
- * @param connCnt Connection counter (set only for incoming connections).
- */
- ConnectionKey(UUID nodeId, int idx, long connCnt) {
- this.nodeId = nodeId;
- this.idx = idx;
- this.connCnt = connCnt;
- }
-
- /**
- * @return Connection counter.
- */
- long connectCount() {
- return connCnt;
- }
-
- /**
- * @return Node ID.
- */
- UUID nodeId() {
- return nodeId;
- }
-
- /**
- * @return Connection index.
- */
- int connectionIndex() {
- return idx;
- }
-
- /** {@inheritDoc} */
- @Override public boolean equals(Object o) {
- if (this == o)
- return true;
-
- if (o == null || getClass() != o.getClass())
- return false;
-
- ConnectionKey key = (ConnectionKey) o;
-
- return idx == key.idx && nodeId.equals(key.nodeId);
- }
-
- /** {@inheritDoc} */
- @Override public int hashCode() {
- int res = nodeId.hashCode();
- res = 31 * res + idx;
- return res;
- }
-
- /** {@inheritDoc} */
- @Override public String toString() {
- return S.toString(ConnectionKey.class, this);
- }
- }
-
- /**
- *
- */
interface ConnectionPolicy {
/**
* @return Thread connection index.
http://git-wip-us.apache.org/repos/asf/ignite/blob/a64b941d/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/ConnectionKey.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/ConnectionKey.java b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/ConnectionKey.java
new file mode 100644
index 0000000..0559df7
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/ConnectionKey.java
@@ -0,0 +1,117 @@
+/*
+ * 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.spi.communication.tcp.internal;
+
+import java.util.UUID;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Connection Key.
+ */
+public class ConnectionKey {
+ /** */
+ private final UUID nodeId;
+
+ /** */
+ private final int idx;
+
+ /** */
+ private final long connCnt;
+
+ /** */
+ private final boolean dummy;
+
+ /**
+ * Creates ConnectionKey with false value of dummy flag.
+ *
+ * @param nodeId Node ID. Should be not null.
+ * @param idx Connection index.
+ * @param connCnt Connection counter (set only for incoming connections).
+ */
+ public ConnectionKey(@NotNull UUID nodeId, int idx, long connCnt) {
+ this(nodeId, idx, connCnt, false);
+ }
+
+ /**
+ * @param nodeId Node ID. Should be not null.
+ * @param idx Connection index.
+ * @param connCnt Connection counter (set only for incoming connections).
+ * @param dummy Indicates that session with this ConnectionKey is temporary
+ * (for now dummy sessions are used only for Communication Failure Resolving process).
+ */
+ public ConnectionKey(@NotNull UUID nodeId, int idx, long connCnt, boolean dummy) {
+ this.nodeId = nodeId;
+ this.idx = idx;
+ this.connCnt = connCnt;
+ this.dummy = dummy;
+ }
+
+ /**
+ * @return Connection counter.
+ */
+ public long connectCount() {
+ return connCnt;
+ }
+
+ /**
+ * @return Node ID.
+ */
+ public UUID nodeId() {
+ return nodeId;
+ }
+
+ /**
+ * @return Connection index.
+ */
+ public int connectionIndex() {
+ return idx;
+ }
+
+ /**
+ * @return {@code True} if this ConnectionKey is dummy and serves temporary session.
+ */
+ public boolean dummy() {
+ return dummy;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean equals(Object o) {
+ if (this == o)
+ return true;
+
+ if (o == null || getClass() != o.getClass())
+ return false;
+
+ ConnectionKey key = (ConnectionKey) o;
+
+ return idx == key.idx && nodeId.equals(key.nodeId);
+ }
+
+ /** {@inheritDoc} */
+ @Override public int hashCode() {
+ int res = nodeId.hashCode();
+ res = 31 * res + idx;
+ return res;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(ConnectionKey.class, this);
+ }
+}