diff --git a/build.sbt b/build.sbt
index dde3ec25..d816d1a4 100644
--- a/build.sbt
+++ b/build.sbt
@@ -1,4 +1,5 @@
 import com.typesafe.tools.mima.core._
+import explicitdeps.ExplicitDepsPlugin.autoImport.moduleFilterRemoveValue
 
 lazy val root = project
   .in(file("."))
diff --git a/core/src/main/scala/org/http4s/jdkhttpclient/JdkHttpClient.scala b/core/src/main/scala/org/http4s/jdkhttpclient/JdkHttpClient.scala
index 1c0472d7..b32eb82b 100644
--- a/core/src/main/scala/org/http4s/jdkhttpclient/JdkHttpClient.scala
+++ b/core/src/main/scala/org/http4s/jdkhttpclient/JdkHttpClient.scala
@@ -243,9 +243,22 @@ object JdkHttpClient {
     * [[cats.effect.kernel.Async.executor executor]], sets the
     * [[org.http4s.client.defaults.ConnectTimeout default http4s connect timeout]], and disables
     * [[https://github.com/http4s/http4s-jdk-http-client/issues/200 TLS 1.3 on JDK 11]].
+    *
+    * On Java 21 and higher, it actively closes the underlying client, releasing its resources
+    * early. On earlier Java versions, closing the underlying client is not possible, so the release
+    * is a no-op. On these Java versions (and there only), you can safely use
+    * [[cats.effect.Resource allocated]] to avoid dealing with resource management.
     */
-  def simple[F[_]](implicit F: Async[F]): F[Client[F]] =
-    defaultHttpClient[F].map(apply(_))
+  def simple[F[_]](implicit F: Async[F]): Resource[F, Client[F]] =
+    defaultHttpClientResource[F].map(apply(_))
+
+  private[jdkhttpclient] def defaultHttpClientResource[F[_]](implicit
+      F: Async[F]
+  ): Resource[F, HttpClient] =
+    Resource.make[F, HttpClient](defaultHttpClient[F]) {
+      case c: AutoCloseable => Sync[F].blocking(c.close())
+      case _ => Applicative[F].unit
+    }
 
   private[jdkhttpclient] def defaultHttpClient[F[_]](implicit F: Async[F]): F[HttpClient] =
     F.executor.flatMap { exec =>
diff --git a/core/src/main/scala/org/http4s/jdkhttpclient/JdkWSClient.scala b/core/src/main/scala/org/http4s/jdkhttpclient/JdkWSClient.scala
index f8d128e1..49488e2f 100644
--- a/core/src/main/scala/org/http4s/jdkhttpclient/JdkWSClient.scala
+++ b/core/src/main/scala/org/http4s/jdkhttpclient/JdkWSClient.scala
@@ -131,6 +131,11 @@ object JdkWSClient {
                       })
                     } yield ()
                   }
+              // If the input side is still open (no close received from server), the JDK will not clean up the connection.
+              // This also implies the client can't be shutdown on Java 21+ as it waits for all open connections
+              // to be be closed. As we don't expect/handle anything coming on the input anymore
+              // at this point, we can safely abort.
+              _ <- F.delay(webSocket.abort())
             } yield ()
           }
           .map { case (webSocket, queue, closedDef, sendSem) =>
@@ -164,7 +169,12 @@ object JdkWSClient {
     * [[cats.effect.kernel.Async.executor executor]], sets the
     * [[org.http4s.client.defaults.ConnectTimeout default http4s connect timeout]], and disables
     * [[https://github.com/http4s/http4s-jdk-http-client/issues/200 TLS 1.3 on JDK 11]].
+    *
+    * * On Java 21 and higher, it actively closes the underlying client, releasing its resources
+    * early. On earlier Java versions, closing the underlying client is not possible, so the release
+    * is a no-op. On these Java versions (and there only), you can safely use
+    * [[cats.effect.Resource allocated]] to avoid dealing with resource management.
     */
-  def simple[F[_]](implicit F: Async[F]): F[WSClient[F]] =
-    JdkHttpClient.defaultHttpClient[F].map(apply(_))
+  def simple[F[_]](implicit F: Async[F]): Resource[F, WSClient[F]] =
+    JdkHttpClient.defaultHttpClientResource[F].map(apply(_))
 }
diff --git a/core/src/test/scala/org/http4s/jdkhttpclient/BodyLeakExample.scala b/core/src/test/scala/org/http4s/jdkhttpclient/BodyLeakExample.scala
index 7da879a8..b7bdad87 100644
--- a/core/src/test/scala/org/http4s/jdkhttpclient/BodyLeakExample.scala
+++ b/core/src/test/scala/org/http4s/jdkhttpclient/BodyLeakExample.scala
@@ -51,7 +51,7 @@ object BodyLeakExample extends IOApp {
       .withPort(port"8080")
       .withHttpApp(app)
       .build
-      .product(Resource.eval(JdkHttpClient.simple[IO]))
+      .product(JdkHttpClient.simple[IO])
       .use { case (_, client) =>
         for {
           counter <- Ref.of[IO, Long](0L)
diff --git a/core/src/test/scala/org/http4s/jdkhttpclient/DeadlockWorkaround.scala b/core/src/test/scala/org/http4s/jdkhttpclient/DeadlockWorkaround.scala
index e6d903bb..d471ca44 100644
--- a/core/src/test/scala/org/http4s/jdkhttpclient/DeadlockWorkaround.scala
+++ b/core/src/test/scala/org/http4s/jdkhttpclient/DeadlockWorkaround.scala
@@ -29,7 +29,7 @@ class DeadlockWorkaround extends CatsEffectSuite {
   test("fail to connect via TLSv1.3 on Java 11") {
     if (Runtime.version().feature() > 11) IO.pure(true)
     else
-      (JdkHttpClient.simple[IO], JdkWSClient.simple[IO]).flatMapN { (http, ws) =>
+      (JdkHttpClient.simple[IO], JdkWSClient.simple[IO]).tupled.use { case (http, ws) =>
         def testSSLFailure(r: IO[Unit]) = r.intercept[SSLHandshakeException]
         testSSLFailure(http.expect[Unit](uri"https://tls13.1d.pw")) *>
           testSSLFailure(ws.connectHighLevel(WSRequest(uri"wss://tls13.1d.pw")).use(_ => IO.unit))
diff --git a/core/src/test/scala/org/http4s/jdkhttpclient/JdkHttpClientSpec.scala b/core/src/test/scala/org/http4s/jdkhttpclient/JdkHttpClientSpec.scala
index c7308651..77ef29b7 100644
--- a/core/src/test/scala/org/http4s/jdkhttpclient/JdkHttpClientSpec.scala
+++ b/core/src/test/scala/org/http4s/jdkhttpclient/JdkHttpClientSpec.scala
@@ -28,7 +28,7 @@ import org.typelevel.ci._
 import scala.concurrent.duration._
 
 class JdkHttpClientSpec extends ClientRouteTestBattery("JdkHttpClient") {
-  def clientResource: Resource[IO, Client[IO]] = Resource.eval(JdkHttpClient.simple[IO])
+  def clientResource: Resource[IO, Client[IO]] = JdkHttpClient.simple[IO]
 
   // regression test for https://github.com/http4s/http4s-jdk-http-client/issues/395
   test("Don't error with empty body and explicit Content-Length: 0") {
diff --git a/core/src/test/scala/org/http4s/jdkhttpclient/JdkWSClientSpec.scala b/core/src/test/scala/org/http4s/jdkhttpclient/JdkWSClientSpec.scala
index d81e63de..d6604b7f 100644
--- a/core/src/test/scala/org/http4s/jdkhttpclient/JdkWSClientSpec.scala
+++ b/core/src/test/scala/org/http4s/jdkhttpclient/JdkWSClientSpec.scala
@@ -36,7 +36,7 @@ import scala.concurrent.duration._
 class JdkWSClientSpec extends CatsEffectSuite {
 
   val webSocket: IOFixture[WSClient[IO]] =
-    ResourceSuiteLocalFixture("webSocket", Resource.eval(JdkWSClient.simple[IO]))
+    ResourceSuiteLocalFixture("webSocket", JdkWSClient.simple[IO])
   val echoServerUri: IOFixture[Uri] =
     ResourceSuiteLocalFixture(
       "echoServerUri",
diff --git a/docs/README.md b/docs/README.md
index 987fccb7..ee80e913 100644
--- a/docs/README.md
+++ b/docs/README.md
@@ -50,7 +50,7 @@ import org.http4s.jdkhttpclient.JdkHttpClient
 // It comes for free with `cats.effect.IOApp`:
 import cats.effect.unsafe.implicits.global
 
-val client: IO[Client[IO]] = JdkHttpClient.simple[IO]
+val client: Resource[IO, Client[IO]] = JdkHttpClient.simple[IO]
 ```
 
 #### Custom clients
@@ -91,7 +91,7 @@ def fetchStatus[F[_]](c: Client[F], uri: Uri): F[Status] =
   c.status(Request[F](Method.GET, uri = uri))
 
 client
-  .flatMap(c => fetchStatus(c, uri"https://http4s.org/"))
+  .use(c => fetchStatus(c, uri"https://http4s.org/"))
   .unsafeRunSync()
 ```
 
@@ -103,7 +103,7 @@ create a new `HttpClient` instance on every invocation:
 
 ```scala mdoc
 def fetchStatusInefficiently[F[_]: Async](uri: Uri): F[Status] =
-  JdkHttpClient.simple[F].flatMap(_.status(Request[F](Method.GET, uri = uri)))
+  JdkHttpClient.simple[F].use(_.status(Request[F](Method.GET, uri = uri)))
 ```
 
 @:@