CVE-2022-39368: Incomplete Cleanup (CWE-459)
value: Eclipse Californium is a Java implementation of RFC7252 - Constrained Application Protocol for IoT Cloud services. In versions prior to 3.7.0, and 2.7.4, Californium is vulnerable to a Denial of Service. Failing handshakes don't cleanup counters for throttling, causing the threshold to be reached without being released again. This results in permanently dropping records. The issue was reported for certificate based handshakes, but may also affect PSK based handshakes. It generally affects client and server as well. This issue is patched in version 3.7.0 and 2.7.4. There are no known workarounds. main: commit 726bac57659410da463dcf404b3e79a7312ac0b9 2.7.x: commit 5648a0c27c2c2667c98419254557a14bac2b1f3f
CWE description: The product does not properly "clean up" and remove temporary or supporting resources after they have been used.
CVSS v3 base score: 8.2
Published: 2022-11-10T00:15Z
Repository: https://github.com/eclipse-californium/californium
Language: Java
Files changed: element-connector/src/main/java/org/eclipse/californium/elements/util/ExecutorsUtil.java, element-connector/src/main/java/org/eclipse/californium/elements/util/LimitedRunnable.java, element-connector/src/main/java/org/eclipse/californium/elements/util/SerialExecutor.java, scandium-core/api-changes.json, scandium-core/src/main/java/org/eclipse/californium/scandium/DTLSConnector.java, scandium-core/src/main/java/org/eclipse/californium/scandium/DtlsBindingPreprocessingConnector.java, scandium-core/src/main/java/org/eclipse/californium/scandium/dtls/Connection.java, scandium-core/src/main/java/org/eclipse/californium/scandium/dtls/InMemoryConnectionStore.java, scandium-core/src/main/java/org/eclipse/californium/scandium/dtls/InMemoryReadWriteLockConnectionStore.java, scandium-core/src/test/java/org/eclipse/californium/scandium/DTLSConnectorAdvancedTest.java, scandium-core/src/test/java/org/eclipse/californium/scandium/DTLSConnectorHandshakeTest.java, scandium-core/src/test/java/org/eclipse/californium/scandium/DTLSConnectorTest.java
Commit date: 2022-09-19 15:48:20 +0200
Commit message: Fix pending job counters.

Run pending jobs on connection shutdown to align the counters.

Fixes issue #2065

Signed-off-by: Achim Kraus <achim.kraus@cloudcoap.net>

Diff: diff --git a/element-connector/src/main/java/org/eclipse/californium/elements/util/ExecutorsUtil.java b/element-connector/src/main/java/org/eclipse/californium/elements/util/ExecutorsUtil.java
index 504cc7637..ea54d18b5 100644
--- a/element-connector/src/main/java/org/eclipse/californium/elements/util/ExecutorsUtil.java
+++ b/element-connector/src/main/java/org/eclipse/californium/elements/util/ExecutorsUtil.java
@@ -232,10 +232,29 @@ public class ExecutorsUtil {
 				LOGGER.warn("await termination {} ms exceeded the maximum {} ms", time, timeMaxToWaitInMs);
 			}
 		}
 	}
 
+	/**
+	 * Run all jobs.
+	 * 
+	 * On {@link Throwable} write only warning and continue to run the other
+	 * jobs.
+	 * 
+	 * @param jobs list of jobs to run.
+	 * @since 3.7
+	 */
+	public static void runAll(List<Runnable> jobs) {
+		for (Runnable job : jobs) {
+			try {
+				job.run();
+			} catch (Throwable e) {
+				LOGGER.warn("Ignoring error:", e);
+			}
+		}
+	}
+
 	/**
 	 * Set remove on cancel policy of provided executor.
 	 * 
 	 * Ignored, if not supported by java vm or executors implementation.
 	 * 
diff --git a/element-connector/src/main/java/org/eclipse/californium/elements/util/LimitedRunnable.java b/element-connector/src/main/java/org/eclipse/californium/elements/util/LimitedRunnable.java
new file mode 100644
index 000000000..54d4af890
--- /dev/null
+++ b/element-connector/src/main/java/org/eclipse/californium/elements/util/LimitedRunnable.java
@@ -0,0 +1,107 @@
+/*******************************************************************************
+ * Copyright (c) 2022 Contributors to the Eclipse Foundation.
+ * 
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v2.0
+ * and Eclipse Distribution License v1.0 which accompany this distribution.
+ * 
+ * The Eclipse Public License is available at
+ *    http://www.eclipse.org/legal/epl-v20.html
+ * and the Eclipse Distribution License is available at
+ *    http://www.eclipse.org/org/documents/edl-v10.html.
+ * 
+ ******************************************************************************/
+package org.eclipse.californium.elements.util;
+
+import java.util.concurrent.Executor;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Limited jobs.
+ *
+ * Limits pending jobs based on counters.
+ * 
+ * Note: for proper operation, ensure the at the end of
+ * {@link Runnable#run()} the {@link LimitedRunnable#onDequeueing()} is
+ * called.
+ * 
+ * <pre>
+ *	public void run() {
+ *		try {
+ *			if (running.get() &amp;&amp; connection.isExecuting()) {
+ *				// process
+ *			}
+ *		} finally {
+ *			onDequeueing();
+ *		}
+ *	}
+ * </pre>
+ * 
+ * @since 3.7
+ */
+public abstract class LimitedRunnable implements Runnable {
+
+	/**
+	 * Down counter to limit pending jobs.
+	 */
+	private final AtomicInteger counter;
+	/**
+	 * Indicator for overflows. {@code true}, if the counter indicates an
+	 * overflow, {@code false}, otherwise.
+	 */
+	private volatile boolean overflow;
+
+	/**
+	 * Create job limited by the provided counter
+	 * 
+	 * @param counter counter in count-down mode
+	 */
+	public LimitedRunnable(AtomicInteger counter) {
+		this.counter = counter;
+	}
+
+	/**
+	 * Queue job.
+	 * 
+	 * @throws RejectedExecutionException if limit is exceeded
+	 */
+	public void onQueueing() {
+		if (counter.decrementAndGet() < 0) {
+			overflow = true;
+			throw new RejectedExecutionException("queue overflow!");
+		}
+	}
+
+	/**
+	 * Dequeue job.
+	 */
+	public void onDequeueing() {
+		counter.incrementAndGet();
+	}
+
+	/**
+	 * Checks, if queueing this job causes an counter overflow.
+	 * 
+	 * @return {@code true}, if the counter indicates an overflow,
+	 *         {@code false}, otherwise.
+	 */
+	public boolean isOverflown() {
+		return overflow;
+	}
+
+	/**
+	 * Execute this job.
+	 * 
+	 * @param executor executor to execute jobs.
+	 */
+	public void execute(Executor executor) {
+		try {
+			onQueueing();
+			executor.execute(this);
+		} catch (RejectedExecutionException ex) {
+			onDequeueing();
+			throw ex;
+		}
+	}
+}
\ No newline at end of file
diff --git a/element-connector/src/main/java/org/eclipse/californium/elements/util/SerialExecutor.java b/element-connector/src/main/java/org/eclipse/californium/elements/util/SerialExecutor.java
index ccc97c15b..9c4eeae23 100644
--- a/element-connector/src/main/java/org/eclipse/californium/elements/util/SerialExecutor.java
+++ b/element-connector/src/main/java/org/eclipse/californium/elements/util/SerialExecutor.java
@@ -88,14 +88,18 @@ public class SerialExecutor extends AbstractExecutorService {
 	/**
 	 * Create serial executor
 	 * 
 	 * @param executor target executor. If {@code null}, the executor is
 	 *            shutdown.
+	 * @throws IllegalArgumentException if the executor is also a
+	 *             {@link SerialExecutor}
 	 */
 	public SerialExecutor(final Executor executor) {
 		if (executor == null) {
 			shutdown = true;
+		} else if (executor instanceof SerialExecutor) {
+			throw new IllegalArgumentException("Sequences of SerialExecutors are not supported!");
 		}
 		this.executor = executor;
 	}
 
 	@Override
@@ -209,11 +213,11 @@ public class SerialExecutor extends AbstractExecutorService {
 			lock.unlock();
 		}
 	}
 
 	/**
-	 * Shutdown this executor and add all pending task from {@link #tasks} to
+	 * Shutdown this executor and add all pending jobs from {@link #tasks} to
 	 * the provided collection.
 	 * 
 	 * @param jobs collection to add pending jobs.
 	 * @return number of added jobs
 	 * @see #shutdownNow()
@@ -278,35 +282,35 @@ public class SerialExecutor extends AbstractExecutorService {
 				executor.execute(new Runnable() {
 
 					@Override
 					public void run() {
 						try {
+							setOwner();
+							ExecutionListener current = listener.get();
 							try {
-								setOwner();
-								ExecutionListener current = listener.get();
+								if (current != null) {
+									current.beforeExecution();
+								}
+								command.run();
+							} catch (Throwable t) {
+								LOGGER.error("unexpected error occurred:", t);
+							} finally {
 								try {
 									if (current != null) {
-										current.beforeExecution();
+										current.afterExecution();
 									}
-									command.run();
 								} catch (Throwable t) {
-									LOGGER.error("unexpected error occurred:", t);
-								} finally {
-									try {
-										if (current != null) {
-											current.afterExecution();
-										}
-									} catch (Throwable t) {
-										LOGGER.error("unexpected error occurred:", t);
-									}
-									clearOwner();
+									LOGGER.error("unexpected error occurred after execution:", t);
 								}
-							} finally {
+								clearOwner();
+							}
+						} finally {
+							try {
 								scheduleNextJob();
+							} catch (RejectedExecutionException ex) {
+								LOGGER.debug("shutdown?", ex);
 							}
-						} catch (RejectedExecutionException ex) {
-							LOGGER.debug("shutdown?", ex);
 						}
 					}
 				});
 			} else if (shutdown) {
 				terminated.signalAll();
@@ -330,11 +334,12 @@ public class SerialExecutor extends AbstractExecutorService {
 	}
 
 	/**
 	 * Execution listener.
 	 * 
-	 * Called before and after executing a task.
+	 * Called before and after executing a task. The calling thread is the same
+	 * as the the one executing the job.
 	 * 
 	 * @since 2.4
 	 */
 	public interface ExecutionListener {
 
diff --git a/scandium-core/api-changes.json b/scandium-core/api-changes.json
index 372b7ae9c..e45325c24 100644
--- a/scandium-core/api-changes.json
+++ b/scandium-core/api-changes.json
@@ -398,7 +398,32 @@
 						"justification": "Work In Progress, 3.5,0 was an emergency release for CVE-2022-21449"					
 					}
 				]
 			}
 		}
+	],
+	"3.7.0": [
+		{
+			"extension": "revapi.differences",
+			"configuration": {
+				"ignore": true,
+				"differences": [
+					{
+						"code": "java.method.removed",
+						"old": "method boolean org.eclipse.californium.scandium.DTLSConnector::executeHandshakeResult(java.util.concurrent.Executor, org.eclipse.californium.scandium.dtls.Connection, java.lang.Runnable)",
+						"justification": "Not part of the public API"
+					},
+					{
+						"code": "java.method.removed",
+						"old": "method boolean org.eclipse.californium.scandium.DTLSConnector::executeInbound(java.util.concurrent.Executor, java.net.InetSocketAddress, java.lang.Runnable)",
+						"justification": "Not part of the public API"
+					},
+					{
+						"code": "java.method.removed",
+						"old": "method boolean org.eclipse.californium.scandium.DTLSConnector::executeOutbound(java.util.concurrent.Executor, java.net.InetSocketAddress, java.lang.Runnable)",
+						"justification": "Not part of the public API"
+					}
+				]
+			}
+		}
 	]
 }
diff --git a/scandium-core/src/main/java/org/eclipse/californium/scandium/DTLSConnector.java b/scandium-core/src/main/java/org/eclipse/californium/scandium/DTLSConnector.java
index ba7419e31..c06c8ec31 100644
--- a/scandium-core/src/main/java/org/eclipse/californium/scandium/DTLSConnector.java
+++ b/scandium-core/src/main/java/org/eclipse/californium/scandium/DTLSConnector.java
@@ -178,10 +178,11 @@ import org.eclipse.californium.elements.util.ClockUtil;
 import org.eclipse.californium.elements.util.DaemonThreadFactory;
 import org.eclipse.californium.elements.util.DatagramReader;
 import org.eclipse.californium.elements.util.ExecutorsUtil;
 import org.eclipse.californium.elements.util.FilteredLogger;
 import org.eclipse.californium.elements.util.LeastRecentlyUsedCache;
+import org.eclipse.californium.elements.util.LimitedRunnable;
 import org.eclipse.californium.elements.util.NamedThreadFactory;
 import org.eclipse.californium.elements.util.NetworkInterfacesUtil;
 import org.eclipse.californium.elements.util.NoPublicAPI;
 import org.eclipse.californium.elements.util.SerialExecutor;
 import org.eclipse.californium.elements.util.StringUtil;
@@ -768,10 +769,50 @@ public class DTLSConnector implements Connector, PersistentConnector, Persistent
 	 */
 	public DtlsHealth getHealthHandler() {
 		return health;
 	}
 
+	/**
+	 * Update health statistic.
+	 * 
+	 * Update {@link DtlsHealthExtended#setConnections(int)},
+	 * {@link DtlsHealthExtended2#setPendingIncomingJobs(int)},
+	 * {@link DtlsHealthExtended2#setPendingOutgoingJobs(int)}, and
+	 * {@link DtlsHealthExtended2#setPendingHandshakeJobs(int)}.
+	 * 
+	 * @return {@code true}, if some pending jobs left, {@code false}, if not.
+	 * @since 3.7
+	 */
+	public boolean updateHealth() {
+		boolean pending = false;
+		if (health instanceof DtlsHealthExtended) {
+			((DtlsHealthExtended) health).setConnections(maxConnections - connectionStore.remainingCapacity());
+		}
+		if (health instanceof DtlsHealthExtended2) {
+			DtlsHealthExtended2 health2 = (DtlsHealthExtended2) health;
+			int jobs = maxPendingOutboundJobs - pendingOutboundJobsCountdown.get();
+			health2.setPendingOutgoingJobs(jobs);
+			if (jobs > 0) {
+				LOGGER.debug("Pending out jobs {}", jobs);
+			}
+			pending = jobs > 0;
+			jobs = maxPendingInboundJobs - pendingInboundJobsCountdown.get();
+			health2.setPendingIncomingJobs(jobs);
+			if (jobs > 0) {
+				LOGGER.debug("Pending in jobs {}", jobs);
+			}
+			pending |= jobs > 0;
+			jobs = maxPendingHandshakeResultJobs - pendingHandshakeResultJobsCountdown.get();
+			health2.setPendingHandshakeJobs(jobs);
+			if (jobs > 0) {
+				LOGGER.debug("Pending handshake jobs {}", jobs);
+			}
+			pending |= jobs > 0;
+		}
+		return pending;
+	}
+
 	/**
 	 * Create default health handler.
 	 * 
 	 * @param configuration configuration
 	 * @return default health handler.
@@ -1275,18 +1316,18 @@ public class DTLSConnector implements Connector, PersistentConnector, Persistent
 						if (healthStatusIntervalMillis > 0
 								&& TimeUnit.NANOSECONDS.toMillis(now - lastNanos) > healthStatusIntervalMillis) {
 							health.dump(config.getLoggingTag(), maxConnections,
 									connectionStore.remainingCapacity(), pendingHandshakesWithoutVerifiedPeer.get());
 							lastNanos = now;
-						} else if (health instanceof DtlsHealthExtended) {
-							((DtlsHealthExtended) health)
-									.setConnections(maxConnections - connectionStore.remainingCapacity());
+						} else {
+							updateHealth();
 						}
 					}
 
 				}, intervalMillis, intervalMillis, TimeUnit.MILLISECONDS);
 			}
+			updateHealth();
 		}
 
 		recentHandshakeCleaner = timer.scheduleWithFixedDelay(new Runnable() {
 
 			private int calls = 0;
@@ -1420,17 +1461,11 @@ public class DTLSConnector implements Connector, PersistentConnector, Persistent
 					LOGGER.warn("Shutdown DTLS connector on [{}] executor not terminated in time!", lastBindAddress);
 				}
 			} catch (InterruptedException e) {
 			}
 		}
-		for (Runnable job : pending) {
-			try {
-				job.run();
-			} catch (Exception e) {
-				LOGGER.warn("Shutdown DTLS connector:", e);
-			}
-		}
+		ExecutorsUtil.runAll(pending);
 		if (stop) {
 			LOGGER.debug("DTLS connector on [{}] stopped.", lastBindAddress);
 		}
 	}
 
@@ -1840,22 +1875,27 @@ public class DTLSConnector implements Connector, PersistentConnector, Persistent
 				if (health != null) {
 					health.receivingRecord(true);
 				}
 				return;
 			}
-			executeInbound(getExecutorService(), peerAddress, new Runnable() {
+			executeInbound(getExecutorService(), peerAddress, new LimitedRunnable(pendingInboundJobsCountdown) {
 
 				@Override
 				public void run() {
-					if (MDC_SUPPORT) {
-						MDC.put("PEER", StringUtil.toString(firstRecord.getPeerAddress()));
-					}
-					processNewClientHello(firstRecord);
-					if (MDC_SUPPORT) {
-						MDC.clear();
+					try {
+						if (running.get()) {
+							if (MDC_SUPPORT) {
+								MDC.put("PEER", StringUtil.toString(firstRecord.getPeerAddress()));
+							}
+							processNewClientHello(firstRecord);
+							if (MDC_SUPPORT) {
+								MDC.clear();
+							}
+						}
+					} finally {
+						onDequeueing();
 					}
-					pendingInboundJobsCountdown.incrementAndGet();
 				}
 			});
 			return;
 		}
 
@@ -1879,20 +1919,20 @@ public class DTLSConnector implements Connector, PersistentConnector, Persistent
 		SerialExecutor serialExecutor = connection.getExecutor();
 
 		for (final Record record : records) {
 			record.setAddress(peerAddress, router);
 			try {
-				if (!executeInbound(serialExecutor, peerAddress, new Runnable() {
+				if (!executeInbound(serialExecutor, peerAddress, new LimitedRunnable(pendingInboundJobsCountdown) {
 
 					@Override
 					public void run() {
 						try {
-							if (running.get()) {
+							if (running.get() && connection.isExecuting()) {
 								processRecord(record, connection);
 							}
 						} finally {
-							pendingInboundJobsCountdown.incrementAndGet();
+							onDequeueing();
 						}
 					}
 				})) {
 					break;
 				}
@@ -1913,37 +1953,28 @@ public class DTLSConnector implements Connector, PersistentConnector, Persistent
 	 * @param job job to be executed.
 	 * @return {@code true}, if execution was accepted, {@code false}, if
 	 *         execution was denied
 	 * @since 3.5
 	 */
-	protected boolean executeInbound(Executor executor, InetSocketAddress peer, Runnable job) {
-		boolean pending = false;
+	@NoPublicAPI
+	protected boolean executeInbound(Executor executor, InetSocketAddress peer, LimitedRunnable job) {
 		try {
-			int count = pendingInboundJobsCountdown.decrementAndGet();
-			if (count >= 0) {
-				executor.execute(job);
-				pending = true;
-				if (health instanceof DtlsHealthExtended2) {
-					((DtlsHealthExtended2) health).setPendingIncomingJobs(maxPendingInboundJobs - count);
-				}
-			} else {
+			job.execute(executor);
+			return true;
+		} catch (RejectedExecutionException e) {
+			if (job.isOverflown()) {
 				DROP_LOGGER_IN_FILTERED.info("Inbound jobs overflow! Dropping inbound message from peer [{}]",
 						StringUtil.toLog(peer));
-			}
-		} catch (RejectedExecutionException e) {
-			// dont't terminate connection on shutdown!
-			LOGGER.debug("Execution rejected while processing record from peer [{}]",
-					StringUtil.toLog(peer), e);
-		} finally {
-			if (!pending) {
-				if (health != null) {
-					health.receivingRecord(true);
-				}
-				pendingInboundJobsCountdown.incrementAndGet();
+			} else {
+				// dont't terminate connection on shutdown!
+				LOGGER.debug("Execution rejected while processing record from peer [{}]", StringUtil.toLog(peer), e);
 			}
 		}
-		return pending;
+		if (health != null) {
+			health.receivingRecord(true);
+		}
+		return false;
 	}
 
 	/**
 	 * Process received record.
 	 * 
@@ -2928,51 +2959,42 @@ public class DTLSConnector implements Connector, PersistentConnector, Persistent
 			}
 			throw error;
 		}
 
 		final long now = ClockUtil.nanoRealtime();
-		try {
-			SerialExecutor executor = connection.getExecutor();
-			if (!executeOutbound(executor, message.getInetSocketAddress(), new Runnable() {
+		SerialExecutor executor = connection.getExecutor();
+		if (!executeOutbound(executor, message.getInetSocketAddress(), new LimitedRunnable(pendingOutboundJobsCountdown) {
 
-				@Override
-				public void run() {
-					try {
-						if (running.get()) {
-							sendMessage(now, message, connection);
-						} else {
-							DROP_LOGGER.trace("DTLSConnector drops {} outgoing bytes to {}, connector not running!",
-									message.getSize(), StringUtil.toLog(message.getInetSocketAddress()));
-							message.onError(new InterruptedIOException("Connector is not running."));
-							if (health != null) {
-								health.sendingRecord(true);
-							}
-						}
-					} catch (Exception e) {
-						if (running.get()) {
-							LOGGER.warn("Exception thrown by executor thread [{}]",
-									Thread.currentThread().getName(), e);
-						}
-						DROP_LOGGER.trace("DTLSConnector drops {} outgoing bytes to {} {}", message.getSize(),
-								StringUtil.toLog(message.getInetSocketAddress()), e.getMessage());
+			@Override
+			public void run() {
+				try {
+					if (running.get() && connection.isExecuting()) {
+						sendMessage(now, message, connection);
+					} else {
+						DROP_LOGGER.trace("DTLSConnector drops {} outgoing bytes to {}, connector not running!",
+								message.getSize(), StringUtil.toLog(message.getInetSocketAddress()));
+						message.onError(new InterruptedIOException("Connector is not running."));
 						if (health != null) {
 							health.sendingRecord(true);
 						}
-						message.onError(e);
-					} finally {
-						pendingOutboundJobsCountdown.incrementAndGet();
 					}
+				} catch (Exception e) {
+					if (running.get()) {
+						LOGGER.warn("Exception thrown by executor thread [{}]", Thread.currentThread().getName(), e);
+					}
+					DROP_LOGGER.trace("DTLSConnector drops {} outgoing bytes to {} {}", message.getSize(),
+							StringUtil.toLog(message.getInetSocketAddress()), e.getMessage());
+					if (health != null) {
+						health.sendingRecord(true);
+					}
+					message.onError(e);
+				} finally {
+					onDequeueing();
 				}
-			})) {
-				message.onError(new IllegalStateException("Outbound message overflow!"));
 			}
-		} catch (RejectedExecutionException e) {
-			LOGGER.debug("Execution rejected while sending application record [peer: {}]",
-					StringUtil.toLog(message.getInetSocketAddress()), e);
-			DROP_LOGGER.trace("DTLSConnector drops {} outgoing bytes to {}, {}", message.getSize(),
-					StringUtil.toLog(message.getInetSocketAddress()), e.getMessage());
-			message.onError(new InterruptedIOException("Connector is not running."));
+		})) {
+			message.onError(new IllegalStateException("Outbound message overflow!"));
 		}
 	}
 
 	/**
 	 * Execute outbound job.
@@ -2982,33 +3004,28 @@ public class DTLSConnector implements Connector, PersistentConnector, Persistent
 	 * @param job job to be executed.
 	 * @return {@code true}, if execution was accepted, {@code false}, if
 	 *         execution was denied
 	 * @since 3.5
 	 */
-	protected boolean executeOutbound(Executor executor, InetSocketAddress peer, Runnable job) {
-		boolean pending = false;
+	@NoPublicAPI
+	protected boolean executeOutbound(Executor executor, InetSocketAddress peer, LimitedRunnable job) {
 		try {
-			int count = pendingOutboundJobsCountdown.decrementAndGet();
-			if (count >= 0) {
-				executor.execute(job);
-				pending = true;
-				if (health instanceof DtlsHealthExtended2) {
-					((DtlsHealthExtended2) health).setPendingOutgoingJobs(maxPendingOutboundJobs - count);
-				}
-			} else {
+			job.execute(executor);
+			return true;
+		} catch (RejectedExecutionException e) {
+			if (job.isOverflown()) {
 				DROP_LOGGER_OUT_FILTERED.info("Outbound jobs overflow! Dropping outbound message to peer [{}]",
 						StringUtil.toLog(peer));
-			}
-		} finally {
-			if (!pending) {
-				if (health != null) {
-					health.sendingRecord(true);
-				}
-				pendingOutboundJobsCountdown.incrementAndGet();
+			} else {
+				// dont't terminate connection on shutdown!
+				LOGGER.debug("Execution rejected while processing record to peer [{}]", StringUtil.toLog(peer), e);
 			}
 		}
-		return pending;
+		if (health != null) {
+			health.sendingRecord(true);
+		}
+		return false;
 	}
 
 	/**
 	 * Sends a raw message to a peer.
 	 * <p>
@@ -3404,16 +3421,16 @@ public class DTLSConnector implements Connector, PersistentConnector, Persistent
 		final Connection connection = connectionStore.get(handshakeResult.getConnectionId());
 		if (connection != null) {
 			if (connection.hasOngoingHandshake()) {
 				try {
 					SerialExecutor serialExecutor = connection.getExecutor();
-					executeHandshakeResult(serialExecutor, connection, new Runnable() {
+					executeHandshakeResult(serialExecutor, connection, new LimitedRunnable(pendingHandshakeResultJobsCountdown) {
 
 						@Override
 						public void run() {
 							try {
-								if (running.get()) {
+								if (running.get() && connection.isExecuting()) {
 									Handshaker handshaker = connection.getOngoingHandshake();
 									if (handshaker != null) {
 										handshaker.processAsyncHandshakeResult(handshakeResult);
 									} else {
 										LOGGER.debug("No ongoing handshake for result [{}]", connection);
@@ -3425,17 +3442,14 @@ public class DTLSConnector implements Connector, PersistentConnector, Persistent
 							} catch (HandshakeException e) {
 								processExceptionDuringHandshake(null, connection, e);
 							} catch (IllegalStateException e) {
 								LOGGER.warn("Exception while processing handshake result [{}]", connection, e);
 							} finally {
-								pendingHandshakeResultJobsCountdown.incrementAndGet();
+								onDequeueing();
 							}
 						}
 					});
-				} catch (RejectedExecutionException e) {
-					// dont't terminate connection on shutdown!
-					LOGGER.debug("Execution rejected while processing handshake result [{}]", connection);
 				} catch (RuntimeException e) {
 					LOGGER.warn("Unexpected error occurred while processing handshake result [{}]", connection, e);
 				}
 			} else {
 				LOGGER.debug("No ongoing handshake for handshake result [{}]", connection);
@@ -3453,30 +3467,25 @@ public class DTLSConnector implements Connector, PersistentConnector, Persistent
 	 * @param job job to be executed.
 	 * @return {@code true}, if execution was accepted, {@code false}, if
 	 *         execution was denied
 	 * @since 3.5
 	 */
-	protected boolean executeHandshakeResult(Executor executor, Connection connection, Runnable job) {
-		boolean pending = false;
+	@NoPublicAPI
+	protected boolean executeHandshakeResult(Executor executor, Connection connection, LimitedRunnable job) {
 		try {
-			int count = pendingHandshakeResultJobsCountdown.decrementAndGet();
-			if (count >= 0) {
-				executor.execute(job);
-				pending = true;
-				if (health instanceof DtlsHealthExtended2) {
-					((DtlsHealthExtended2) health).setPendingHandshakeJobs(maxPendingHandshakeResultJobs - count);
-				}
-			} else {
+			job.execute(executor);
+			return true;
+		} catch (RejectedExecutionException e) {
+			if (job.isOverflown()) {
 				DROP_LOGGER_HANDSHAKE_RESULTS_FILTERED
 						.info("Handshake result jobs overflow! Dropping handshake result [{}]", connection);
-			}
-		} finally {
-			if (!pending) {
-				pendingHandshakeResultJobsCountdown.incrementAndGet();
+			} else {
+				// dont't terminate connection on shutdown!
+				LOGGER.debug("Execution rejected while processing handshake result [{}]", connection, e);
 			}
 		}
-		return pending;
+		return false;
 	}
 
 	/**
 	 * Gets the auto handshake timeout.
 	 * 
diff --git a/scandium-core/src/main/java/org/eclipse/californium/scandium/DtlsBindingPreprocessingConnector.java b/scandium-core/src/main/java/org/eclipse/californium/scandium/DtlsBindingPreprocessingConnector.java
index 80ed3e1e0..fea7ea020 100644
--- a/scandium-core/src/main/java/org/eclipse/californium/scandium/DtlsBindingPreprocessingConnector.java
+++ b/scandium-core/src/main/java/org/eclipse/californium/scandium/DtlsBindingPreprocessingConnector.java
@@ -80,11 +80,11 @@ public class DtlsBindingPreprocessingConnector extends DTLSConnector {
 	 * in order to continue records processing.
 	 * </p>
 	 * <p>
 	 * If created action is running on separate threads, consider limiting such pending jobs in order
 	 * to protect against DoS attack. Similar approach may be seen in
-	 * {@link DTLSConnector#executeInbound(Executor, InetSocketAddress, Runnable)}
+	 * {@link DTLSConnector#executeInbound(Executor, InetSocketAddress, org.eclipse.californium.scandium.DTLSConnector.LimitedRunnable)}
 	 * </p>
 	 */
 	public interface DtlsBindingActionWithCallback {
 		/**
 		 * @param cid DTLS connection ID.
diff --git a/scandium-core/src/main/java/org/eclipse/californium/scandium/dtls/Connection.java b/scandium-core/src/main/java/org/eclipse/californium/scandium/dtls/Connection.java
index 70f894b9e..42805fb71 100644
--- a/scandium-core/src/main/java/org/eclipse/californium/scandium/dtls/Connection.java
+++ b/scandium-core/src/main/java/org/eclipse/californium/scandium/dtls/Connection.java
@@ -40,10 +40,11 @@ package org.eclipse.californium.scandium.dtls;
 
 import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.security.Principal;
 import java.util.ConcurrentModificationException;
+import java.util.List;
 import java.util.Objects;
 import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -52,10 +53,11 @@ import org.eclipse.californium.elements.MapBasedEndpointContext.Attributes;
 import org.eclipse.californium.elements.util.Bytes;
 import org.eclipse.californium.elements.util.ClockUtil;
 import org.eclipse.californium.elements.util.DataStreamReader;
 import org.eclipse.californium.elements.util.DatagramReader;
 import org.eclipse.californium.elements.util.DatagramWriter;
+import org.eclipse.californium.elements.util.ExecutorsUtil;
 import org.eclipse.californium.elements.util.SerialExecutor;
 import org.eclipse.californium.elements.util.SerialExecutor.ExecutionListener;
 import org.eclipse.californium.elements.util.SerializationUtil;
 import org.eclipse.californium.elements.util.StringUtil;
 import org.eclipse.californium.scandium.ConnectionListener;
@@ -117,11 +119,11 @@ public final class Connection {
 	/**
 	 * Expired real time nanoseconds of the last message send or received.
 	 */
 	private long lastMessageNanos;
 	private long lastPeerAddressNanos;
-	private SerialExecutor serialExecutor;
+	private volatile SerialExecutor serialExecutor;
 	private InetSocketAddress peerAddress;
 	private InetSocketAddress router;
 	private ConnectionId cid;
 	/**
 	 * Data of this connection specific for the used {@link DatagramFilter}.
@@ -226,10 +228,29 @@ public final class Connection {
 	 */
 	public boolean isExecuting() {
 		return serialExecutor != null && !serialExecutor.isShutdown();
 	}
 
+	/**
+	 * Shutdown executor and run all pending jobs.
+	 * 
+	 * The jobs are intended to check {@link #isExecuting()} in order to detect
+	 * the shutdown.
+	 * 
+	 * @return number of pending jobs.
+	 * @since 3.7
+	 */
+	public int shutdown() {
+		SerialExecutor executor = getExecutor();
+		if (executor != null) {
+			List<Runnable> pendings = executor.shutdownNow();
+			ExecutorsUtil.runAll(pendings);
+			return pendings.size();
+		}
+		return 0;
+	}
+
 	/**
 	 * Get session listener of connection.
 	 * 
 	 * @return session listener.
 	 */
diff --git a/scandium-core/src/main/java/org/eclipse/californium/scandium/dtls/InMemoryConnectionStore.java b/scandium-core/src/main/java/org/eclipse/californium/scandium/dtls/InMemoryConnectionStore.java
index c2e8bb9cc..75d3efc08 100644
--- a/scandium-core/src/main/java/org/eclipse/californium/scandium/dtls/InMemoryConnectionStore.java
+++ b/scandium-core/src/main/java/org/eclipse/californium/scandium/dtls/InMemoryConnectionStore.java
@@ -502,28 +502,19 @@ public class InMemoryConnectionStore implements ResumptionSupportingConnectionSt
 		boolean removed;
 		SessionId sessionId = connection.getEstablishedSessionIdentifier();
 		synchronized (this) {
 			removed = connections.remove(connection.getConnectionId(), connection) == connection;
 			if (removed) {
-				if (connection.isExecuting()) {
-					List<Runnable> pendings = connection.getExecutor().shutdownNow();
-					if (LOGGER.isTraceEnabled()) {
-						LOGGER.trace("{}connection: remove {} (size {}, left jobs: {})", tag, connection,
-								connections.size(), pendings.size(), new Throwable("connection removed!"));
-					} else if (pendings.isEmpty()) {
-						LOGGER.debug("{}connection: remove {} (size {})", tag, connection, connections.size());
-					} else {
-						LOGGER.debug("{}connection: remove {} (size {}, left jobs: {})", tag, connection,
-								connections.size(), pendings.size());
-					}
+				int pendings = connection.shutdown();
+				if (LOGGER.isTraceEnabled()) {
+					LOGGER.trace("{}connection: remove {} (size {}, left jobs: {})", tag, connection,
+							connections.size(), pendings, new Throwable("connection removed!"));
+				} else if (pendings == 0) {
+					LOGGER.debug("{}connection: remove {} (size {})", tag, connection, connections.size());
 				} else {
-					if (LOGGER.isTraceEnabled()) {
-						LOGGER.trace("{}connection: remove {} (size {})", tag, connection, connections.size(),
-								new Throwable("connection removed!"));
-					} else {
-						LOGGER.debug("{}connection: remove {} (size {})", tag, connection, connections.size());
-					}
+					LOGGER.debug("{}connection: remove {} (size {}, left jobs: {})", tag, connection,
+							connections.size(), pendings);
 				}
 				connection.startByClientHello(null);
 				removeByAddressConnections(connection);
 				removeByEstablishedSessions(sessionId, connection);
 				ConnectionListener listener = connectionListener;
diff --git a/scandium-core/src/main/java/org/eclipse/californium/scandium/dtls/InMemoryReadWriteLockConnectionStore.java b/scandium-core/src/main/java/org/eclipse/californium/scandium/dtls/InMemoryReadWriteLockConnectionStore.java
index 0aa656aa8..39e7ea306 100644
--- a/scandium-core/src/main/java/org/eclipse/californium/scandium/dtls/InMemoryReadWriteLockConnectionStore.java
+++ b/scandium-core/src/main/java/org/eclipse/californium/scandium/dtls/InMemoryReadWriteLockConnectionStore.java
@@ -580,28 +580,19 @@ public class InMemoryReadWriteLockConnectionStore implements ReadWriteLockConnec
 		Principal principal = session == null ? null : session.getPeerIdentity();
 		connections.writeLock().lock();
 		try {
 			removed = connections.remove(connection.getConnectionId(), connection) == connection;
 			if (removed) {
-				if (connection.isExecuting()) {
-					List<Runnable> pendings = connection.getExecutor().shutdownNow();
-					if (LOGGER.isTraceEnabled()) {
-						LOGGER.trace("{}connection: remove {} (size {}, left jobs: {})", tag, connection,
-								connections.size(), pendings.size(), new Throwable("connection removed!"));
-					} else if (pendings.isEmpty()) {
-						LOGGER.debug("{}connection: remove {} (size {})", tag, connection, connections.size());
-					} else {
-						LOGGER.debug("{}connection: remove {} (size {}, left jobs: {})", tag, connection,
-								connections.size(), pendings.size());
-					}
+				int pendings = connection.shutdown();
+				if (LOGGER.isTraceEnabled()) {
+					LOGGER.trace("{}connection: remove {} (size {}, left jobs: {})", tag, connection,
+							connections.size(), pendings, new Throwable("connection removed!"));
+				} else if (pendings == 0) {
+					LOGGER.debug("{}connection: remove {} (size {})", tag, connection, connections.size());
 				} else {
-					if (LOGGER.isTraceEnabled()) {
-						LOGGER.trace("{}connection: remove {} (size {})", tag, connection, connections.size(),
-								new Throwable("connection removed!"));
-					} else {
-						LOGGER.debug("{}connection: remove {} (size {})", tag, connection, connections.size());
-					}
+					LOGGER.debug("{}connection: remove {} (size {}, left jobs: {})", tag, connection,
+							connections.size(), pendings);
 				}
 				connection.startByClientHello(null);
 				removeByAddressConnections(connection);
 				removeByEstablishedSessions(sessionId, connection);
 				removeByPrincipal(principal, connection);
diff --git a/scandium-core/src/test/java/org/eclipse/californium/scandium/DTLSConnectorAdvancedTest.java b/scandium-core/src/test/java/org/eclipse/californium/scandium/DTLSConnectorAdvancedTest.java
index 206c10567..c83266e92 100644
--- a/scandium-core/src/test/java/org/eclipse/californium/scandium/DTLSConnectorAdvancedTest.java
+++ b/scandium-core/src/test/java/org/eclipse/californium/scandium/DTLSConnectorAdvancedTest.java
@@ -3211,11 +3211,11 @@ public class DTLSConnectorAdvancedTest {
 			rawClient.stop();
 		}
 	}
 
 	@Test
-	public void testDisableHelloVerifRequestForPsk() throws Exception {
+	public void testDisableHelloVerifyRequestForPsk() throws Exception {
 		alternativeServerHelper = new ConnectorHelper(network);
 
 		alternativeServerHelper.serverBuilder
 				.set(DtlsConfig.DTLS_RETRANSMISSION_TIMEOUT, RETRANSMISSION_TIMEOUT_MS, TimeUnit.MILLISECONDS)
 				.set(DtlsConfig.DTLS_MAX_RETRANSMISSIONS, MAX_RETRANSMISSIONS)
@@ -3272,11 +3272,11 @@ public class DTLSConnectorAdvancedTest {
 			serverHealth.reset();
 		}
 	}
 
 	@Test
-	public void testDisabledHelloVerifRequestForPskWithCertificate() throws Exception {
+	public void testDisabledHelloVerifyRequestForPskWithCertificate() throws Exception {
 		alternativeServerHelper = new ConnectorHelper(network);
 
 		alternativeServerHelper.serverBuilder
 				.set(DtlsConfig.DTLS_RETRANSMISSION_TIMEOUT, RETRANSMISSION_TIMEOUT_MS, TimeUnit.MILLISECONDS)
 				.set(DtlsConfig.DTLS_MAX_RETRANSMISSIONS, MAX_RETRANSMISSIONS)
@@ -3334,11 +3334,11 @@ public class DTLSConnectorAdvancedTest {
 			serverHealth.reset();
 		}
 	}
 
 	@Test
-	public void testDisabledHelloVerifRequestWithCertificate() throws Exception {
+	public void testDisabledHelloVerifyRequestWithCertificate() throws Exception {
 		alternativeServerHelper = new ConnectorHelper(network);
 
 		alternativeServerHelper.serverBuilder
 				.set(DtlsConfig.DTLS_RETRANSMISSION_TIMEOUT, RETRANSMISSION_TIMEOUT_MS, TimeUnit.MILLISECONDS)
 				.set(DtlsConfig.DTLS_MAX_RETRANSMISSIONS, MAX_RETRANSMISSIONS)
@@ -3752,11 +3752,11 @@ public class DTLSConnectorAdvancedTest {
 			byte[] cidBytes = new byte[4];
 			RandomManager.currentRandom().nextBytes(cidBytes);
 			cid = new ConnectionId(cidBytes);
 		}
 		Connection connection = new Connection(peer);
-		connection.setConnectorContext(new SerialExecutor(executor), null);
+		connection.setConnectorContext(executor, null);
 		connection.setConnectionId(cid);
 		return connection;
 	}
 
 	public static class TestRecordLayer implements RecordLayer {
diff --git a/scandium-core/src/test/java/org/eclipse/californium/scandium/DTLSConnectorHandshakeTest.java b/scandium-core/src/test/java/org/eclipse/californium/scandium/DTLSConnectorHandshakeTest.java
index 66f05f48b..5ec135baf 100644
--- a/scandium-core/src/test/java/org/eclipse/californium/scandium/DTLSConnectorHandshakeTest.java
+++ b/scandium-core/src/test/java/org/eclipse/californium/scandium/DTLSConnectorHandshakeTest.java
@@ -74,10 +74,12 @@ import org.eclipse.californium.elements.util.JceNames;
 import org.eclipse.californium.elements.util.JceProviderUtil;
 import org.eclipse.californium.elements.util.SimpleMessageCallback;
 import org.eclipse.californium.elements.util.SslContextUtil;
 import org.eclipse.californium.elements.util.SslContextUtil.Credentials;
 import org.eclipse.californium.elements.util.TestCertificatesTools;
+import org.eclipse.californium.elements.util.TestCondition;
+import org.eclipse.californium.elements.util.TestConditionTools;
 import org.eclipse.californium.elements.util.TestScope;
 import org.eclipse.californium.scandium.ConnectorHelper.AlertCatcher;
 import org.eclipse.californium.scandium.ConnectorHelper.BuilderSetup;
 import org.eclipse.californium.scandium.ConnectorHelper.BuilderSetups;
 import org.eclipse.californium.scandium.ConnectorHelper.LatchSessionListener;
@@ -161,10 +163,12 @@ public class DTLSConnectorHandshakeTest {
 	ConnectorHelper serverHelper;
 
 	AsyncAdvancedPskStore serverPskStore;
 	AsyncNewAdvancedCertificateVerifier serverVerifier;
 
+	DtlsHealthLogger serverHealth;
+
 	DtlsConnectorConfig.Builder clientBuilder;
 	DTLSConnector client;
 	TestContext clientTestContext;
 	AlertCatcher clientAlertCatcher;
 	ApplicationLevelInfoSupplier clientInfoSupplier;
@@ -377,10 +381,27 @@ public class DTLSConnectorHandshakeTest {
 	/**
 	 * Destroys the server and client.
 	 */
 	@After
 	public void cleanUp() {
+		if (serverHelper != null && serverHelper.server != null) {
+			assertThat(serverHelper.server.isRunning(), is(true));
+			try {
+				// wait until no pending jobs left
+				TestConditionTools.waitForCondition(6000, 100, TimeUnit.MILLISECONDS, new TestCondition() {
+
+					@Override
+					public boolean isFulFilled() throws IllegalStateException {
+						return !serverHelper.server.updateHealth();
+					}
+				});
+				TestConditionTools.assertStatisticCounter("jobs left", serverHealth, "pending in jobs", is(0L));
+				TestConditionTools.assertStatisticCounter("jobs left", serverHealth, "pending out jobs", is(0L));
+				TestConditionTools.assertStatisticCounter("jobs left", serverHealth, "pending handshake jobs", is(0L));
+			} catch (InterruptedException e) {
+			}
+		}
 		for (AsyncAdvancedPskStore pskStore : clientsPskStores) {
 			pskStore.shutdown();
 		}
 		clientsPskStores.clear();
 		for (AsyncNewAdvancedCertificateVerifier verfier : clientsCertificateVerifiers) {
@@ -441,11 +462,12 @@ public class DTLSConnectorHandshakeTest {
 						.setTrustAllCertificates().setTrustAllRPKs().build();
 				serverBuilder.setAdvancedCertificateVerifier(serverVerifier);
 				serverVerifier.setDelay(DtlsTestTools.DEFAULT_HANDSHAKE_RESULT_DELAY_MILLIS);
 			}
 		}
-
+		serverHealth = new DtlsHealthLogger("server");
+		serverBuilder.setHealthHandler(serverHealth);
 		serverBuilderSetup.setup(serverBuilder);
 		serverHelper.startServer();
 	}
 
 	private DTLSSession startClientPsk(String hostname) throws Exception {
diff --git a/scandium-core/src/test/java/org/eclipse/californium/scandium/DTLSConnectorTest.java b/scandium-core/src/test/java/org/eclipse/californium/scandium/DTLSConnectorTest.java
index 7804ecc1b..e879ad1d8 100644
--- a/scandium-core/src/test/java/org/eclipse/californium/scandium/DTLSConnectorTest.java
+++ b/scandium-core/src/test/java/org/eclipse/californium/scandium/DTLSConnectorTest.java
@@ -81,11 +81,10 @@ import org.eclipse.californium.elements.rule.LoggingRule;
 import org.eclipse.californium.elements.rule.TestNameLoggerRule;
 import org.eclipse.californium.elements.rule.ThreadsRule;
 import org.eclipse.californium.elements.util.DatagramReader;
 import org.eclipse.californium.elements.util.DatagramWriter;
 import org.eclipse.californium.elements.util.ExecutorsUtil;
-import org.eclipse.californium.elements.util.SerialExecutor;
 import org.eclipse.californium.elements.util.SimpleMessageCallback;
 import org.eclipse.californium.elements.util.TestThreadFactory;
 import org.eclipse.californium.scandium.ConnectorHelper.LatchDecrementingRawDataChannel;
 import org.eclipse.californium.scandium.ConnectorHelper.LatchSessionListener;
 import org.eclipse.californium.scandium.ConnectorHelper.RecordCollectorDataHandler;
@@ -819,11 +818,11 @@ public class DTLSConnectorTest {
 		DatagramWriter writer = new DatagramWriter(128);
 		connection.writeTo(writer);
 		DatagramReader reader = new DatagramReader(writer.toByteArray());
 		Connection connection2 = Connection.fromReader(reader, 0);
 		clientConnectionStore.remove(connection, true);
-		connection2.setConnectorContext(new SerialExecutor(executor), null);
+		connection2.setConnectorContext(executor, null);
 		clientConnectionStore.put(connection2);
 
 		// Restart it
 		client.restart();
 		assertEquals(clientTestContext.getClientAddress(), client.getAddress());
@@ -1033,15 +1032,15 @@ public class DTLSConnectorTest {
 	public void testConnectorTerminatesHandshakeIfConnectionStoreIsExhausted() throws Exception {
 		logging.setLoggingLevel("ERROR", InMemoryConnectionStore.class, InMemoryReadWriteLockConnectionStore.class);
 		serverHelper.serverConnectionStore.clear();
 		assertEquals(SERVER_CONNECTION_STORE_CAPACITY, serverHelper.serverConnectionStore.remainingCapacity());
 		assertTrue(serverHelper.serverConnectionStore.put(new Connection(new InetSocketAddress("192.168.0.1", 5050))
-				.setConnectorContext(new SerialExecutor(executor), null)));
+				.setConnectorContext(executor, null)));
 		assertTrue(serverHelper.serverConnectionStore.put(new Connection(new InetSocketAddress("192.168.0.2", 5050))
-				.setConnectorContext(new SerialExecutor(executor), null)));
+				.setConnectorContext(executor, null)));
 		assertTrue(serverHelper.serverConnectionStore.put(new Connection(new InetSocketAddress("192.168.0.3", 5050))
-				.setConnectorContext(new SerialExecutor(executor), null)));
+				.setConnectorContext(executor, null)));
 
 		LatchDecrementingRawDataChannel clientRawDataChannel = new LatchDecrementingRawDataChannel(1);
 		client.setRawDataReceiver(clientRawDataChannel);
 		client.start();
 		InetSocketAddress clientEndpoint = client.getAddress();

Vulnerable code: 	 * Shutdown this executor and add all pending task from {@link #tasks} to
								setOwner();
								ExecutionListener current = listener.get();
										current.beforeExecution();
									command.run();
									LOGGER.error("unexpected error occurred:", t);
								} finally {
									try {
										if (current != null) {
											current.afterExecution();
										}
									} catch (Throwable t) {
										LOGGER.error("unexpected error occurred:", t);
									}
									clearOwner();
							} finally {
						} catch (RejectedExecutionException ex) {
							LOGGER.debug("shutdown?", ex);
	 * Called before and after executing a task.
						} else if (health instanceof DtlsHealthExtended) {
							((DtlsHealthExtended) health)
									.setConnections(maxConnections - connectionStore.remainingCapacity());
		for (Runnable job : pending) {
			try {
				job.run();
			} catch (Exception e) {
				LOGGER.warn("Shutdown DTLS connector:", e);
			}
		}
			executeInbound(getExecutorService(), peerAddress, new Runnable() {
					if (MDC_SUPPORT) {
						MDC.put("PEER", StringUtil.toString(firstRecord.getPeerAddress()));
					}
					processNewClientHello(firstRecord);
					if (MDC_SUPPORT) {
						MDC.clear();
					pendingInboundJobsCountdown.incrementAndGet();
				if (!executeInbound(serialExecutor, peerAddress, new Runnable() {
							if (running.get()) {
							pendingInboundJobsCountdown.incrementAndGet();
	protected boolean executeInbound(Executor executor, InetSocketAddress peer, Runnable job) {
		boolean pending = false;
			int count = pendingInboundJobsCountdown.decrementAndGet();
			if (count >= 0) {
				executor.execute(job);
				pending = true;
				if (health instanceof DtlsHealthExtended2) {
					((DtlsHealthExtended2) health).setPendingIncomingJobs(maxPendingInboundJobs - count);
				}
			} else {
			}
		} catch (RejectedExecutionException e) {
			// dont't terminate connection on shutdown!
			LOGGER.debug("Execution rejected while processing record from peer [{}]",
					StringUtil.toLog(peer), e);
		} finally {
			if (!pending) {
				if (health != null) {
					health.receivingRecord(true);
				}
				pendingInboundJobsCountdown.incrementAndGet();
		return pending;
		try {
			SerialExecutor executor = connection.getExecutor();
			if (!executeOutbound(executor, message.getInetSocketAddress(), new Runnable() {
				@Override
				public void run() {
					try {
						if (running.get()) {
							sendMessage(now, message, connection);
						} else {
							DROP_LOGGER.trace("DTLSConnector drops {} outgoing bytes to {}, connector not running!",
									message.getSize(), StringUtil.toLog(message.getInetSocketAddress()));
							message.onError(new InterruptedIOException("Connector is not running."));
							if (health != null) {
								health.sendingRecord(true);
							}
						}
					} catch (Exception e) {
						if (running.get()) {
							LOGGER.warn("Exception thrown by executor thread [{}]",
									Thread.currentThread().getName(), e);
						}
						DROP_LOGGER.trace("DTLSConnector drops {} outgoing bytes to {} {}", message.getSize(),
								StringUtil.toLog(message.getInetSocketAddress()), e.getMessage());
						message.onError(e);
					} finally {
						pendingOutboundJobsCountdown.incrementAndGet();
			})) {
				message.onError(new IllegalStateException("Outbound message overflow!"));
		} catch (RejectedExecutionException e) {
			LOGGER.debug("Execution rejected while sending application record [peer: {}]",
					StringUtil.toLog(message.getInetSocketAddress()), e);
			DROP_LOGGER.trace("DTLSConnector drops {} outgoing bytes to {}, {}", message.getSize(),
					StringUtil.toLog(message.getInetSocketAddress()), e.getMessage());
			message.onError(new InterruptedIOException("Connector is not running."));
	protected boolean executeOutbound(Executor executor, InetSocketAddress peer, Runnable job) {
		boolean pending = false;
			int count = pendingOutboundJobsCountdown.decrementAndGet();
			if (count >= 0) {
				executor.execute(job);
				pending = true;
				if (health instanceof DtlsHealthExtended2) {
					((DtlsHealthExtended2) health).setPendingOutgoingJobs(maxPendingOutboundJobs - count);
				}
			} else {
			}
		} finally {
			if (!pending) {
				if (health != null) {
					health.sendingRecord(true);
				}
				pendingOutboundJobsCountdown.incrementAndGet();
		return pending;
					executeHandshakeResult(serialExecutor, connection, new Runnable() {
								if (running.get()) {
								pendingHandshakeResultJobsCountdown.incrementAndGet();
				} catch (RejectedExecutionException e) {
					// dont't terminate connection on shutdown!
					LOGGER.debug("Execution rejected while processing handshake result [{}]", connection);
	protected boolean executeHandshakeResult(Executor executor, Connection connection, Runnable job) {
		boolean pending = false;
			int count = pendingHandshakeResultJobsCountdown.decrementAndGet();
			if (count >= 0) {
				executor.execute(job);
				pending = true;
				if (health instanceof DtlsHealthExtended2) {
					((DtlsHealthExtended2) health).setPendingHandshakeJobs(maxPendingHandshakeResultJobs - count);
				}
			} else {
			}
		} finally {
			if (!pending) {
				pendingHandshakeResultJobsCountdown.incrementAndGet();
		return pending;
	 * {@link DTLSConnector#executeInbound(Executor, InetSocketAddress, Runnable)}
	private SerialExecutor serialExecutor;
				if (connection.isExecuting()) {
					List<Runnable> pendings = connection.getExecutor().shutdownNow();
					if (LOGGER.isTraceEnabled()) {
						LOGGER.trace("{}connection: remove {} (size {}, left jobs: {})", tag, connection,
								connections.size(), pendings.size(), new Throwable("connection removed!"));
					} else if (pendings.isEmpty()) {
						LOGGER.debug("{}connection: remove {} (size {})", tag, connection, connections.size());
					} else {
						LOGGER.debug("{}connection: remove {} (size {}, left jobs: {})", tag, connection,
								connections.size(), pendings.size());
					}
					if (LOGGER.isTraceEnabled()) {
						LOGGER.trace("{}connection: remove {} (size {})", tag, connection, connections.size(),
								new Throwable("connection removed!"));
					} else {
						LOGGER.debug("{}connection: remove {} (size {})", tag, connection, connections.size());
					}
				if (connection.isExecuting()) {
					List<Runnable> pendings = connection.getExecutor().shutdownNow();
					if (LOGGER.isTraceEnabled()) {
						LOGGER.trace("{}connection: remove {} (size {}, left jobs: {})", tag, connection,
								connections.size(), pendings.size(), new Throwable("connection removed!"));
					} else if (pendings.isEmpty()) {
						LOGGER.debug("{}connection: remove {} (size {})", tag, connection, connections.size());
					} else {
						LOGGER.debug("{}connection: remove {} (size {}, left jobs: {})", tag, connection,
								connections.size(), pendings.size());
					}
					if (LOGGER.isTraceEnabled()) {
						LOGGER.trace("{}connection: remove {} (size {})", tag, connection, connections.size(),
								new Throwable("connection removed!"));
					} else {
						LOGGER.debug("{}connection: remove {} (size {})", tag, connection, connections.size());
					}
	public void testDisableHelloVerifRequestForPsk() throws Exception {
	public void testDisabledHelloVerifRequestForPskWithCertificate() throws Exception {
	public void testDisabledHelloVerifRequestWithCertificate() throws Exception {
		connection.setConnectorContext(new SerialExecutor(executor), null);

import org.eclipse.californium.elements.util.SerialExecutor;
		connection2.setConnectorContext(new SerialExecutor(executor), null);
				.setConnectorContext(new SerialExecutor(executor), null)));
				.setConnectorContext(new SerialExecutor(executor), null)));
				.setConnectorContext(new SerialExecutor(executor), null)));
Fixed code: 	/**
	 * Run all jobs.
	 * 
	 * On {@link Throwable} write only warning and continue to run the other
	 * jobs.
	 * 
	 * @param jobs list of jobs to run.
	 * @since 3.7
	 */
	public static void runAll(List<Runnable> jobs) {
		for (Runnable job : jobs) {
			try {
				job.run();
			} catch (Throwable e) {
				LOGGER.warn("Ignoring error:", e);
			}
		}
	}

/*******************************************************************************
 * Copyright (c) 2022 Contributors to the Eclipse Foundation.
 * 
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v2.0
 * and Eclipse Distribution License v1.0 which accompany this distribution.
 * 
 * The Eclipse Public License is available at
 *    http://www.eclipse.org/legal/epl-v20.html
 * and the Eclipse Distribution License is available at
 *    http://www.eclipse.org/org/documents/edl-v10.html.
 * 
 ******************************************************************************/
package org.eclipse.californium.elements.util;

import java.util.concurrent.Executor;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Limited jobs.
 *
 * Limits pending jobs based on counters.
 * 
 * Note: for proper operation, ensure the at the end of
 * {@link Runnable#run()} the {@link LimitedRunnable#onDequeueing()} is
 * called.
 * 
 * <pre>
 *	public void run() {
 *		try {
 *			if (running.get() &amp;&amp; connection.isExecuting()) {
 *				// process
 *			}
 *		} finally {
 *			onDequeueing();
 *		}
 *	}
 * </pre>
 * 
 * @since 3.7
 */
public abstract class LimitedRunnable implements Runnable {

	/**
	 * Down counter to limit pending jobs.
	 */
	private final AtomicInteger counter;
	/**
	 * Indicator for overflows. {@code true}, if the counter indicates an
	 * overflow, {@code false}, otherwise.
	 */
	private volatile boolean overflow;

	/**
	 * Create job limited by the provided counter
	 * 
	 * @param counter counter in count-down mode
	 */
	public LimitedRunnable(AtomicInteger counter) {
		this.counter = counter;
	}

	/**
	 * Queue job.
	 * 
	 * @throws RejectedExecutionException if limit is exceeded
	 */
	public void onQueueing() {
		if (counter.decrementAndGet() < 0) {
			overflow = true;
			throw new RejectedExecutionException("queue overflow!");
		}
	}

	/**
	 * Dequeue job.
	 */
	public void onDequeueing() {
		counter.incrementAndGet();
	}

	/**
	 * Checks, if queueing this job causes an counter overflow.
	 * 
	 * @return {@code true}, if the counter indicates an overflow,
	 *         {@code false}, otherwise.
	 */
	public boolean isOverflown() {
		return overflow;
	}

	/**
	 * Execute this job.
	 * 
	 * @param executor executor to execute jobs.
	 */
	public void execute(Executor executor) {
		try {
			onQueueing();
			executor.execute(this);
		} catch (RejectedExecutionException ex) {
			onDequeueing();
			throw ex;
		}
	}
}
	 * @throws IllegalArgumentException if the executor is also a
	 *             {@link SerialExecutor}
		} else if (executor instanceof SerialExecutor) {
			throw new IllegalArgumentException("Sequences of SerialExecutors are not supported!");
	 * Shutdown this executor and add all pending jobs from {@link #tasks} to
							setOwner();
							ExecutionListener current = listener.get();
								if (current != null) {
									current.beforeExecution();
								}
								command.run();
							} catch (Throwable t) {
								LOGGER.error("unexpected error occurred:", t);
							} finally {
										current.afterExecution();
									LOGGER.error("unexpected error occurred after execution:", t);
								clearOwner();
							}
						} finally {
							try {
							} catch (RejectedExecutionException ex) {
								LOGGER.debug("shutdown?", ex);
	 * Called before and after executing a task. The calling thread is the same
	 * as the the one executing the job.
	],
	"3.7.0": [
		{
			"extension": "revapi.differences",
			"configuration": {
				"ignore": true,
				"differences": [
					{
						"code": "java.method.removed",
						"old": "method boolean org.eclipse.californium.scandium.DTLSConnector::executeHandshakeResult(java.util.concurrent.Executor, org.eclipse.californium.scandium.dtls.Connection, java.lang.Runnable)",
						"justification": "Not part of the public API"
					},
					{
						"code": "java.method.removed",
						"old": "method boolean org.eclipse.californium.scandium.DTLSConnector::executeInbound(java.util.concurrent.Executor, java.net.InetSocketAddress, java.lang.Runnable)",
						"justification": "Not part of the public API"
					},
					{
						"code": "java.method.removed",
						"old": "method boolean org.eclipse.californium.scandium.DTLSConnector::executeOutbound(java.util.concurrent.Executor, java.net.InetSocketAddress, java.lang.Runnable)",
						"justification": "Not part of the public API"
					}
				]
			}
		}
import org.eclipse.californium.elements.util.LimitedRunnable;
	/**
	 * Update health statistic.
	 * 
	 * Update {@link DtlsHealthExtended#setConnections(int)},
	 * {@link DtlsHealthExtended2#setPendingIncomingJobs(int)},
	 * {@link DtlsHealthExtended2#setPendingOutgoingJobs(int)}, and
	 * {@link DtlsHealthExtended2#setPendingHandshakeJobs(int)}.
	 * 
	 * @return {@code true}, if some pending jobs left, {@code false}, if not.
	 * @since 3.7
	 */
	public boolean updateHealth() {
		boolean pending = false;
		if (health instanceof DtlsHealthExtended) {
			((DtlsHealthExtended) health).setConnections(maxConnections - connectionStore.remainingCapacity());
		}
		if (health instanceof DtlsHealthExtended2) {
			DtlsHealthExtended2 health2 = (DtlsHealthExtended2) health;
			int jobs = maxPendingOutboundJobs - pendingOutboundJobsCountdown.get();
			health2.setPendingOutgoingJobs(jobs);
			if (jobs > 0) {
				LOGGER.debug("Pending out jobs {}", jobs);
			}
			pending = jobs > 0;
			jobs = maxPendingInboundJobs - pendingInboundJobsCountdown.get();
			health2.setPendingIncomingJobs(jobs);
			if (jobs > 0) {
				LOGGER.debug("Pending in jobs {}", jobs);
			}
			pending |= jobs > 0;
			jobs = maxPendingHandshakeResultJobs - pendingHandshakeResultJobsCountdown.get();
			health2.setPendingHandshakeJobs(jobs);
			if (jobs > 0) {
				LOGGER.debug("Pending handshake jobs {}", jobs);
			}
			pending |= jobs > 0;
		}
		return pending;
	}

						} else {
							updateHealth();
			updateHealth();
		ExecutorsUtil.runAll(pending);
			executeInbound(getExecutorService(), peerAddress, new LimitedRunnable(pendingInboundJobsCountdown) {
					try {
						if (running.get()) {
							if (MDC_SUPPORT) {
								MDC.put("PEER", StringUtil.toString(firstRecord.getPeerAddress()));
							}
							processNewClientHello(firstRecord);
							if (MDC_SUPPORT) {
								MDC.clear();
							}
						}
					} finally {
						onDequeueing();
				if (!executeInbound(serialExecutor, peerAddress, new LimitedRunnable(pendingInboundJobsCountdown) {
							if (running.get() && connection.isExecuting()) {
							onDequeueing();
	@NoPublicAPI
	protected boolean executeInbound(Executor executor, InetSocketAddress peer, LimitedRunnable job) {
			job.execute(executor);
			return true;
		} catch (RejectedExecutionException e) {
			if (job.isOverflown()) {
			} else {
				// dont't terminate connection on shutdown!
				LOGGER.debug("Execution rejected while processing record from peer [{}]", StringUtil.toLog(peer), e);
		if (health != null) {
			health.receivingRecord(true);
		}
		return false;
		SerialExecutor executor = connection.getExecutor();
		if (!executeOutbound(executor, message.getInetSocketAddress(), new LimitedRunnable(pendingOutboundJobsCountdown) {
			@Override
			public void run() {
				try {
					if (running.get() && connection.isExecuting()) {
						sendMessage(now, message, connection);
					} else {
						DROP_LOGGER.trace("DTLSConnector drops {} outgoing bytes to {}, connector not running!",
								message.getSize(), StringUtil.toLog(message.getInetSocketAddress()));
						message.onError(new InterruptedIOException("Connector is not running."));
				} catch (Exception e) {
					if (running.get()) {
						LOGGER.warn("Exception thrown by executor thread [{}]", Thread.currentThread().getName(), e);
					}
					DROP_LOGGER.trace("DTLSConnector drops {} outgoing bytes to {} {}", message.getSize(),
							StringUtil.toLog(message.getInetSocketAddress()), e.getMessage());
					if (health != null) {
						health.sendingRecord(true);
					}
					message.onError(e);
				} finally {
					onDequeueing();
		})) {
			message.onError(new IllegalStateException("Outbound message overflow!"));
	@NoPublicAPI
	protected boolean executeOutbound(Executor executor, InetSocketAddress peer, LimitedRunnable job) {
			job.execute(executor);
			return true;
		} catch (RejectedExecutionException e) {
			if (job.isOverflown()) {
			} else {
				// dont't terminate connection on shutdown!
				LOGGER.debug("Execution rejected while processing record to peer [{}]", StringUtil.toLog(peer), e);
		if (health != null) {
			health.sendingRecord(true);
		}
		return false;
					executeHandshakeResult(serialExecutor, connection, new LimitedRunnable(pendingHandshakeResultJobsCountdown) {
								if (running.get() && connection.isExecuting()) {
								onDequeueing();
	@NoPublicAPI
	protected boolean executeHandshakeResult(Executor executor, Connection connection, LimitedRunnable job) {
			job.execute(executor);
			return true;
		} catch (RejectedExecutionException e) {
			if (job.isOverflown()) {
			} else {
				// dont't terminate connection on shutdown!
				LOGGER.debug("Execution rejected while processing handshake result [{}]", connection, e);
		return false;
	 * {@link DTLSConnector#executeInbound(Executor, InetSocketAddress, org.eclipse.californium.scandium.DTLSConnector.LimitedRunnable)}
import java.util.List;
import org.eclipse.californium.elements.util.ExecutorsUtil;
	private volatile SerialExecutor serialExecutor;
	/**
	 * Shutdown executor and run all pending jobs.
	 * 
	 * The jobs are intended to check {@link #isExecuting()} in order to detect
	 * the shutdown.
	 * 
	 * @return number of pending jobs.
	 * @since 3.7
	 */
	public int shutdown() {
		SerialExecutor executor = getExecutor();
		if (executor != null) {
			List<Runnable> pendings = executor.shutdownNow();
			ExecutorsUtil.runAll(pendings);
			return pendings.size();
		}
		return 0;
	}

				int pendings = connection.shutdown();
				if (LOGGER.isTraceEnabled()) {
					LOGGER.trace("{}connection: remove {} (size {}, left jobs: {})", tag, connection,
							connections.size(), pendings, new Throwable("connection removed!"));
				} else if (pendings == 0) {
					LOGGER.debug("{}connection: remove {} (size {})", tag, connection, connections.size());
					LOGGER.debug("{}connection: remove {} (size {}, left jobs: {})", tag, connection,
							connections.size(), pendings);
				int pendings = connection.shutdown();
				if (LOGGER.isTraceEnabled()) {
					LOGGER.trace("{}connection: remove {} (size {}, left jobs: {})", tag, connection,
							connections.size(), pendings, new Throwable("connection removed!"));
				} else if (pendings == 0) {
					LOGGER.debug("{}connection: remove {} (size {})", tag, connection, connections.size());
					LOGGER.debug("{}connection: remove {} (size {}, left jobs: {})", tag, connection,
							connections.size(), pendings);
	public void testDisableHelloVerifyRequestForPsk() throws Exception {
	public void testDisabledHelloVerifyRequestForPskWithCertificate() throws Exception {
	public void testDisabledHelloVerifyRequestWithCertificate() throws Exception {
		connection.setConnectorContext(executor, null);
import org.eclipse.californium.elements.util.TestCondition;
import org.eclipse.californium.elements.util.TestConditionTools;
	DtlsHealthLogger serverHealth;

		if (serverHelper != null && serverHelper.server != null) {
			assertThat(serverHelper.server.isRunning(), is(true));
			try {
				// wait until no pending jobs left
				TestConditionTools.waitForCondition(6000, 100, TimeUnit.MILLISECONDS, new TestCondition() {

					@Override
					public boolean isFulFilled() throws IllegalStateException {
						return !serverHelper.server.updateHealth();
					}
				});
				TestConditionTools.assertStatisticCounter("jobs left", serverHealth, "pending in jobs", is(0L));
				TestConditionTools.assertStatisticCounter("jobs left", serverHealth, "pending out jobs", is(0L));
				TestConditionTools.assertStatisticCounter("jobs left", serverHealth, "pending handshake jobs", is(0L));
			} catch (InterruptedException e) {
			}
		}
		serverHealth = new DtlsHealthLogger("server");
		serverBuilder.setHealthHandler(serverHealth);
		connection2.setConnectorContext(executor, null);
				.setConnectorContext(executor, null)));
				.setConnectorContext(executor, null)));
				.setConnectorContext(executor, null)));