Repository: ignite Updated Branches: refs/heads/ignite-1093-2 d5d78c0de -> f34d950db
1093 Project: http://git-wip-us.apache.org/repos/asf/ignite/repo Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/f34d950d Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/f34d950d Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/f34d950d Branch: refs/heads/ignite-1093-2 Commit: f34d950db1750a9306a0e4a19091c6074ecd5ddd Parents: d5d78c0 Author: Anton Vinogradov <[email protected]> Authored: Sat Oct 17 12:28:10 2015 +0300 Committer: Anton Vinogradov <[email protected]> Committed: Sat Oct 17 12:28:10 2015 +0300 ---------------------------------------------------------------------- .../dht/preloader/GridDhtPartitionDemander.java | 17 ----------------- .../dht/preloader/GridDhtPartitionSupplier.java | 4 ++++ 2 files changed, 4 insertions(+), 17 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/ignite/blob/f34d950d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionDemander.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionDemander.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionDemander.java index d9f7b1c..aac1d2c 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionDemander.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionDemander.java @@ -473,8 +473,6 @@ public class GridDhtPartitionDemander { return; // Supple message based on another future. if (topologyChanged(fut)) { // Topology already changed (for the future that supply message based on). - fut.cancel(); - return; } @@ -583,15 +581,11 @@ public class GridDhtPartitionDemander { cctx.io().sendOrderedMessage(node, GridCachePartitionExchangeManager.rebalanceTopic(idx), d, cctx.ioPolicy(), cctx.config().getRebalanceTimeout()); } - else - fut.cancel(); } catch (ClusterTopologyCheckedException e) { if (log.isDebugEnabled()) log.debug("Node left during rebalancing [node=" + node.id() + ", msg=" + e.getMessage() + ']'); - - fut.cancel(); } catch (IgniteCheckedException ex) { U.error(log, "Failed to receive partitions from node (rebalancing will not " + @@ -1355,8 +1349,6 @@ public class GridDhtPartitionDemander { Collection<Integer> missed = new HashSet<>(); if (topologyChanged(fut)) { - fut.cancel(); - return; } @@ -1371,16 +1363,7 @@ public class GridDhtPartitionDemander { missed.addAll(set); } } - catch (ClusterTopologyCheckedException e) { - if (log.isDebugEnabled()) - log.debug("Node left during rebalancing (will retry) [node=" + node.id() + - ", msg=" + e.getMessage() + ']'); - - fut.cancel(); - } catch (InterruptedException e) { - fut.cancel(); - throw new IgniteCheckedException(e); } } http://git-wip-us.apache.org/repos/asf/ignite/blob/f34d950d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionSupplier.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionSupplier.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionSupplier.java index b070ac0..694088b 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionSupplier.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionSupplier.java @@ -130,6 +130,10 @@ class GridDhtPartitionSupplier { if (lsnr != null) cctx.events().removeListener(lsnr); + for (Map.Entry<T2<UUID, Integer>, SupplyContext> entry : scMap.entrySet()) { + clearContext(entry.getValue(), log); + } + stopOldListeners(); }
