IGNITE-28681 Introduce dedicated method for obtaining queued elements.

Author: petrov-mg

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java

@@ -851,7 +851,7 @@ public static Object rebalanceTopic(int idx) {
         // Finish all exchange futures.
         ExchangeFutureSet exchFuts0 = exchFuts;
 
-        for (CachePartitionExchangeWorkerTask task : exchWorker) {
+        for (CachePartitionExchangeWorkerTask task : exchWorker.queuedElements()) {
             if (task instanceof GridDhtPartitionsExchangeFuture)
                 ((GridDhtPartitionsExchangeFuture)task).onDone(stopErr);
         }
@@ -2445,7 +2445,7 @@ public boolean mergeExchanges(final GridDhtPartitionsExchangeFuture curFut, Grid
         if (exchWorker.waitForExchangeFuture(resVer))
             return true;
 
-        for (CachePartitionExchangeWorkerTask task : exchWorker) {
+        for (CachePartitionExchangeWorkerTask task : exchWorker.queuedElements()) {
             if (task instanceof GridDhtPartitionsExchangeFuture) {
                 GridDhtPartitionsExchangeFuture fut = (GridDhtPartitionsExchangeFuture)task;
 
@@ -2525,7 +2525,7 @@ public boolean mergeExchangesOnCoordinator(
         synchronized (curFut.mutex()) {
             int awaited = 0;
 
-            for (CachePartitionExchangeWorkerTask task : exchWorker) {
+            for (CachePartitionExchangeWorkerTask task : exchWorker.queuedElements()) {
                 if (task instanceof GridDhtPartitionsExchangeFuture) {
                     GridDhtPartitionsExchangeFuture fut = (GridDhtPartitionsExchangeFuture)task;
 
@@ -2624,7 +2624,7 @@ private void waitForTestVersion(AffinityTopologyVersion exchMergeTestWaitVer, Gr
         while (U.currentTimeMillis() < end) {
             boolean found = false;
 
-            for (CachePartitionExchangeWorkerTask task : exchWorker) {
+            for (CachePartitionExchangeWorkerTask task : exchWorker.queuedElements()) {
                 if (task instanceof GridDhtPartitionsExchangeFuture) {
                     GridDhtPartitionsExchangeFuture fut = (GridDhtPartitionsExchangeFuture)task;
 
@@ -2864,7 +2864,7 @@ private void removeMergedFutures(AffinityTopologyVersion resVer, GridDhtPartitio
             if (resVer.compareTo(exchFut.initialVersion()) != 0) {
                 waitForExchangeFuture(resVer);
 
-                for (CachePartitionExchangeWorkerTask task : this) {
+                for (CachePartitionExchangeWorkerTask task : queuedElements()) {
                     if (task instanceof GridDhtPartitionsExchangeFuture) {
                         GridDhtPartitionsExchangeFuture fut0 = (GridDhtPartitionsExchangeFuture)task;
 
@@ -2914,7 +2914,7 @@ void processCustomTask(CachePartitionExchangeWorkerTask task) {
          */
         boolean hasPendingExchange() {
             if (!isQueueEmpty()) {
-                for (CachePartitionExchangeWorkerTask task : this) {
+                for (CachePartitionExchangeWorkerTask task : queuedElements()) {
                     if (isExchangeTask(task))
                         return true;
                 }
@@ -2928,7 +2928,7 @@ boolean hasPendingExchange() {
          */
         boolean hasPendingServerExchange() {
             if (!isQueueEmpty()) {
-                for (CachePartitionExchangeWorkerTask task : this) {
+                for (CachePartitionExchangeWorkerTask task : queuedElements()) {
                     if (task instanceof GridDhtPartitionsExchangeFuture) {
                         if (((GridDhtPartitionsExchangeFuture)task).changedAffinity())
                             return true;
@@ -2948,7 +2948,7 @@ void dumpExchangeDebugInfo() {
             if (DIAGNOSTIC_WARN_LIMIT > 0) {
                 int cnt = 0;
 
-                for (CachePartitionExchangeWorkerTask task : this) {
+                for (CachePartitionExchangeWorkerTask task : queuedElements()) {
                     if (task instanceof GridDhtPartitionsExchangeFuture) {
                         U.warn(log, ">>> " + ((GridDhtPartitionsExchangeFuture)task).shortInfo());
 

modules/core/src/main/java/org/apache/ignite/internal/util/worker/AsynchronousQueueProcessor.java

@@ -26,6 +26,7 @@
 import org.apache.ignite.internal.thread.context.OperationContext;
 import org.apache.ignite.internal.thread.context.OperationContextSnapshot;
 import org.apache.ignite.internal.thread.context.function.OperationContextAwareWrapper;
+import org.apache.ignite.internal.util.typedef.F;
 import org.apache.ignite.internal.worker.WorkersRegistry;
 import org.apache.ignite.thread.IgniteThread;
 import org.jetbrains.annotations.NotNull;
@@ -41,7 +42,7 @@
  * @param <T> Type of items to be processed.
  * @param <W> Type of wrapper over processing item that are stored in the underlying queue.
  */
-public abstract class AsynchronousQueueProcessor<T, W extends OperationContextAwareWrapper<T>> extends GridWorker implements Iterable<T> {
+public abstract class AsynchronousQueueProcessor<T, W extends OperationContextAwareWrapper<T>> extends GridWorker {
     /** */
     private final BlockingQueue<W> workerQueue;
 
@@ -137,6 +138,15 @@ public boolean removeQueuedElement(Object o) {
         return workerQueue.removeIf(w -> o.equals(w.delegate()));
     }
 
+    /** */
+    @NotNull public Iterable<T> queuedElements() {
+        return new Iterable<>() {
+            @Override public @NotNull Iterator<T> iterator() {
+                return F.iterator(workerQueue, OperationContextAwareWrapper::delegate, true);
+            }
+        };
+    }
+
     /** */
     public void clearQueue() {
         workerQueue.clear();
@@ -152,21 +162,6 @@ public boolean isQueueEmpty() {
         return workerQueue.isEmpty();
     }
 
-    /** {@inheritDoc} */
-    @Override public @NotNull Iterator<T> iterator() {
-        Iterator<W> iter = workerQueue.iterator();
-
-        return new Iterator<>() {
-            @Override public boolean hasNext() {
-                return iter.hasNext();
-            }
-
-            @Override public T next() {
-                return iter.next().delegate();
-            }
-        };
-    }
-
     /** */
     public void drainQueue(Consumer<? super T> consumer) {
         W element;

modules/core/src/test/java/org/apache/ignite/cache/affinity/PendingExchangeTest.java

@@ -264,7 +264,7 @@ private void waitForExchnagesBegin(GridCachePartitionExchangeManager exchangeMan
             assertTrue(GridTestUtils.waitForCondition(() -> {
                 int exFuts = 0;
 
-                for (CachePartitionExchangeWorkerTask task : exchWorker) {
+                for (CachePartitionExchangeWorkerTask task : exchWorker.queuedElements()) {
                     if (task instanceof GridDhtPartitionsExchangeFuture)
                         exFuts++;
                 }