feat: add Microservices Bulkhead pattern (#3228) (#3597)

* feat: add Microservices Bulkhead pattern (#3228)

* fix: cancel queued calls on bulkhead shutdown

Shutdown now cancels the queued futures so callers do not hang, a shutdown racing a submit is no longer counted as a rejection, and interrupted inventory checks are logged. The class diagram is a rendered PNG.

---------

Co-authored-by: Ilkka Seppälä <iluwatar@users.noreply.github.com>
This commit is contained in:
DoksanbirandIlkka Seppälä authored and GitHub committed 2026-09-13 09:53:54 +03:00
1 parent dd24a29f80
commit 4cabb204f2
15 files changed
+1352

No files matched your search

@@ -0,0 +1,134 @@
/*
* This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt).
*
* The MIT License
* Copyright © 2014-2022 Ilkka Seppälä
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in
* all copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
* THE SOFTWARE.
*/
package com.iluwatar.bulkhead;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import lombok.extern.slf4j.Slf4j;
/**
* The Bulkhead pattern partitions the resources of a service so that a failure or an overload in
* one dependency cannot sink the whole service, just like the watertight compartments in a ship's
* hull keep a single leak from flooding the entire vessel.
*
* <p>In this example an order service calls two downstream dependencies: a payment provider that
* has become very slow and an inventory system that is perfectly healthy. The demo first sends both
* kinds of calls through one shared thread pool. The slow payment calls fill every thread and the
* queue, so a request for the healthy inventory system is rejected although nothing is wrong with
* it. The demo then gives each dependency its own {@link Bulkhead}. The payment compartment still
* saturates and rejects the excess calls fast, but the inventory compartment keeps answering
* immediately because the payment calls can no longer consume its threads.
*/
@Slf4j
public class App {
private static final Duration PAYMENT_LATENCY = Duration.ofMillis(300);
private static final Duration WAIT_TIMEOUT = Duration.ofSeconds(5);
/**
* Program entry point.
*
* @param args command line arguments, not used
*/
public static void main(String[] args) {
var payment = new PaymentService(PAYMENT_LATENCY);
var inventory = new InventoryService();
LOGGER.info("--- Scenario 1: one shared thread pool for every downstream call ---");
try (var sharedPool = new Bulkhead("shared-pool", 2, 2)) {
var paymentFutures = flood(sharedPool, payment, "order", 4);
callInventory(sharedPool, inventory, "order-5");
awaitAll(paymentFutures);
}
LOGGER.info("--- Scenario 2: a dedicated bulkhead for each downstream dependency ---");
try (var paymentBulkhead = new Bulkhead("payment", 2, 2);
var inventoryBulkhead = new Bulkhead("inventory", 2, 2)) {
var paymentFutures = flood(paymentBulkhead, payment, "order", 10);
for (var i = 1; i <= 3; i++) {
callInventory(inventoryBulkhead, inventory, "order-" + i);
}
awaitAll(paymentFutures);
LOGGER.info(
"Bulkhead '{}' rejected {} of 10 calls, bulkhead '{}' rejected {} of 3 calls",
paymentBulkhead.getName(),
paymentBulkhead.getRejectedCalls(),
inventoryBulkhead.getName(),
inventoryBulkhead.getRejectedCalls());
}
}
static List<Future<String>> flood(
Bulkhead bulkhead, RemoteService service, String requestPrefix, int calls) {
var accepted = new ArrayList<Future<String>>();
for (var i = 1; i <= calls; i++) {
var request = requestPrefix + "-" + i;
try {
accepted.add(bulkhead.submit(() -> service.call(request)));
} catch (BulkheadFullException e) {
LOGGER.info("Request '{}' rejected immediately: {}", request, e.getMessage());
}
}
return accepted;
}
static void callInventory(Bulkhead bulkhead, RemoteService inventory, String request) {
try {
var response = bulkhead.submit(() -> inventory.call(request));
LOGGER.info(
"Inventory response: {}", response.get(WAIT_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS));
} catch (BulkheadFullException e) {
LOGGER.error(
"Inventory check for '{}' rejected although the inventory system is healthy: {}",
request,
e.getMessage());
} catch (ExecutionException | TimeoutException e) {
LOGGER.error("Inventory check for '{}' failed", request, e);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
LOGGER.warn(
"Inventory check for '{}' was interrupted while waiting for the response", request);
}
}
static void awaitAll(List<Future<String>> futures) {
for (var future : futures) {
try {
LOGGER.info(
"Payment response: {}", future.get(WAIT_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS));
} catch (ExecutionException | TimeoutException e) {
LOGGER.error("Payment call failed", e);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return;
}
}
}
}
@@ -0,0 +1,157 @@
/*
* This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt).
*
* The MIT License
* Copyright © 2014-2022 Ilkka Seppälä
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in
* all copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
* THE SOFTWARE.
*/
package com.iluwatar.bulkhead;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
/**
* Isolates the calls made to one downstream dependency inside a dedicated, bounded thread pool.
*
* <p>The pool has a fixed number of worker threads and a bounded waiting queue. When both are full
* the bulkhead does not block the caller and it cannot borrow threads from anywhere else: the call
* fails fast with a {@link BulkheadFullException}. Each dependency gets its own instance, so a slow
* or hanging dependency can only exhaust its own compartment while the rest of the system keeps
* serving requests.
*/
@Slf4j
public class Bulkhead implements AutoCloseable {
@Getter private final String name;
@Getter private final int maxConcurrentCalls;
@Getter private final int maxQueueSize;
private final ThreadPoolExecutor executor;
private final AtomicLong rejectedCalls = new AtomicLong();
/**
* Creates a bulkhead with its own thread pool.
*
* @param name name of the compartment, used in logs and worker thread names
* @param maxConcurrentCalls number of calls that may run at the same time
* @param maxQueueSize number of calls that may wait for a free thread; zero disables queueing
*/
public Bulkhead(String name, int maxConcurrentCalls, int maxQueueSize) {
if (maxConcurrentCalls < 1) {
throw new IllegalArgumentException("maxConcurrentCalls must be at least 1");
}
if (maxQueueSize < 0) {
throw new IllegalArgumentException("maxQueueSize must not be negative");
}
this.name = name;
this.maxConcurrentCalls = maxConcurrentCalls;
this.maxQueueSize = maxQueueSize;
BlockingQueue<Runnable> queue =
maxQueueSize == 0 ? new SynchronousQueue<>() : new ArrayBlockingQueue<>(maxQueueSize);
var threadCounter = new AtomicInteger();
this.executor =
new ThreadPoolExecutor(
maxConcurrentCalls,
maxConcurrentCalls,
0L,
TimeUnit.MILLISECONDS,
queue,
runnable ->
new Thread(runnable, "bulkhead-" + name + "-" + threadCounter.incrementAndGet()),
new ThreadPoolExecutor.AbortPolicy());
}
/**
* Submits a call to this compartment.
*
* @param task the call to execute
* @param <T> type of the result
* @return a future that completes with the result of the call
* @throws BulkheadFullException if every thread is busy and the queue is full
* @throws IllegalStateException if the bulkhead has been shut down
*/
public <T> Future<T> submit(Callable<T> task) {
if (executor.isShutdown()) {
throw new IllegalStateException("Bulkhead '" + name + "' is shut down");
}
try {
var future = executor.submit(task);
LOGGER.debug(
"Bulkhead '{}' accepted call ({} active, {} queued)",
name,
executor.getActiveCount(),
executor.getQueue().size());
return future;
} catch (RejectedExecutionException e) {
if (executor.isShutdown()) {
throw new IllegalStateException("Bulkhead '" + name + "' is shut down");
}
rejectedCalls.incrementAndGet();
LOGGER.warn(
"Bulkhead '{}' is full ({} active, {} queued), rejecting call",
name,
executor.getActiveCount(),
executor.getQueue().size());
throw new BulkheadFullException(name);
}
}
/** Number of calls currently running. */
public int getActiveCalls() {
return executor.getActiveCount();
}
/** Number of calls waiting for a free thread. */
public int getQueuedCalls() {
return executor.getQueue().size();
}
/** Number of calls rejected since the bulkhead was created. */
public long getRejectedCalls() {
return rejectedCalls.get();
}
/**
* Stops the compartment, interrupting calls that are still running and cancelling the calls that
* were still waiting in the queue, so that callers blocked on their future are released instead
* of waiting for a result that will never be produced.
*/
public void shutdown() {
for (var pending : executor.shutdownNow()) {
if (pending instanceof Future<?> future) {
future.cancel(false);
}
}
}
@Override
public void close() {
shutdown();
}
}
@@ -0,0 +1,46 @@
/*
* This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt).
*
* The MIT License
* Copyright © 2014-2022 Ilkka Seppälä
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in
* all copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
* THE SOFTWARE.
*/
package com.iluwatar.bulkhead;
/**
* Thrown by a {@link Bulkhead} when a call cannot be accepted because every worker thread is busy
* and the waiting queue is full. The caller receives this exception immediately instead of
* blocking, which is the fail-fast behaviour the pattern relies on: the caller can degrade
* gracefully while the overloaded dependency keeps consuming only the capacity reserved for it.
*/
public class BulkheadFullException extends RuntimeException {
private final String bulkheadName;
public BulkheadFullException(String bulkheadName) {
super("Bulkhead '" + bulkheadName + "' is full, call rejected");
this.bulkheadName = bulkheadName;
}
/** Name of the bulkhead that rejected the call. */
public String getBulkheadName() {
return bulkheadName;
}
}
@@ -0,0 +1,41 @@
/*
* This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt).
*
* The MIT License
* Copyright © 2014-2022 Ilkka Seppälä
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in
* all copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
* THE SOFTWARE.
*/
package com.iluwatar.bulkhead;
import lombok.extern.slf4j.Slf4j;
/**
* Simulates a healthy inventory system that answers immediately. It represents the dependency that
* should keep working even while another dependency is overloaded.
*/
@Slf4j
public class InventoryService implements RemoteService {
@Override
public String call(String request) {
LOGGER.info("Inventory system received '{}'", request);
return "Inventory reserved for " + request;
}
}
@@ -0,0 +1,60 @@
/*
* This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt).
*
* The MIT License
* Copyright © 2014-2022 Ilkka Seppälä
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in
* all copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
* THE SOFTWARE.
*/
package com.iluwatar.bulkhead;
import java.time.Duration;
import lombok.extern.slf4j.Slf4j;
/**
* Simulates a payment provider that has become slow. Every call takes the configured latency, so a
* burst of payment requests keeps the threads that serve it busy for a long time. Without a
* bulkhead these calls would also occupy the threads needed by healthy dependencies.
*/
@Slf4j
public class PaymentService implements RemoteService {
private final Duration latency;
/**
* Creates a payment service.
*
* @param latency time every call takes to complete
*/
public PaymentService(Duration latency) {
this.latency = latency;
}
@Override
public String call(String request) {
LOGGER.info("Payment provider received '{}', it will take {} ms", request, latency.toMillis());
try {
Thread.sleep(latency);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new IllegalStateException("Payment for '" + request + "' was interrupted", e);
}
return "Payment approved for " + request;
}
}
@@ -0,0 +1,42 @@
/*
* This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt).
*
* The MIT License
* Copyright © 2014-2022 Ilkka Seppälä
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in
* all copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
* THE SOFTWARE.
*/
package com.iluwatar.bulkhead;
/**
* A remote dependency that a service calls over the network, such as a payment provider or an
* inventory system. Every call is routed through a {@link Bulkhead} so that the caller never lets
* one dependency monopolise its threads.
*/
@FunctionalInterface
public interface RemoteService {
/**
* Performs the remote call.
*
* @param request identifier of the request, used in the response and in the logs
* @return the response of the dependency
*/
String call(String request);
}