diff --git a/docs/docs/multimodal-table/blob.mdx b/docs/docs/multimodal-table/blob.mdx
index ed92e7a832b1..a06a01e70429 100644
--- a/docs/docs/multimodal-table/blob.mdx
+++ b/docs/docs/multimodal-table/blob.mdx
@@ -174,6 +174,29 @@ CREATE TABLE image_table (
 );
 ```
 
+### HTTP descriptor connection keep-alive
+
+For descriptor URLs using HTTP or HTTPS, `blob-descriptor.http.keep-alive-timeout` controls the
+maximum idle age at which a pooled connection may still be reused after a response. Connections
+that have remained idle beyond this cap are removed before the next configured request leases a
+connection. The value must be greater than zero. If a server returns a shorter
+`Keep-Alive: timeout` value, Paimon uses the server value; otherwise this option is also the
+fallback keep-alive duration. Leaving the option unset preserves the HTTP client's existing
+behavior.
+
+This option is not a response timeout and does not limit how long an active response body may take
+to download. It can be set on the target table or supplied as a Flink SQL dynamic option:
+
+```sql
+ALTER TABLE image_table SET (
+    'blob-descriptor.http.keep-alive-timeout' = '60s'
+);
+
+INSERT INTO image_table
+/*+ OPTIONS('blob-descriptor.http.keep-alive-timeout' = '55s') */
+SELECT * FROM source_images;
+```
+
 ## Creating a Table
 
 The recommended way to create a blob table in SQL is to use the **comment directive** `__BLOB_FIELD`, `__BLOB_DESCRIPTOR_FIELD`, or `__BLOB_VIEW_FIELD` on the column. Paimon automatically converts the column to the corresponding BLOB type and registers it in the corresponding option.
diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html
index 3fb25ebce15f..5923efd84a12 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -74,6 +74,12 @@
             <td>String</td>
             <td>Comma-separated field names to treat as BLOB fields and store as serialized BlobDescriptor bytes inline in data files.</td>
         </tr>
+        <tr>
+            <td><h5>blob-descriptor.http.keep-alive-timeout</h5></td>
+            <td style="word-wrap: break-word;">(none)</td>
+            <td>Duration</td>
+            <td>The maximum idle time for a persistent HTTP connection used to fetch descriptor-backed BLOB content. When a server supplies a Keep-Alive timeout, the shorter timeout is used. This is not an HTTP response timeout. The value must be greater than 0. When unset, the HTTP client's existing keep-alive behavior is preserved.</td>
+        </tr>
         <tr>
             <td><h5>blob-descriptor.source-table</h5></td>
             <td style="word-wrap: break-word;">(none)</td>
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index a19d1818c247..29f0c2e8b2d0 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -2714,6 +2714,18 @@ public String toString() {
                                     + "loader, including external tables in REST catalogs. When set, "
                                     + "other blob-descriptor.* FileIO options are ignored.");
 
+    public static final ConfigOption<Duration> BLOB_DESCRIPTOR_HTTP_KEEP_ALIVE_TIMEOUT =
+            key(BLOB_DESCRIPTOR_PREFIX + "http.keep-alive-timeout")
+                    .durationType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "The maximum idle time for a persistent HTTP connection used to fetch "
+                                    + "descriptor-backed BLOB content. When a server supplies a "
+                                    + "Keep-Alive timeout, the shorter timeout is used. This is not "
+                                    + "an HTTP response timeout. The value must be greater than 0. "
+                                    + "When unset, the HTTP client's existing keep-alive behavior is "
+                                    + "preserved.");
+
     public static final ConfigOption<Boolean> BLOB_WRITE_NULL_ON_MISSING_FILE =
             key("blob-write-null-on-missing-file")
                     .booleanType()
@@ -4461,6 +4473,17 @@ public boolean blobAsDescriptor() {
         return options.get(BLOB_AS_DESCRIPTOR);
     }
 
+    public Optional<Duration> blobDescriptorHttpKeepAliveTimeout() {
+        Optional<Duration> timeout = options.getOptional(BLOB_DESCRIPTOR_HTTP_KEEP_ALIVE_TIMEOUT);
+        timeout.ifPresent(
+                value ->
+                        checkArgument(
+                                !value.isZero() && !value.isNegative(),
+                                "Option '%s' must be greater than 0.",
+                                BLOB_DESCRIPTOR_HTTP_KEEP_ALIVE_TIMEOUT.key()));
+        return timeout;
+    }
+
     public boolean blobWriteNullOnMissingFile() {
         return options.get(BLOB_WRITE_NULL_ON_MISSING_FILE);
     }
diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/HttpClientUtils.java b/paimon-api/src/main/java/org/apache/paimon/rest/HttpClientUtils.java
index 1444c8118d14..e5717999ce87 100644
--- a/paimon-api/src/main/java/org/apache/paimon/rest/HttpClientUtils.java
+++ b/paimon-api/src/main/java/org/apache/paimon/rest/HttpClientUtils.java
@@ -22,18 +22,21 @@
 import org.apache.paimon.rest.interceptor.TimingInterceptor;
 import org.apache.paimon.utils.SensitiveConfigUtils;
 
+import org.apache.hc.client5.http.ConnectionKeepAliveStrategy;
 import org.apache.hc.client5.http.classic.methods.HttpDelete;
 import org.apache.hc.client5.http.classic.methods.HttpGet;
 import org.apache.hc.client5.http.classic.methods.HttpHead;
 import org.apache.hc.client5.http.classic.methods.HttpPost;
 import org.apache.hc.client5.http.config.RequestConfig;
 import org.apache.hc.client5.http.entity.DecompressingEntity;
+import org.apache.hc.client5.http.impl.DefaultConnectionKeepAliveStrategy;
 import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
 import org.apache.hc.client5.http.impl.classic.CloseableHttpResponse;
 import org.apache.hc.client5.http.impl.classic.HttpClientBuilder;
 import org.apache.hc.client5.http.impl.classic.HttpClients;
+import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManager;
 import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManagerBuilder;
-import org.apache.hc.client5.http.io.HttpClientConnectionManager;
+import org.apache.hc.client5.http.protocol.HttpClientContext;
 import org.apache.hc.client5.http.ssl.DefaultClientTlsStrategy;
 import org.apache.hc.client5.http.ssl.HttpsSupport;
 import org.apache.hc.core5.http.ClassicHttpRequest;
@@ -41,20 +44,28 @@
 import org.apache.hc.core5.http.Header;
 import org.apache.hc.core5.http.HttpEntity;
 import org.apache.hc.core5.http.HttpHeaders;
+import org.apache.hc.core5.http.HttpResponse;
 import org.apache.hc.core5.http.HttpStatus;
 import org.apache.hc.core5.http.TruncatedChunkException;
+import org.apache.hc.core5.http.protocol.HttpContext;
 import org.apache.hc.core5.reactor.ssl.SSLBufferMode;
 import org.apache.hc.core5.ssl.SSLContexts;
+import org.apache.hc.core5.util.TimeValue;
 import org.apache.hc.core5.util.Timeout;
 
+import javax.annotation.Nullable;
+
 import java.io.IOException;
 import java.io.InputStream;
 import java.security.MessageDigest;
 import java.security.NoSuchAlgorithmException;
+import java.time.Duration;
 import java.util.function.Function;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
 /** Utils for {@link HttpClientBuilder}. */
 public class HttpClientUtils {
 
@@ -66,11 +77,26 @@ public class HttpClientUtils {
                     .setConnectionRequestTimeout(Timeout.ofMinutes(3))
                     .setResponseTimeout(Timeout.ofMinutes(3))
                     .build();
+    static final String KEEP_ALIVE_TIMEOUT_ATTRIBUTE = "org.apache.paimon.http.keep-alive-timeout";
+    static final ConnectionKeepAliveStrategy KEEP_ALIVE_STRATEGY =
+            HttpClientUtils::getKeepAliveDuration;
+    private static final PoolingHttpClientConnectionManager BLOB_HTTP_CONNECTION_MANAGER =
+            configureConnectionManager();
+    private static final CloseableHttpClient BLOB_HTTP_CLIENT =
+            createLoggingBuilder(BLOB_HTTP_CONNECTION_MANAGER).build();
 
     public static final CloseableHttpClient DEFAULT_HTTP_CLIENT = createLoggingBuilder().build();
 
     public static HttpClientBuilder createLoggingBuilder() {
-        HttpClientBuilder clientBuilder = createBuilder();
+        return addLoggingInterceptors(createBuilder());
+    }
+
+    private static HttpClientBuilder createLoggingBuilder(
+            PoolingHttpClientConnectionManager connectionManager) {
+        return addLoggingInterceptors(createBuilder(connectionManager));
+    }
+
+    private static HttpClientBuilder addLoggingInterceptors(HttpClientBuilder clientBuilder) {
         clientBuilder
                 .addRequestInterceptorFirst(new TimingInterceptor())
                 .addResponseInterceptorLast(new LoggingInterceptor());
@@ -78,15 +104,21 @@ public static HttpClientBuilder createLoggingBuilder() {
     }
 
     public static HttpClientBuilder createBuilder() {
+        return createBuilder(configureConnectionManager());
+    }
+
+    private static HttpClientBuilder createBuilder(
+            PoolingHttpClientConnectionManager connectionManager) {
         HttpClientBuilder clientBuilder = HttpClients.custom();
         clientBuilder.setDefaultRequestConfig(DEFAULT_REQUEST_CONFIG);
 
-        clientBuilder.setConnectionManager(configureConnectionManager());
+        clientBuilder.setConnectionManager(connectionManager);
+        clientBuilder.setKeepAliveStrategy(KEEP_ALIVE_STRATEGY);
         clientBuilder.setRetryStrategy(new ExponentialHttpRequestRetryStrategy(5));
         return clientBuilder;
     }
 
-    private static HttpClientConnectionManager configureConnectionManager() {
+    private static PoolingHttpClientConnectionManager configureConnectionManager() {
         PoolingHttpClientConnectionManagerBuilder connectionManagerBuilder =
                 PoolingHttpClientConnectionManagerBuilder.create();
         connectionManagerBuilder.useSystemProperties().setMaxConnTotal(100).setMaxConnPerRoute(100);
@@ -105,7 +137,13 @@ private static HttpClientConnectionManager configureConnectionManager() {
     }
 
     public static InputStream getAsInputStream(String uri) throws IOException {
-        return new ResumableHttpInputStream(uri);
+        return new ResumableHttpInputStream(uri, null);
+    }
+
+    public static InputStream getAsInputStream(String uri, Duration keepAliveTimeout)
+            throws IOException {
+        validateKeepAliveTimeout(keepAliveTimeout);
+        return new ResumableHttpInputStream(uri, keepAliveTimeout);
     }
 
     /**
@@ -115,11 +153,21 @@ public static InputStream getAsInputStream(String uri) throws IOException {
      * different status than GET.
      */
     public static boolean exists(String uri) throws IOException {
-        int headStatusCode = headStatusCode(uri);
+        return existsInternal(uri, null);
+    }
+
+    public static boolean exists(String uri, Duration keepAliveTimeout) throws IOException {
+        validateKeepAliveTimeout(keepAliveTimeout);
+        return existsInternal(uri, keepAliveTimeout);
+    }
+
+    private static boolean existsInternal(String uri, @Nullable Duration keepAliveTimeout)
+            throws IOException {
+        int headStatusCode = headStatusCode(uri, keepAliveTimeout);
         if (headStatusCode == HttpStatus.SC_OK) {
             return true;
         }
-        int rangeStatusCode = getRangeStatusCode(uri);
+        int rangeStatusCode = getRangeStatusCode(uri, keepAliveTimeout);
         if (rangeStatusCode == HttpStatus.SC_OK
                 || rangeStatusCode == HttpStatus.SC_PARTIAL_CONTENT
                 || rangeStatusCode == HttpStatus.SC_REQUESTED_RANGE_NOT_SATISFIABLE) {
@@ -195,17 +243,19 @@ private static Integer parseStatusCodeSuffix(String statusText) {
         }
     }
 
-    private static int headStatusCode(String uri) throws IOException {
+    private static int headStatusCode(String uri, @Nullable Duration keepAliveTimeout)
+            throws IOException {
         HttpHead httpHead = newHttpHead(uri);
-        try (CloseableHttpResponse response = execute(httpHead, uri)) {
+        try (CloseableHttpResponse response = execute(httpHead, uri, keepAliveTimeout)) {
             return response.getCode();
         }
     }
 
-    private static int getRangeStatusCode(String uri) throws IOException {
+    private static int getRangeStatusCode(String uri, @Nullable Duration keepAliveTimeout)
+            throws IOException {
         HttpGet httpGet = newHttpGet(uri);
         httpGet.addHeader("Range", "bytes=0-0");
-        try (CloseableHttpResponse response = execute(httpGet, uri)) {
+        try (CloseableHttpResponse response = execute(httpGet, uri, keepAliveTimeout)) {
             return response.getCode();
         }
     }
@@ -216,16 +266,56 @@ private static int getRangeStatusCode(String uri) throws IOException {
      * to &lt;Location&gt;") echo the target URL, which for a signed URL is a credential; only the
      * sanitized request URI is reported.
      */
-    private static CloseableHttpResponse execute(ClassicHttpRequest request, String uri)
+    private static CloseableHttpResponse execute(
+            ClassicHttpRequest request, String uri, @Nullable Duration keepAliveTimeout)
             throws IOException {
         try {
-            return DEFAULT_HTTP_CLIENT.execute(request);
+            if (keepAliveTimeout == null) {
+                return DEFAULT_HTTP_CLIENT.execute(request);
+            }
+
+            TimeValue timeout = TimeValue.of(keepAliveTimeout);
+            // This pool is dedicated to explicitly configured HTTP BLOB reads, so enforcing one
+            // table's cap cannot close idle REST Catalog connections. It is still shared by capped
+            // BLOB reads to avoid creating one client and pool per table.
+            BLOB_HTTP_CONNECTION_MANAGER.closeIdle(timeout);
+            HttpClientContext context = HttpClientContext.create();
+            context.setRequestConfig(
+                    RequestConfig.copy(DEFAULT_REQUEST_CONFIG)
+                            .setConnectionKeepAlive(timeout)
+                            .build());
+            context.setAttribute(KEEP_ALIVE_TIMEOUT_ATTRIBUTE, timeout);
+            return BLOB_HTTP_CLIENT.execute(request, context);
         } catch (IOException | RuntimeException e) {
             throw new IOException(
                     "HTTP request failed for uri: " + SensitiveConfigUtils.sanitizeUri(uri));
         }
     }
 
+    private static TimeValue getKeepAliveDuration(HttpResponse response, HttpContext context) {
+        TimeValue keepAlive =
+                DefaultConnectionKeepAliveStrategy.INSTANCE.getKeepAliveDuration(response, context);
+        Object configured = context.getAttribute(KEEP_ALIVE_TIMEOUT_ATTRIBUTE);
+        if (!(configured instanceof TimeValue)) {
+            return keepAlive;
+        }
+
+        TimeValue cap = (TimeValue) configured;
+        return keepAlive == null
+                        || !TimeValue.isNonNegative(keepAlive)
+                        || cap.compareTo(keepAlive) < 0
+                ? cap
+                : keepAlive;
+    }
+
+    private static void validateKeepAliveTimeout(Duration keepAliveTimeout) {
+        checkArgument(
+                keepAliveTimeout != null
+                        && !keepAliveTimeout.isZero()
+                        && !keepAliveTimeout.isNegative(),
+                "HTTP connection keep-alive timeout must be greater than 0.");
+    }
+
     public static HttpGet newHttpGet(String uri) {
         return newRequest(uri, HttpGet::new);
     }
@@ -269,6 +359,7 @@ private static RuntimeException httpError(int statusCode) {
     private static class ResumableHttpInputStream extends InputStream {
 
         private final String uri;
+        @Nullable private final Duration keepAliveTimeout;
         private final byte[] singleByte = new byte[1];
 
         private CloseableHttpResponse response;
@@ -283,8 +374,10 @@ private static class ResumableHttpInputStream extends InputStream {
         private IOException terminalFailure;
         private final MessageDigest deliveredDigest = sha256();
 
-        private ResumableHttpInputStream(String uri) throws IOException {
+        private ResumableHttpInputStream(String uri, @Nullable Duration keepAliveTimeout)
+                throws IOException {
             this.uri = uri;
+            this.keepAliveTimeout = keepAliveTimeout;
             openInitialResponse();
         }
 
@@ -372,7 +465,7 @@ public void close() throws IOException {
 
         private void openInitialResponse() throws IOException {
             HttpGet request = newBodyGet(uri);
-            CloseableHttpResponse newResponse = execute(request, uri);
+            CloseableHttpResponse newResponse = execute(request, uri, keepAliveTimeout);
             boolean accepted = false;
             try {
                 if (newResponse.getCode() == HttpStatus.SC_NOT_ACCEPTABLE) {
@@ -458,7 +551,7 @@ private void resume(String reason) throws IOException {
             request.addHeader(HttpHeaders.RANGE, "bytes=" + position + "-");
             request.addHeader(HttpHeaders.IF_RANGE, strongEtag);
 
-            CloseableHttpResponse newResponse = execute(request, uri);
+            CloseableHttpResponse newResponse = execute(request, uri, keepAliveTimeout);
             boolean accepted = false;
             try {
                 if (newResponse.getCode() != HttpStatus.SC_PARTIAL_CONTENT) {
@@ -514,7 +607,7 @@ private void replayFromStart() throws IOException {
             byte[] expectedPrefixDigest = digestSnapshot(deliveredDigest);
             while (true) {
                 HttpGet request = newBodyGet(uri);
-                CloseableHttpResponse newResponse = execute(request, uri);
+                CloseableHttpResponse newResponse = execute(request, uri, keepAliveTimeout);
                 boolean accepted = false;
                 try {
                     if (newResponse.getCode() != HttpStatus.SC_OK) {
@@ -594,7 +687,7 @@ private void verifyReplayedPrefix(InputStream newStream, byte[] expectedPrefixDi
          */
         private void openContentDecodedResponse() throws IOException {
             HttpGet request = newHttpGet(uri);
-            CloseableHttpResponse newResponse = execute(request, uri);
+            CloseableHttpResponse newResponse = execute(request, uri, keepAliveTimeout);
             boolean accepted = false;
             try {
                 if (newResponse.getCode() != HttpStatus.SC_OK) {
diff --git a/paimon-api/src/test/java/org/apache/paimon/rest/HttpClientUtilsTest.java b/paimon-api/src/test/java/org/apache/paimon/rest/HttpClientUtilsTest.java
index 5b1c3c27004c..ffb05edf700a 100644
--- a/paimon-api/src/test/java/org/apache/paimon/rest/HttpClientUtilsTest.java
+++ b/paimon-api/src/test/java/org/apache/paimon/rest/HttpClientUtilsTest.java
@@ -23,6 +23,11 @@
 import com.sun.net.httpserver.HttpExchange;
 import com.sun.net.httpserver.HttpHandler;
 import com.sun.net.httpserver.HttpServer;
+import org.apache.hc.client5.http.config.RequestConfig;
+import org.apache.hc.client5.http.protocol.HttpClientContext;
+import org.apache.hc.core5.http.HttpHeaders;
+import org.apache.hc.core5.http.message.BasicClassicHttpResponse;
+import org.apache.hc.core5.util.TimeValue;
 import org.assertj.core.api.ThrowableAssert;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
@@ -33,8 +38,11 @@
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.net.InetSocketAddress;
+import java.time.Duration;
 import java.util.Arrays;
+import java.util.List;
 import java.util.Random;
+import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.zip.GZIPOutputStream;
@@ -73,6 +81,75 @@ public void testExistsReturnsTrueForAvailableResource() throws Exception {
         assertThat(HttpClientUtils.exists(url("/ok"))).isTrue();
     }
 
+    @Test
+    public void testConfiguredKeepAliveTimeoutCapsServerTimeout() {
+        assertThat(keepAliveDuration(Duration.ofSeconds(60), "timeout=75").toSeconds())
+                .isEqualTo(60);
+    }
+
+    @Test
+    public void testConfiguredKeepAliveTimeoutKeepsShorterServerTimeout() {
+        assertThat(keepAliveDuration(Duration.ofSeconds(60), "timeout=30").toSeconds())
+                .isEqualTo(30);
+    }
+
+    @Test
+    public void testConfiguredKeepAliveTimeoutIsFallbackWithoutServerHeader() {
+        assertThat(keepAliveDuration(Duration.ofSeconds(60), null).toSeconds()).isEqualTo(60);
+    }
+
+    @Test
+    public void testConfiguredKeepAliveTimeoutCapsIndefiniteServerTimeout() {
+        assertThat(keepAliveDuration(Duration.ofSeconds(60), "timeout=-1").toSeconds())
+                .isEqualTo(60);
+    }
+
+    @Test
+    public void testUnconfiguredKeepAlivePreservesHttpClientDefault() {
+        BasicClassicHttpResponse response = new BasicClassicHttpResponse(200);
+        HttpClientContext context = HttpClientContext.create();
+        context.setRequestConfig(RequestConfig.custom().build());
+
+        assertThat(
+                        HttpClientUtils.KEEP_ALIVE_STRATEGY
+                                .getKeepAliveDuration(response, context)
+                                .toMinutes())
+                .isEqualTo(3);
+    }
+
+    @Test
+    public void testRejectNonPositiveKeepAliveTimeout() {
+        assertThatThrownBy(() -> HttpClientUtils.getAsInputStream(url("/ok"), Duration.ZERO))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining("greater than 0");
+        assertThatThrownBy(() -> HttpClientUtils.exists(url("/ok"), Duration.ofSeconds(-1)))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining("greater than 0");
+    }
+
+    @Test
+    public void testConfiguredKeepAliveUsesDedicatedPoolAndClosesOlderIdleConnection()
+            throws Exception {
+        List<Integer> remotePorts = new CopyOnWriteArrayList<>();
+        registerHandler(
+                "/pool-state",
+                exchange -> {
+                    remotePorts.add(exchange.getRemoteAddress().getPort());
+                    respond(exchange, 200, "ok".getBytes());
+                });
+
+        readAllAndClose(url("/pool-state"), null);
+        readAllAndClose(url("/pool-state"), Duration.ofSeconds(10));
+        readAllAndClose(url("/pool-state"), Duration.ofSeconds(10));
+        Thread.sleep(150);
+        readAllAndClose(url("/pool-state"), Duration.ofMillis(50));
+
+        assertThat(remotePorts).hasSize(4);
+        assertThat(remotePorts.get(1)).isNotEqualTo(remotePorts.get(0));
+        assertThat(remotePorts.get(2)).isEqualTo(remotePorts.get(1));
+        assertThat(remotePorts.get(3)).isNotEqualTo(remotePorts.get(2));
+    }
+
     @Test
     public void testExistsReturnsFalseForMissingResource() throws Exception {
         registerHandler(
@@ -96,7 +173,7 @@ public void testExistsFallsBackToRangeGetWhenHeadNotAllowed() throws Exception {
                     respond(exchange, 200, "abc".getBytes());
                 });
 
-        assertThat(HttpClientUtils.exists(url("/no-head"))).isTrue();
+        assertThat(HttpClientUtils.exists(url("/no-head"), Duration.ofSeconds(60))).isTrue();
     }
 
     @Test
@@ -302,7 +379,8 @@ public void testGetAsInputStreamFallsBackWhenIdentityEncodingIsRejected() throws
                     respond(exchange, 200, compressed);
                 });
 
-        try (InputStream in = HttpClientUtils.getAsInputStream(url("/encoded-only"))) {
+        try (InputStream in =
+                HttpClientUtils.getAsInputStream(url("/encoded-only"), Duration.ofSeconds(60))) {
             assertThat(readAll(in)).isEqualTo(payload);
         }
         assertThat(requestCount).hasValue(2);
@@ -362,7 +440,8 @@ public void testGetAsInputStreamResumesTruncatedResponseBody() throws Exception
                     respond(exchange, 206, remaining);
                 });
 
-        try (InputStream in = HttpClientUtils.getAsInputStream(url("/truncated"))) {
+        try (InputStream in =
+                HttpClientUtils.getAsInputStream(url("/truncated"), Duration.ofSeconds(60))) {
             assertThat(readAll(in)).isEqualTo(payload);
         }
         assertThat(requestCount).hasValue(2);
@@ -391,7 +470,9 @@ public void testGetAsInputStreamReplaysAndVerifiesWithoutResourceValidator() thr
                     respond(exchange, 200, payload);
                 });
 
-        try (InputStream in = HttpClientUtils.getAsInputStream(url("/truncated-no-validator"))) {
+        try (InputStream in =
+                HttpClientUtils.getAsInputStream(
+                        url("/truncated-no-validator"), Duration.ofSeconds(60))) {
             assertThat(readAll(in)).isEqualTo(payload);
         }
         assertThat(requestCount).hasValue(2);
@@ -781,6 +862,27 @@ private void registerHandler(String path, HttpHandler handler) {
         server.createContext(path, handler);
     }
 
+    private static TimeValue keepAliveDuration(Duration cap, String serverKeepAlive) {
+        TimeValue timeout = TimeValue.of(cap);
+        BasicClassicHttpResponse response = new BasicClassicHttpResponse(200);
+        if (serverKeepAlive != null) {
+            response.addHeader(HttpHeaders.KEEP_ALIVE, serverKeepAlive);
+        }
+        HttpClientContext context = HttpClientContext.create();
+        context.setRequestConfig(RequestConfig.custom().setConnectionKeepAlive(timeout).build());
+        context.setAttribute(HttpClientUtils.KEEP_ALIVE_TIMEOUT_ATTRIBUTE, timeout);
+        return HttpClientUtils.KEEP_ALIVE_STRATEGY.getKeepAliveDuration(response, context);
+    }
+
+    private static void readAllAndClose(String uri, Duration keepAliveTimeout) throws IOException {
+        try (InputStream inputStream =
+                keepAliveTimeout == null
+                        ? HttpClientUtils.getAsInputStream(uri)
+                        : HttpClientUtils.getAsInputStream(uri, keepAliveTimeout)) {
+            readAll(inputStream);
+        }
+    }
+
     private String url(String path) {
         return "http://127.0.0.1:" + port + path;
     }
diff --git a/paimon-common/src/main/java/org/apache/paimon/utils/BlobDescriptorUtils.java b/paimon-common/src/main/java/org/apache/paimon/utils/BlobDescriptorUtils.java
index 1cc1366bc889..22a34c5e85d1 100644
--- a/paimon-common/src/main/java/org/apache/paimon/utils/BlobDescriptorUtils.java
+++ b/paimon-common/src/main/java/org/apache/paimon/utils/BlobDescriptorUtils.java
@@ -31,6 +31,7 @@
 import java.util.Map;
 import java.util.Objects;
 
+import static org.apache.paimon.CoreOptions.BLOB_DESCRIPTOR_HTTP_KEEP_ALIVE_TIMEOUT;
 import static org.apache.paimon.CoreOptions.BLOB_DESCRIPTOR_PREFIX;
 
 /** Utils for {@link BlobDescriptor}. */
@@ -73,7 +74,9 @@ public static CatalogContext getCatalogContext(
         Map<String, String> descriptorSpecified = new HashMap<>();
         for (Map.Entry<String, String> entry : tableOptions.toMap().entrySet()) {
             String key = entry.getKey();
-            if (key != null && key.startsWith(BLOB_DESCRIPTOR_PREFIX)) {
+            if (key != null
+                    && key.startsWith(BLOB_DESCRIPTOR_PREFIX)
+                    && !key.equals(BLOB_DESCRIPTOR_HTTP_KEEP_ALIVE_TIMEOUT.key())) {
                 descriptorSpecified.put(
                         key.substring(BLOB_DESCRIPTOR_PREFIX.length()), entry.getValue());
             }
diff --git a/paimon-common/src/main/java/org/apache/paimon/utils/UriReader.java b/paimon-common/src/main/java/org/apache/paimon/utils/UriReader.java
index 236bd6b6f697..dd0798ecd15e 100644
--- a/paimon-common/src/main/java/org/apache/paimon/utils/UriReader.java
+++ b/paimon-common/src/main/java/org/apache/paimon/utils/UriReader.java
@@ -23,7 +23,10 @@
 import org.apache.paimon.fs.SeekableInputStream;
 import org.apache.paimon.rest.HttpClientUtils;
 
+import javax.annotation.Nullable;
+
 import java.io.IOException;
+import java.time.Duration;
 
 /** An interface to read uri as a stream. */
 public interface UriReader {
@@ -38,6 +41,10 @@ static UriReader fromHttp() {
         return new HttpUriReader();
     }
 
+    static UriReader fromHttp(Duration keepAliveTimeout) {
+        return new HttpUriReader(keepAliveTimeout);
+    }
+
     /** A {@link UriReader} uses {@link FileIO} to read file. */
     class FileUriReader implements UriReader {
 
@@ -60,13 +67,33 @@ public boolean exists(String uri) throws IOException {
     /** A {@link UriReader} reads http uri. */
     class HttpUriReader implements UriReader {
 
+        @Nullable private final Duration keepAliveTimeout;
+
+        public HttpUriReader() {
+            this.keepAliveTimeout = null;
+        }
+
+        public HttpUriReader(Duration keepAliveTimeout) {
+            this.keepAliveTimeout = keepAliveTimeout;
+        }
+
         @Override
         public SeekableInputStream newInputStream(String uri) throws IOException {
-            return SeekableInputStream.wrap(HttpClientUtils.getAsInputStream(uri));
+            return SeekableInputStream.wrap(
+                    keepAliveTimeout == null
+                            ? HttpClientUtils.getAsInputStream(uri)
+                            : HttpClientUtils.getAsInputStream(uri, keepAliveTimeout));
         }
 
         public boolean exists(String uri) throws IOException {
-            return HttpClientUtils.exists(uri);
+            return keepAliveTimeout == null
+                    ? HttpClientUtils.exists(uri)
+                    : HttpClientUtils.exists(uri, keepAliveTimeout);
+        }
+
+        @Nullable
+        Duration keepAliveTimeout() {
+            return keepAliveTimeout;
         }
     }
 }
diff --git a/paimon-common/src/main/java/org/apache/paimon/utils/UriReaderFactory.java b/paimon-common/src/main/java/org/apache/paimon/utils/UriReaderFactory.java
index 4d1f84cfcbb3..97129f749792 100644
--- a/paimon-common/src/main/java/org/apache/paimon/utils/UriReaderFactory.java
+++ b/paimon-common/src/main/java/org/apache/paimon/utils/UriReaderFactory.java
@@ -18,9 +18,12 @@
 
 package org.apache.paimon.utils;
 
+import org.apache.paimon.CoreOptions;
 import org.apache.paimon.catalog.CatalogContext;
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.Path;
+import org.apache.paimon.options.ConfigOption;
+import org.apache.paimon.options.ConfigOptions;
 
 import javax.annotation.Nullable;
 
@@ -28,26 +31,51 @@
 import java.io.ObjectInputStream;
 import java.io.Serializable;
 import java.net.URI;
+import java.time.Duration;
 import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
 
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
 /** A factory to create and cache {@link UriReader}. */
 public class UriReaderFactory implements Serializable {
 
     private static final long serialVersionUID = -8477284718943635074L;
+    private static final ConfigOption<Duration> HTTP_KEEP_ALIVE_TIMEOUT =
+            ConfigOptions.key(
+                            CoreOptions.BLOB_DESCRIPTOR_HTTP_KEEP_ALIVE_TIMEOUT
+                                    .key()
+                                    .substring(CoreOptions.BLOB_DESCRIPTOR_PREFIX.length()))
+                    .durationType()
+                    .noDefaultValue();
 
     @Nullable private final CatalogContext context;
+    @Nullable private final Duration httpKeepAliveTimeout;
     private transient Map<UriKey, UriReader> readers;
 
     public UriReaderFactory(@Nullable CatalogContext context) {
+        this(context, keepAliveTimeout(context));
+    }
+
+    public UriReaderFactory(
+            @Nullable CatalogContext context, @Nullable Duration httpKeepAliveTimeout) {
         this.context = context;
+        validateKeepAliveTimeout(httpKeepAliveTimeout);
+        this.httpKeepAliveTimeout = httpKeepAliveTimeout;
         this.readers = new ConcurrentHashMap<>();
     }
 
     /** Creates a factory which uses the provided {@link FileIO} for non-HTTP URIs. */
     public static UriReaderFactory fromFileIO(FileIO fileIO) {
-        return new ProvidedFileIOUriReaderFactory(fileIO);
+        return new ProvidedFileIOUriReaderFactory(fileIO, null);
+    }
+
+    /**
+     * Creates a factory with a keep-alive timeout for HTTP URIs and the provided {@link FileIO}.
+     */
+    public static UriReaderFactory fromFileIO(FileIO fileIO, Duration httpKeepAliveTimeout) {
+        return new ProvidedFileIOUriReaderFactory(fileIO, httpKeepAliveTimeout);
     }
 
     public UriReader create(String input) {
@@ -91,7 +119,9 @@ private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundE
 
     protected UriReader newReader(URI uri) {
         if (isHttp(uri)) {
-            return UriReader.fromHttp();
+            return httpKeepAliveTimeout == null
+                    ? UriReader.fromHttp()
+                    : UriReader.fromHttp(httpKeepAliveTimeout);
         }
 
         try {
@@ -107,6 +137,22 @@ private static boolean isHttp(URI uri) {
                 || "https".equalsIgnoreCase(uri.getScheme());
     }
 
+    @Nullable
+    private static Duration keepAliveTimeout(@Nullable CatalogContext context) {
+        return context == null
+                ? null
+                : context.options().getOptional(HTTP_KEEP_ALIVE_TIMEOUT).orElse(null);
+    }
+
+    private static void validateKeepAliveTimeout(@Nullable Duration keepAliveTimeout) {
+        if (keepAliveTimeout != null) {
+            checkArgument(
+                    !keepAliveTimeout.isZero() && !keepAliveTimeout.isNegative(),
+                    "Option '%s' must be greater than 0.",
+                    HTTP_KEEP_ALIVE_TIMEOUT.key());
+        }
+    }
+
     private static final class ProvidedFileIOUriReaderFactory extends UriReaderFactory {
 
         private static final long serialVersionUID = 1L;
@@ -116,8 +162,9 @@ private static final class ProvidedFileIOUriReaderFactory extends UriReaderFacto
         // rebuild the transient reader cache with table-scoped credentials.
         private final FileIO fileIO;
 
-        private ProvidedFileIOUriReaderFactory(FileIO fileIO) {
-            super(null);
+        private ProvidedFileIOUriReaderFactory(
+                FileIO fileIO, @Nullable Duration httpKeepAliveTimeout) {
+            super(null, httpKeepAliveTimeout);
             this.fileIO = Objects.requireNonNull(fileIO);
         }
 
diff --git a/paimon-common/src/test/java/org/apache/paimon/utils/BlobDescriptorUtilsTest.java b/paimon-common/src/test/java/org/apache/paimon/utils/BlobDescriptorUtilsTest.java
index d970a289eb5a..da360cb9fd4f 100644
--- a/paimon-common/src/test/java/org/apache/paimon/utils/BlobDescriptorUtilsTest.java
+++ b/paimon-common/src/test/java/org/apache/paimon/utils/BlobDescriptorUtilsTest.java
@@ -18,19 +18,50 @@
 
 package org.apache.paimon.utils;
 
+import org.apache.paimon.catalog.CatalogContext;
 import org.apache.paimon.data.BlobDescriptor;
 import org.apache.paimon.fs.Path;
+import org.apache.paimon.options.Options;
 
 import org.junit.jupiter.api.Test;
 
 import java.io.IOException;
 
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatCode;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /** Tests for {@link BlobDescriptorUtils}. */
 class BlobDescriptorUtilsTest {
 
+    @Test
+    void testHttpKeepAliveTimeoutPreservesCurrentContext() {
+        Options currentOptions = new Options();
+        currentOptions.setString("warehouse", "oss://bucket/warehouse");
+        CatalogContext currentContext = CatalogContext.create(currentOptions);
+        Options tableOptions = new Options();
+        tableOptions.setString("blob-descriptor.http.keep-alive-timeout", "60s");
+
+        CatalogContext context =
+                BlobDescriptorUtils.getCatalogContext(currentContext, tableOptions);
+
+        assertThat(context).isSameAs(currentContext);
+    }
+
+    @Test
+    void testHttpKeepAliveTimeoutIsNotPassedToExternalFileIO() {
+        Options tableOptions = new Options();
+        tableOptions.setString("blob-descriptor.http.keep-alive-timeout", "60s");
+        tableOptions.setString("blob-descriptor.fs.oss.endpoint", "oss-cn-test.aliyuncs.com");
+
+        CatalogContext context = BlobDescriptorUtils.getCatalogContext(null, tableOptions);
+
+        assertThat(context.options().toMap())
+                .containsEntry("fs.oss.endpoint", "oss-cn-test.aliyuncs.com")
+                .doesNotContainKey("http.keep-alive-timeout")
+                .doesNotContainKey("blob-descriptor.http.keep-alive-timeout");
+    }
+
     @Test
     void testValidateTableRoot() {
         Path tableRoot = new Path("oss://bucket/table");
diff --git a/paimon-common/src/test/java/org/apache/paimon/utils/UriReaderFactoryTest.java b/paimon-common/src/test/java/org/apache/paimon/utils/UriReaderFactoryTest.java
index 16ea5676fb42..552696b87b67 100644
--- a/paimon-common/src/test/java/org/apache/paimon/utils/UriReaderFactoryTest.java
+++ b/paimon-common/src/test/java/org/apache/paimon/utils/UriReaderFactoryTest.java
@@ -37,6 +37,7 @@
 import java.io.OutputStream;
 import java.net.InetSocketAddress;
 import java.nio.file.Files;
+import java.time.Duration;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -70,6 +71,33 @@ public void tearDownHttpServer() {
     public void testCreateHttpUriReader() {
         UriReader reader = factory.create("http://example.com/file.txt");
         assertThat(reader).isInstanceOf(HttpUriReader.class);
+        assertThat(((HttpUriReader) reader).keepAliveTimeout()).isNull();
+    }
+
+    @Test
+    public void testCreateHttpUriReaderWithKeepAliveTimeout() throws Exception {
+        Options options = new Options();
+        options.setString("http.keep-alive-timeout", "60s");
+        UriReaderFactory configured = new UriReaderFactory(CatalogContext.create(options));
+
+        HttpUriReader original = (HttpUriReader) configured.create("http://example.com/file.txt");
+        assertThat(original.keepAliveTimeout()).isEqualTo(Duration.ofSeconds(60));
+
+        UriReaderFactory deserialized = InstantiationUtil.clone(configured);
+        HttpUriReader restored = (HttpUriReader) deserialized.create("http://example.com/file.txt");
+        assertThat(restored.keepAliveTimeout()).isEqualTo(Duration.ofSeconds(60));
+        assertThat(restored).isNotSameAs(original);
+    }
+
+    @Test
+    public void testRejectNonPositiveKeepAliveTimeout() {
+        Options options = new Options();
+        options.setString("http.keep-alive-timeout", "0s");
+
+        assertThatThrownBy(() -> new UriReaderFactory(CatalogContext.create(options)))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining("http.keep-alive-timeout")
+                .hasMessageContaining("greater than 0");
     }
 
     @Test
diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
index 5d55a73cba89..b590251f8685 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
@@ -233,6 +233,7 @@ public static void validateTableSchema(TableSchema schema, Set<String> dynamicOp
                 FileFormat.fromIdentifier(options.formatType(), new Options(schema.options()));
         RowType tableRowType = new RowType(schema.fields());
         validateGeospatialTypes(schema, options, tableRowType);
+        options.blobDescriptorHttpKeepAliveTimeout();
         validateBlobFields(tableRowType, options);
         Set<String> blobDescriptorFields = validateBlobDescriptorFields(tableRowType, options);
         Set<String> blobViewFields =
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/BlobDescriptorReaderFactory.java b/paimon-core/src/main/java/org/apache/paimon/table/BlobDescriptorReaderFactory.java
index bcbef2980451..899eb6604133 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/BlobDescriptorReaderFactory.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/BlobDescriptorReaderFactory.java
@@ -18,6 +18,7 @@
 
 package org.apache.paimon.table;
 
+import org.apache.paimon.CoreOptions;
 import org.apache.paimon.catalog.Catalog;
 import org.apache.paimon.catalog.CatalogContext;
 import org.apache.paimon.catalog.CatalogFactory;
@@ -28,6 +29,10 @@
 import org.apache.paimon.utils.BlobDescriptorUtils;
 import org.apache.paimon.utils.UriReaderFactory;
 
+import javax.annotation.Nullable;
+
+import java.time.Duration;
+
 import static org.apache.paimon.CoreOptions.BLOB_DESCRIPTOR_SOURCE_TABLE;
 import static org.apache.paimon.utils.Preconditions.checkNotNull;
 
@@ -37,19 +42,23 @@ public final class BlobDescriptorReaderFactory {
     private BlobDescriptorReaderFactory() {}
 
     public static UriReaderFactory create(FileStoreTable table) {
-        Options tableOptions = table.coreOptions().toConfiguration();
+        CoreOptions coreOptions = table.coreOptions();
+        Options tableOptions = coreOptions.toConfiguration();
+        Duration httpKeepAliveTimeout =
+                coreOptions.blobDescriptorHttpKeepAliveTimeout().orElse(null);
         String sourceTable = tableOptions.get(BLOB_DESCRIPTOR_SOURCE_TABLE);
         if (sourceTable != null) {
-            return fromSourceTable(table, sourceTable);
+            return fromSourceTable(table, sourceTable, httpKeepAliveTimeout);
         }
 
         CatalogContext descriptorContext =
                 BlobDescriptorUtils.getCatalogContext(
                         table.catalogEnvironment().catalogContext(), tableOptions);
-        return new UriReaderFactory(descriptorContext);
+        return new UriReaderFactory(descriptorContext, httpKeepAliveTimeout);
     }
 
-    private static UriReaderFactory fromSourceTable(FileStoreTable table, String sourceTable) {
+    private static UriReaderFactory fromSourceTable(
+            FileStoreTable table, String sourceTable, @Nullable Duration httpKeepAliveTimeout) {
         CatalogEnvironment catalogEnvironment = table.catalogEnvironment();
         CatalogLoader catalogLoader =
                 checkNotNull(
@@ -66,7 +75,9 @@ private static UriReaderFactory fromSourceTable(FileStoreTable table, String sou
             FileIO sourceFileIO = catalog.getTable(sourceIdentifier).fileIO();
             // Initialize lazy credentials before serializing FileIO to distributed workers.
             sourceFileIO.isObjectStore();
-            return UriReaderFactory.fromFileIO(sourceFileIO);
+            return httpKeepAliveTimeout == null
+                    ? UriReaderFactory.fromFileIO(sourceFileIO)
+                    : UriReaderFactory.fromFileIO(sourceFileIO, httpKeepAliveTimeout);
         } catch (Exception e) {
             throw new RuntimeException(
                     String.format("Failed to load BLOB descriptor source table '%s'.", sourceTable),
diff --git a/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java b/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java
index ac71ff2552aa..9327338d2431 100644
--- a/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java
@@ -23,6 +23,8 @@
 
 import org.junit.jupiter.api.Test;
 
+import java.time.Duration;
+
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
@@ -270,6 +272,22 @@ public void testBlobCopyBufferSize() {
                 .hasMessageContaining("blob.copy-buffer-size");
     }
 
+    @Test
+    public void testBlobDescriptorHttpKeepAliveTimeout() {
+        Options conf = new Options();
+        assertThat(new CoreOptions(conf).blobDescriptorHttpKeepAliveTimeout()).isEmpty();
+
+        conf.setString(CoreOptions.BLOB_DESCRIPTOR_HTTP_KEEP_ALIVE_TIMEOUT.key(), "60s");
+        assertThat(new CoreOptions(conf).blobDescriptorHttpKeepAliveTimeout())
+                .contains(Duration.ofSeconds(60));
+
+        conf.setString(CoreOptions.BLOB_DESCRIPTOR_HTTP_KEEP_ALIVE_TIMEOUT.key(), "0s");
+        assertThatThrownBy(() -> new CoreOptions(conf).blobDescriptorHttpKeepAliveTimeout())
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining(CoreOptions.BLOB_DESCRIPTOR_HTTP_KEEP_ALIVE_TIMEOUT.key())
+                .hasMessageContaining("greater than 0");
+    }
+
     @Test
     public void testLocalKvDbBlockSize() {
         Options conf = new Options();
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/BlobDescriptorReaderFactoryTest.java b/paimon-core/src/test/java/org/apache/paimon/table/BlobDescriptorReaderFactoryTest.java
index 7aaf3d5b1849..d3585a0e9516 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/BlobDescriptorReaderFactoryTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/BlobDescriptorReaderFactoryTest.java
@@ -32,13 +32,17 @@
 import org.apache.paimon.rest.RESTTokenFileIO;
 import org.apache.paimon.rest.responses.GetTableTokenResponse;
 import org.apache.paimon.utils.InstantiationUtil;
+import org.apache.paimon.utils.UriReader.HttpUriReader;
 import org.apache.paimon.utils.UriReaderFactory;
 
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
 
 import java.nio.file.Files;
+import java.time.Duration;
 import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -52,6 +56,27 @@ public class BlobDescriptorReaderFactoryTest {
 
     @TempDir java.nio.file.Path tempPath;
 
+    @Test
+    public void testPassConfiguredHttpKeepAliveTimeout() {
+        CatalogEnvironment catalogEnvironment = mock(CatalogEnvironment.class);
+        when(catalogEnvironment.catalogContext()).thenReturn(CatalogContext.create(new Options()));
+        FileStoreTable targetTable = mock(FileStoreTable.class);
+        when(targetTable.catalogEnvironment()).thenReturn(catalogEnvironment);
+        when(targetTable.coreOptions())
+                .thenReturn(
+                        CoreOptions.fromMap(
+                                Collections.singletonMap(
+                                        CoreOptions.BLOB_DESCRIPTOR_HTTP_KEEP_ALIVE_TIMEOUT.key(),
+                                        "60s")));
+
+        HttpUriReader reader =
+                (HttpUriReader)
+                        BlobDescriptorReaderFactory.create(targetTable)
+                                .create("http://example.com/blob");
+
+        assertThat(reader).extracting("keepAliveTimeout").isEqualTo(Duration.ofSeconds(60));
+    }
+
     @Test
     public void testUseCatalogContextByDefault() throws Exception {
         java.nio.file.Path tableDirectory = Files.createDirectory(tempPath.resolve("table"));
@@ -110,11 +135,10 @@ public void testRESTTokenFileIOSurvivesSerialization() throws Exception {
 
         FileStoreTable targetTable = mock(FileStoreTable.class);
         when(targetTable.catalogEnvironment()).thenReturn(catalogEnvironment);
-        when(targetTable.coreOptions())
-                .thenReturn(
-                        CoreOptions.fromMap(
-                                Collections.singletonMap(
-                                        "blob-descriptor.source-table", "db.source$branch_rt")));
+        Map<String, String> targetOptions = new HashMap<>();
+        targetOptions.put("blob-descriptor.source-table", "db.source$branch_rt");
+        targetOptions.put(CoreOptions.BLOB_DESCRIPTOR_HTTP_KEEP_ALIVE_TIMEOUT.key(), "60s");
+        when(targetTable.coreOptions()).thenReturn(CoreOptions.fromMap(targetOptions));
 
         UriReaderFactory readerFactory = BlobDescriptorReaderFactory.create(targetTable);
         verify(catalogLoader).load();
@@ -123,6 +147,9 @@ public void testRESTTokenFileIOSurvivesSerialization() throws Exception {
         // the REST client is lost during serialization.
         verify(restApi).loadTableToken(sourceIdentifier);
 
+        HttpUriReader httpReader = (HttpUriReader) readerFactory.create("https://example.com/blob");
+        assertThat(httpReader).extracting("keepAliveTimeout").isEqualTo(Duration.ofSeconds(60));
+
         readerFactory = InstantiationUtil.clone(readerFactory);
         String blobUri = "isolated://" + blobFile;
         UriReaderFactory contextOnlyFactory =
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BlobTableITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BlobTableITCase.java
index 839483f9707e..4ed3265c2ff1 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BlobTableITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BlobTableITCase.java
@@ -1045,7 +1045,9 @@ public void testWriteBlobWithHttpUrlDescriptor() throws Exception {
 
             // Use sys.path_to_descriptor with HTTP URL
             batchSql(
-                    "INSERT INTO blob_table_descriptor VALUES (1, 'http-blob', sys.path_to_descriptor('"
+                    "INSERT INTO blob_table_descriptor"
+                            + " /*+ OPTIONS('blob-descriptor.http.keep-alive-timeout'='60s') */"
+                            + " VALUES (1, 'http-blob', sys.path_to_descriptor('"
                             + httpUrl
                             + "'))");