ServiceProcessor recalculates all service assignments even if there is a pending topology change. This closes
@@ -1530,6 +1530,9 @@ public class GridServiceProcessor extends GridProcessorAdapter implements Ignite
* Topology listener.
*/
private class TopologyListener implements DiscoveryEventListener {
+ /** */
+ private volatile AffinityTopologyVersion currTopVer = null;
+
/** {@inheritDoc} */
@Override public void onEvent(DiscoveryEvent evt, final DiscoCache discoCache) {
GridSpinBusyLock busyLock = GridServiceProcessor.this.busyLock;
@@ -1562,6 +1565,8 @@ public class GridServiceProcessor extends GridProcessorAdapter implements Ignite
else
topVer = new AffinityTopologyVersion((evt).topologyVersion(), 0);
+ currTopVer = topVer;
+
depExe.execute(new DepRunnable() {
@Override public void run0() {
// In case the cache instance isn't tracked by DiscoveryManager anymore.
@@ -1582,6 +1587,19 @@ public class GridServiceProcessor extends GridProcessorAdapter implements Ignite
boolean firstTime = true;
while (it.hasNext()) {
+ // If topology changed again, let next event handle it.
+ AffinityTopologyVersion currTopVer0 = currTopVer;
+
+ if (currTopVer0 != topVer) {
+ if (log.isInfoEnabled())
+ log.info("Service processor detected a topology change during " +
+ "assignments calculation (will abort current iteration and " +
+ "re-calculate on the newer version): " +
+ "[topVer=" + topVer + ", newTopVer=" + currTopVer + ']');
+
+ return;
+ }
+
Cache.Entry<Object, Object> e = it.next();
if (firstTime) {
