[api] Configure keep-alive timeout for HTTP BLOB descriptors

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.

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>

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);
     }

paimon-api/src/main/java/org/apache/paimon/rest/HttpClientUtils.java

@@ -22,39 +22,50 @@
 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;
 import org.apache.hc.core5.http.ConnectionClosedException;
 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,27 +77,48 @@
                     .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());
         return clientBuilder;
     }
 
     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 @@
     }
 
     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 @@
      * 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 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 @@
      * to <Location>") 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 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 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 @@
 
         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 @@
             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 @@
             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 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) {