IGNITE-28662 : test

modules/commons/src/main/java/org/apache/ignite/internal/util/CommonUtils.java

@@ -68,6 +68,7 @@
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.IgniteClientDisconnectedException;
 import org.apache.ignite.IgniteCommonsSystemProperties;
@@ -231,6 +232,12 @@ public abstract class CommonUtils {
     /** Default client connector port range. */
     public static final int DFLT_PORT_RANGE = 100;
 
+    /**
+     * Default timeout in milliseconds to wait for a sub-routines, runnables to finish on {@link IgniteKernal#stop(boolean)}
+     * or other stop or cancelation requests.
+     */
+    public static int DFLT_WAIT_TO_STOP_TIMOEUT = 60_000;
+
     /**
      * Short date format pattern for log messages in "quiet" mode.
      * Only time is included since we don't expect "quiet" mode to be used
@@ -2468,43 +2475,82 @@ public static void cancel(Iterable<? extends GridWorker> ws) {
     }
 
     /**
-     * Joins runnable.
+     * Joins grid worker with a timeout. Logs exceptable failures.
      *
-     * @param w Worker to join.
-     * @param log The logger to possible exception.
-     * @return {@code true} if worker has not been interrupted, {@code false} if it was interrupted.
+     * @param worker Worker to join.
+     * @param timeout Timeout to join in milliseconds. If negative, ignored.
+     * @param log The logger for exceptable failures.
+     * @return {@code True} if {@code worker} has been successfully stopped, wasn't interrupted or timeouted. {@code False}
+     *         if {@code worker} was interrupted or timeouted.
      */
-    public static boolean join(@Nullable GridWorker w, @Nullable IgniteLogger log) {
-        if (w != null)
-            try {
-                w.join();
-            }
-            catch (InterruptedException ignore) {
-                warn(log, "Got interrupted while waiting for completion of runnable: " + w);
+    public static boolean join(@Nullable GridWorker worker, long timeout, IgniteLogger log) {
+        assert log != null;
 
-                Thread.currentThread().interrupt();
+        if (worker == null)
+            return true;
 
-                return false;
-            }
+        try {
+            worker.join(timeout);
+        }
+        catch (InterruptedException ignore) {
+            warn(log, "Got interrupted while waiting for completion of grid worker: " + worker);
+
+            Thread.currentThread().interrupt();
+
+            return false;
+        }
+        catch (TimeoutException te) {
+            warn(log, "The timeout expired while waiting for completion of grid worker: " + worker);
+
+            return false;
+        }
 
         return true;
     }
 
     /**
-     * Joins given collection of runnables.
+     * Joins grid worker with unlimited waiting. Logs exceptable failures.
+     *
+     * @param worker Worker to join.
+     * @param log The logger for exceptable failures.
+     * @return {@code True} if {@code worker} has been successfully stopped, wasn't interrupted or timeouted. {@code False}
+     *         if {@code worker} was interrupted or timeouted.
+     */
+    public static boolean join(@Nullable GridWorker worker, IgniteLogger log) {
+        return join(worker, -1L, log);
+    }
+
+    /**
+     * Joins given collection of runnables with a timoeut. Logs exceptable failures.
      *
      * @param ws Collection of workers to join.
+     * @param timeout Timeout to join in milliseconds. If negative, ignored.
      * @param log The logger to possible exceptions.
      * @return {@code true} if none of the worker have been interrupted,
      *      {@code false} if at least one was interrupted.
      */
-    public static boolean join(Iterable<? extends GridWorker> ws, IgniteLogger log) {
+    public static boolean join(@Nullable Iterable<? extends GridWorker> ws, long timeout, IgniteLogger log) {
+        if (ws == null)
+            return true;
+
         boolean retval = true;
 
-        if (ws != null)
-            for (GridWorker w : ws)
-                if (!join(w, log))
+        if (timeout < 0) {
+            for (GridWorker w : ws) {
+                if (!join(w, -1L, log))
                     retval = false;
+            }
+        }
+        else {
+            long timeThresholdNs = System.nanoTime() + millisToNanos(timeout);
+
+            for (GridWorker w : ws) {
+                long timeout0 = Math.max(0L, nanosToMillis(timeThresholdNs - System.nanoTime()));
+
+                if (!join(w, timeout0, log))
+                    retval = false;
+            }
+        }
 
         return retval;
     }

modules/commons/src/main/java/org/apache/ignite/internal/util/worker/GridWorker.java

@@ -19,6 +19,7 @@
 
 import java.util.concurrent.Executor;
 import java.util.concurrent.Future;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLongFieldUpdater;
 import org.apache.ignite.IgniteInterruptedException;
@@ -227,20 +228,43 @@ public void cancel() {
     }
 
     /**
-     * Joins this runnable.
+     * Joins this runnable with a timeout.
      *
-     * @throws InterruptedException Thrown in case of interruption.
+     * @param timeout Operation timeout in milliseconds. If negative, ignored.
+     * @throws InterruptedException In case of the interruption.
+     * @throws TimeoutException In case of expired {@code timeout} to join.
      */
-    public void join() throws InterruptedException {
-        if (log.isDebugEnabled())
-            log.debug("Joining grid runnable: " + this);
+    public void join(long timeout) throws InterruptedException, TimeoutException {
+        long timeThresholdNs = timeout < 0L ? -1L : System.nanoTime() + CommonUtils.millisToNanos(timeout);
+
+        if (log.isDebugEnabled()) {
+            if (timeout >= 0L)
+                log.debug("Joining grid runnable: " + this + " with the Wtimeout: " + timeout + "ms.");
+            else
+                log.debug("Joining grid runnable: " + this + '.');
+        }
 
         if ((runner == null && isCancelled.get()) || finished)
             return;
 
-        synchronized (mux) {
-            while (!finished)
-                mux.wait();
+        while (!finished) {
+            if (timeout >= 0L) {
+                long leftMs = CommonUtils.nanosToMillis(timeThresholdNs - System.nanoTime());
+
+                if (leftMs < 1L)
+                    throw new TimeoutException("The timeout has expired while waiting to join.");
+
+                synchronized (mux) {
+                    if (!finished)
+                        mux.wait(leftMs);
+                }
+            }
+            else {
+                synchronized (mux) {
+                    if (!finished)
+                        mux.wait();
+                }
+            }
         }
     }
 

modules/commons/src/main/java/org/apache/ignite/internal/util/worker/GridWorkerPool.java

@@ -93,20 +93,36 @@ public void execute(final GridWorker w) throws IgniteCheckedException {
     }
 
     /**
-     * Waits for all workers to finish.
+     * Waits for all workers to finish with a timeout. Logs failures if {@link #log} is not {@code null}.
      *
-     * @param cancel Flag to indicate whether workers should be cancelled
-     *      before waiting for them to finish.
+     * @param cancel Flag to indicate whether workers should be cancelled before waiting for them to finish.
+     * @param timeout Timeout to join in milliseconds. If negative, ignored.
      */
-    public void join(boolean cancel) {
+    public void join(boolean cancel, long timeout) {
+        long timeThresholdNs = timeout < 0L ? -1L : System.nanoTime() + CommonUtils.millisToNanos(timeout);
+
+        if (cancel) {
+            for (GridWorker worker : workers) {
+                try {
+                    if (cancel)
+                        CommonUtils.cancel(worker);
+                }
+                catch (Throwable e) {
+                    if (log != null)
+                        log.warning("Failed to cancel grid worker [" + worker.name() + ']', e);
+                }
+            }
+        }
+
         for (GridWorker worker : workers) {
             try {
-                if (cancel)
-                    CommonUtils.cancel(worker);
-
-                CommonUtils.join(worker, log);
+                CommonUtils.join(
+                    worker,
+                    timeout < 0L ? -1L : CommonUtils.nanosToMillis(timeThresholdNs - System.nanoTime()),
+                    log
+                );
             }
-            catch (Exception e) {
+            catch (Throwable e) {
                 if (log != null)
                     log.warning("Failed to stop grid worker [" + worker.name() + ']', e);
             }

modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java

@@ -96,6 +96,7 @@
 import org.apache.ignite.internal.thread.context.OperationContext;
 import org.apache.ignite.internal.thread.context.Scope;
 import org.apache.ignite.internal.thread.context.function.OperationContextAwareWrapper;
+import org.apache.ignite.internal.util.CommonUtils;
 import org.apache.ignite.internal.util.GridAtomicLong;
 import org.apache.ignite.internal.util.GridBoundedConcurrentLinkedHashMap;
 import org.apache.ignite.internal.util.GridSpinBusyLock;
@@ -1683,9 +1684,10 @@ private static String nodeDescription(ClusterNode node) {
 
         // Stop segment checker.
         if (segChecker != null) {
-            segChecker.cancel();
+            if (cancel)
+                segChecker.cancel();
 
-            U.join(segChecker, log);
+            U.join(segChecker, 0L, log);
         }
 
         if (!locJoin.isDone())
@@ -1700,12 +1702,11 @@ private static String nodeDescription(ClusterNode node) {
         getSpi().setListener(null);
 
         U.cancel(discoEvtHnd);
-
-        U.join(discoEvtHnd, log);
-
         U.cancel(discoMsgNotifier);
 
-        U.join(discoMsgNotifier, log);
+        U.join(discoMsgNotifier, CommonUtils.DFLT_WAIT_TO_STOP_TIMOEUT, log);
+        // Event processing might have crusial routines better to wait for. Or requiring to stop (process) manually.
+        U.join(discoEvtHnd, log);
 
         // Stop SPI itself.
         stopSpi();

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

@@ -5016,7 +5016,7 @@ else if (newSysTtl == CU.TTL_ZERO) {
                     interceptorVal = cctx.config().getInterceptor().onBeforePut(interceptEntry, updated0);
                 }
                 catch (Throwable e) {
-                    throw new IgniteCheckedException(e);
+                    throw new IgniteCheckedException("Cache's interceptor failed in 'onBeforePut'.", e);
                 }
 
                 wasIntercepted = true;

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

@@ -466,7 +466,10 @@ public GridCacheVersion mappedVersion(GridCacheVersion from) {
     public void onStop() {
         stopping = true;
 
-        cancelClientFutures(stopError());
+        var stopErr = stopError();
+
+        cancelClientFutures(stopErr);
+        cancelFuturesWithException(stopErr, dataStreamerFuts);
     }
 
     /** {@inheritDoc} */
@@ -496,7 +499,7 @@ private void cancelFuturesWithException(
             try {
                 ((GridFutureAdapter)fut).onDone(err);
             }
-            catch (Exception e) {
+            catch (Throwable e) {
                 U.warn(log, "Failed to complete future on node stop (will ignore): " + fut, e);
             }
         }
@@ -612,6 +615,8 @@ public GridFutureAdapter addDataStreamerFuture(AffinityTopologyVersion topVer) {
 
         assert add;
 
+        onFutureAdded(fut);
+
         return fut;
     }
 

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStoreManager.java

@@ -990,7 +990,7 @@ public <T> T afterAsyncCompletion(IgniteOutClosure<T> closure) {
          * Cancels async tasks.
          */
         public void awaitAsyncTaskCompletion(boolean cancel) {
-            workerPool.join(cancel);
+            workerPool.join(cancel, -1L);
         }
     }
 

modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobProcessor.java

@@ -495,8 +495,8 @@ public GridJobProcessor(GridKernalContext ctx) {
             }
         }
 
-        U.join(activeJobs.values(), log);
-        U.join(cancelledJobs.values(), log);
+        U.join(activeJobs.values(), -1L, log);
+        U.join(cancelledJobs.values(), -1L, log);
 
         // Ignore topology changes.
         ctx.event().removeLocalEventListener(discoLsnr);

modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/ClientListenerProcessor.java

@@ -49,6 +49,7 @@
 import org.apache.ignite.internal.processors.odbc.odbc.OdbcConnectionContext;
 import org.apache.ignite.internal.systemview.ClientConnectionAttributeViewWalker;
 import org.apache.ignite.internal.systemview.ClientConnectionViewWalker;
+import org.apache.ignite.internal.util.CommonUtils;
 import org.apache.ignite.internal.util.GridSpinBusyLock;
 import org.apache.ignite.internal.util.HostAndPortRange;
 import org.apache.ignite.internal.util.nio.GridNioAsyncNotifyFilter;
@@ -529,7 +530,7 @@ else if (connCtx.managementClient()) {
 
             execSvc = null;
 
-            mgmtPool.join(cancel);
+            mgmtPool.join(cancel, CommonUtils.DFLT_WAIT_TO_STOP_TIMOEUT);
 
             mgmtPool = null;
 

modules/core/src/main/java/org/apache/ignite/internal/processors/service/ServiceDeploymentManager.java

@@ -46,6 +46,7 @@
 import org.apache.ignite.internal.thread.context.OperationContext;
 import org.apache.ignite.internal.thread.context.Scope;
 import org.apache.ignite.internal.thread.context.function.OperationContextAwareWrapper;
+import org.apache.ignite.internal.util.CommonUtils;
 import org.apache.ignite.internal.util.GridSpinBusyLock;
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.internal.util.worker.GridWorker;
@@ -145,7 +146,7 @@ void stopProcessing(IgniteCheckedException stopErr) {
 
             U.cancel(depTaskHandler);
 
-            U.join(depTaskHandler, log);
+            U.join(depTaskHandler, CommonUtils.DFLT_WAIT_TO_STOP_TIMOEUT, log);
 
             depTaskHandler.clearQueue();
 

modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskProcessor.java

@@ -308,7 +308,7 @@ private IgniteClientDisconnectedCheckedException disconnectedError(@Nullable Ign
                 }
             }
 
-            U.join(tasks.values(), log);
+            U.join(tasks.values(), -1L, log);
         }
 
         // Remove discovery and message listeners.