diff --git a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/HttpFunctionProvider.java b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/HttpFunctionProvider.java
index 23001ee..e1ef27e 100644
--- a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/HttpFunctionProvider.java
+++ b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/HttpFunctionProvider.java
@@ -39,6 +39,8 @@
   /** lazily initialized by {code buildHttpClient} */
   @Nullable private OkHttpClient sharedClient;
 
+  private volatile boolean shutdown;
+
   public HttpFunctionProvider(Map<FunctionType, HttpFunctionSpec> supportedTypes) {
     this.supportedTypes = supportedTypes;
   }
@@ -84,11 +86,12 @@
     } else {
       url = HttpUrl.get(spec.endpoint());
     }
-    return new HttpRequestReplyClient(url, clientBuilder.build());
+    return new HttpRequestReplyClient(url, clientBuilder.build(), () -> shutdown);
   }
 
   @Override
   public void shutdown() {
+    shutdown = true;
     OkHttpUtils.closeSilently(sharedClient);
   }
 }
diff --git a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/HttpRequestReplyClient.java b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/HttpRequestReplyClient.java
index d9e5cb3..efff157 100644
--- a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/HttpRequestReplyClient.java
+++ b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/HttpRequestReplyClient.java
@@ -24,6 +24,7 @@
 import java.io.InputStream;
 import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
+import java.util.function.BooleanSupplier;
 import okhttp3.Call;
 import okhttp3.HttpUrl;
 import okhttp3.MediaType;
@@ -43,10 +44,12 @@
 
   private final HttpUrl url;
   private final OkHttpClient client;
+  private final BooleanSupplier isShutdown;
 
-  HttpRequestReplyClient(HttpUrl url, OkHttpClient client) {
+  HttpRequestReplyClient(HttpUrl url, OkHttpClient client, BooleanSupplier isShutdown) {
     this.url = Objects.requireNonNull(url);
     this.client = Objects.requireNonNull(client);
+    this.isShutdown = Objects.requireNonNull(isShutdown);
   }
 
   @Override
@@ -61,7 +64,8 @@
             .build();
 
     Call newCall = client.newCall(request);
-    RetryingCallback callback = new RetryingCallback(requestSummary, metrics, newCall.timeout());
+    RetryingCallback callback =
+        new RetryingCallback(requestSummary, metrics, newCall.timeout(), isShutdown);
     callback.attachToCall(newCall);
     return callback.future().thenApply(HttpRequestReplyClient::parseResponse);
   }
diff --git a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/RetryingCallback.java b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/RetryingCallback.java
index 0a6f834..98184d5 100644
--- a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/RetryingCallback.java
+++ b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/RetryingCallback.java
@@ -22,9 +22,11 @@
 import java.time.Duration;
 import java.util.Arrays;
 import java.util.HashSet;
+import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
+import java.util.function.BooleanSupplier;
 import okhttp3.Call;
 import okhttp3.Callback;
 import okhttp3.Response;
@@ -49,15 +51,20 @@
   private final BoundedExponentialBackoff backoff;
   private final ToFunctionRequestSummary requestSummary;
   private final RemoteInvocationMetrics metrics;
+  private final BooleanSupplier isShutdown;
 
   private long requestStarted;
 
   RetryingCallback(
-      ToFunctionRequestSummary requestSummary, RemoteInvocationMetrics metrics, Timeout timeout) {
+      ToFunctionRequestSummary requestSummary,
+      RemoteInvocationMetrics metrics,
+      Timeout timeout,
+      BooleanSupplier isShutdown) {
     this.resultFuture = new CompletableFuture<>();
     this.backoff = new BoundedExponentialBackoff(INITIAL_BACKOFF_DURATION, duration(timeout));
     this.requestSummary = requestSummary;
     this.metrics = metrics;
+    this.isShutdown = Objects.requireNonNull(isShutdown);
   }
 
   CompletableFuture<Response> future() {