You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by GitBox <gi...@apache.org> on 2020/07/02 12:23:52 UTC
[GitHub] [ignite] vldpyatkov commented on a change in pull request #7971: IGNITE-13193 Added fallback to full rebalance if historical one has f…
vldpyatkov commented on a change in pull request #7971:
URL: https://github.com/apache/ignite/pull/7971#discussion_r448915584
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
##########
@@ -3185,197 +3176,220 @@ private void body0() throws InterruptedException, IgniteCheckedException {
AffinityTopologyVersion resVer = null;
- try {
- if (isCancelled())
- break;
+ if (isCancelled())
+ break;
- if (task instanceof RebalanceReassignExchangeTask)
- exchId = ((RebalanceReassignExchangeTask) task).exchangeId();
- else if (task instanceof ForceRebalanceExchangeTask) {
- forcePreload = true;
+ if (task instanceof RebalanceReassignExchangeTask) {
+ RebalanceReassignExchangeTask reassignTask = (RebalanceReassignExchangeTask)task;
- timeout = 0; // Force refresh.
+ exchId = reassignTask.exchangeId();
- exchId = ((ForceRebalanceExchangeTask)task).exchangeId();
- }
- else {
- assert task instanceof GridDhtPartitionsExchangeFuture : task;
+ GridDhtPartitionsExchangeFuture fut = reassignTask.future();
- exchFut = (GridDhtPartitionsExchangeFuture)task;
+ assert fut.changedAffinity() :
+ "Reassignment request started for exchange future which didn't change affinity " +
+ "[exchId=" + exchId + ", fut=" + exchFut + ']';
- exchId = exchFut.exchangeId();
+ if (fut.hasInapplicableNodesForRebalance() ) {
Review comment:
Extra space
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheOffheapManager.java
##########
@@ -1022,14 +1026,22 @@ private Metas getOrAllocateCacheMetas() throws IgniteCheckedException {
FileWALPointer minPtr = (FileWALPointer)database.checkpointHistory().searchEarliestWalPointer(grp.groupId(), partsCounters);
- WALIterator it = grp.shared().wal().replay(minPtr);
+ try {
+ WALIterator it = grp.shared().wal().replay(minPtr);
+
+ WALHistoricalIterator iterator = new WALHistoricalIterator(log, grp, partCntrs, it);
- WALHistoricalIterator iterator = new WALHistoricalIterator(log, grp, partCntrs, it);
+ // Add historical partitions which are unabled to reserve to missing set.
+ missing.addAll(iterator.missingParts);
- // Add historical partitions which are unabled to reserve to missing set.
- missing.addAll(iterator.missingParts);
+ return iterator;
+ }
+ catch (Exception ex) {
+ if (!X.hasCause(ex, IgniteHistoricalIteratorException.class))
+ throw new IgniteHistoricalIteratorException(ex);
Review comment:
Why do not throw ex here?
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/IgniteHistoricalIterator.java
##########
@@ -36,4 +36,9 @@
* @return {@code True} if all data for given partition has already been returned.
*/
public boolean isDone(int partId);
+
+ /**
+ * @return {@code true} if all historical partitions are processed.
+ */
+ public boolean allHistoricalPartitionsDone();
Review comment:
Where does this method use?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org