From 7866a5368f1919cf06d599480f255340a5bf2f8c Mon Sep 17 00:00:00 2001 From: "hongli.wwj" Date: Thu, 20 Aug 2026 18:23:47 +0800 Subject: [PATCH] [api] Configure keep-alive timeout for HTTP BLOB descriptors --- docs/docs/multimodal-table/blob.mdx | 23 ++++ docs/generated/core_configuration.html | 6 + .../java/org/apache/paimon/CoreOptions.java | 23 ++++ .../apache/paimon/rest/HttpClientUtils.java | 129 +++++++++++++++--- .../paimon/rest/HttpClientUtilsTest.java | 110 ++++++++++++++- .../paimon/utils/BlobDescriptorUtils.java | 5 +- .../org/apache/paimon/utils/UriReader.java | 31 ++++- .../apache/paimon/utils/UriReaderFactory.java | 55 +++++++- .../paimon/utils/BlobDescriptorUtilsTest.java | 31 +++++ .../paimon/utils/UriReaderFactoryTest.java | 28 ++++ .../paimon/schema/SchemaValidation.java | 1 + .../table/BlobDescriptorReaderFactory.java | 21 ++- .../org/apache/paimon/CoreOptionsTest.java | 18 +++ .../BlobDescriptorReaderFactoryTest.java | 37 ++++- .../apache/paimon/flink/BlobTableITCase.java | 4 +- 15 files changed, 482 insertions(+), 40 deletions(-) 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 @@ String Comma-separated field names to treat as BLOB fields and store as serialized BlobDescriptor bytes inline in data files. + +
blob-descriptor.http.keep-alive-timeout
+ (none) + Duration + 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. +
blob-descriptor.source-table
(none) 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 9a6fd514dfbc..fa0c4e2bd220 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 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 BLOB_WRITE_NULL_ON_MISSING_FILE = key("blob-write-null-on-missing-file") .booleanType() @@ -4457,6 +4469,17 @@ public boolean blobAsDescriptor() { return options.get(BLOB_AS_DESCRIPTOR); } + public Optional blobDescriptorHttpKeepAliveTimeout() { + Optional 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 <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 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 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 descriptorSpecified = new HashMap<>(); for (Map.Entry 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 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 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 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 blobDescriptorFields = validateBlobDescriptorFields(tableRowType, options); Set 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 266ff305f406..50dbc6f414e3 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; @@ -267,6 +269,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 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 + "'))");