You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "featzhang (via GitHub)" <gi...@apache.org> on 2026/04/22 02:47:30 UTC
Re: [PR] [FLINK-38857][models] Add HTTP connection pool management for Triton inference [flink]
featzhang commented on code in PR #27568:
URL: https://github.com/apache/flink/pull/27568#discussion_r3121305310
##########
flink-models/flink-model-triton/src/main/java/org/apache/flink/model/triton/TritonUtils.java:
##########
@@ -51,40 +66,148 @@ public class TritonUtils {
private static final Object LOCK = new Object();
- private static final Map<Long, ClientValue> cache = new HashMap<>();
+ private static final Map<ClientKey, ClientValue> cache = new HashMap<>();
+
+ /** Connection pool configuration holder. */
+ public static class ConnectionPoolConfig {
+ public final int maxIdleConnections;
+ public final long keepAliveDurationMs;
+ public final int maxTotalConnections;
+ public final long connectionTimeoutMs;
+ public final boolean reuseEnabled;
+ public final boolean monitoringEnabled;
+
+ public ConnectionPoolConfig(
+ int maxIdleConnections,
+ long keepAliveDurationMs,
+ int maxTotalConnections,
+ long connectionTimeoutMs,
+ boolean reuseEnabled,
+ boolean monitoringEnabled) {
+ this.maxIdleConnections = maxIdleConnections;
+ this.keepAliveDurationMs = keepAliveDurationMs;
+ this.maxTotalConnections = maxTotalConnections;
+ this.connectionTimeoutMs = connectionTimeoutMs;
+ this.reuseEnabled = reuseEnabled;
+ this.monitoringEnabled = monitoringEnabled;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ ConnectionPoolConfig that = (ConnectionPoolConfig) o;
+ return maxIdleConnections == that.maxIdleConnections
+ && keepAliveDurationMs == that.keepAliveDurationMs
+ && maxTotalConnections == that.maxTotalConnections
+ && connectionTimeoutMs == that.connectionTimeoutMs
+ && reuseEnabled == that.reuseEnabled
+ && monitoringEnabled == that.monitoringEnabled;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(
+ maxIdleConnections,
+ keepAliveDurationMs,
+ maxTotalConnections,
+ connectionTimeoutMs,
+ reuseEnabled,
+ monitoringEnabled);
+ }
+ }
/**
* Creates or retrieves a cached HTTP client with the specified configuration.
*
* <p>This method implements reference-counted client pooling. Clients with identical timeout
- * settings are shared across multiple callers.
+ * and pool settings are shared across multiple callers.
*
- * @param timeoutMs Timeout in milliseconds for connect, read, and write operations
+ * @param timeoutMs Timeout in milliseconds for read and write operations
+ * @param poolConfig Connection pool configuration
* @return A shared or new OkHttpClient instance
*/
- public static OkHttpClient createHttpClient(long timeoutMs) {
+ public static OkHttpClient createHttpClient(long timeoutMs, ConnectionPoolConfig poolConfig) {
+ ClientKey key = new ClientKey(timeoutMs, poolConfig);
+
synchronized (LOCK) {
- ClientValue value = cache.get(timeoutMs);
+ ClientValue value = cache.get(key);
if (value != null) {
- LOG.debug("Returning an existing Triton HTTP client.");
+ LOG.debug("Returning existing Triton HTTP client (reference count: {}).",
+ value.referenceCount.get() + 1);
value.referenceCount.incrementAndGet();
Review Comment:
✅ Fixed: Moved incrementAndGet() before LOG.debug to avoid side effects in logging. Updated in commit 3eb6e7b7f1a.
##########
flink-models/flink-model-triton/src/main/java/org/apache/flink/model/triton/TritonUtils.java:
##########
@@ -51,40 +66,148 @@ public class TritonUtils {
private static final Object LOCK = new Object();
- private static final Map<Long, ClientValue> cache = new HashMap<>();
+ private static final Map<ClientKey, ClientValue> cache = new HashMap<>();
+
+ /** Connection pool configuration holder. */
+ public static class ConnectionPoolConfig {
+ public final int maxIdleConnections;
+ public final long keepAliveDurationMs;
+ public final int maxTotalConnections;
+ public final long connectionTimeoutMs;
+ public final boolean reuseEnabled;
+ public final boolean monitoringEnabled;
+
+ public ConnectionPoolConfig(
+ int maxIdleConnections,
+ long keepAliveDurationMs,
+ int maxTotalConnections,
+ long connectionTimeoutMs,
+ boolean reuseEnabled,
+ boolean monitoringEnabled) {
+ this.maxIdleConnections = maxIdleConnections;
+ this.keepAliveDurationMs = keepAliveDurationMs;
+ this.maxTotalConnections = maxTotalConnections;
+ this.connectionTimeoutMs = connectionTimeoutMs;
+ this.reuseEnabled = reuseEnabled;
+ this.monitoringEnabled = monitoringEnabled;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ ConnectionPoolConfig that = (ConnectionPoolConfig) o;
+ return maxIdleConnections == that.maxIdleConnections
+ && keepAliveDurationMs == that.keepAliveDurationMs
+ && maxTotalConnections == that.maxTotalConnections
+ && connectionTimeoutMs == that.connectionTimeoutMs
+ && reuseEnabled == that.reuseEnabled
+ && monitoringEnabled == that.monitoringEnabled;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(
+ maxIdleConnections,
+ keepAliveDurationMs,
+ maxTotalConnections,
+ connectionTimeoutMs,
+ reuseEnabled,
+ monitoringEnabled);
+ }
+ }
/**
* Creates or retrieves a cached HTTP client with the specified configuration.
*
* <p>This method implements reference-counted client pooling. Clients with identical timeout
- * settings are shared across multiple callers.
+ * and pool settings are shared across multiple callers.
*
- * @param timeoutMs Timeout in milliseconds for connect, read, and write operations
+ * @param timeoutMs Timeout in milliseconds for read and write operations
+ * @param poolConfig Connection pool configuration
* @return A shared or new OkHttpClient instance
*/
- public static OkHttpClient createHttpClient(long timeoutMs) {
+ public static OkHttpClient createHttpClient(long timeoutMs, ConnectionPoolConfig poolConfig) {
+ ClientKey key = new ClientKey(timeoutMs, poolConfig);
+
synchronized (LOCK) {
- ClientValue value = cache.get(timeoutMs);
+ ClientValue value = cache.get(key);
if (value != null) {
- LOG.debug("Returning an existing Triton HTTP client.");
+ LOG.debug("Returning existing Triton HTTP client (reference count: {}).",
+ value.referenceCount.get() + 1);
value.referenceCount.incrementAndGet();
return value.client;
}
- LOG.debug("Building a new Triton HTTP client.");
+ LOG.info("Building new Triton HTTP client with connection pool configuration.");
+
+ // Configure connection pool
+ ConnectionPool connectionPool;
+ if (poolConfig.reuseEnabled) {
+ connectionPool =
+ new ConnectionPool(
+ poolConfig.maxIdleConnections,
+ poolConfig.keepAliveDurationMs,
+ TimeUnit.MILLISECONDS);
+ } else {
+ // Disable pooling by setting maxIdle to 0
+ connectionPool = new ConnectionPool(0, 1, TimeUnit.MILLISECONDS);
+ }
+
+ // Configure dispatcher for concurrent requests
+ Dispatcher dispatcher = new Dispatcher();
+ dispatcher.setMaxRequests(poolConfig.maxTotalConnections);
+ dispatcher.setMaxRequestsPerHost(poolConfig.maxTotalConnections);
+
+ // Build HTTP client
OkHttpClient client =
new OkHttpClient.Builder()
- .connectTimeout(timeoutMs, TimeUnit.MILLISECONDS)
+ .connectTimeout(poolConfig.connectionTimeoutMs, TimeUnit.MILLISECONDS)
.readTimeout(timeoutMs, TimeUnit.MILLISECONDS)
.writeTimeout(timeoutMs, TimeUnit.MILLISECONDS)
+ .connectionPool(connectionPool)
+ .dispatcher(dispatcher)
.retryOnConnectionFailure(true)
.build();
- cache.put(timeoutMs, new ClientValue(client));
+ ClientValue clientValue = new ClientValue(client, poolConfig);
+ cache.put(key, clientValue);
+
+ // Start monitoring if enabled
+ if (poolConfig.monitoringEnabled) {
+ clientValue.startMonitoring();
+ }
+
+ LOG.info(
+ "Triton HTTP client created - Pool: maxIdle={}, keepAlive={}ms, maxTotal={}, connTimeout={}ms",
+ poolConfig.maxIdleConnections,
+ poolConfig.keepAliveDurationMs,
+ poolConfig.maxTotalConnections,
+ poolConfig.connectionTimeoutMs);
+
return client;
}
}
+ /**
+ * Backward compatibility: creates client with default pool configuration.
+ *
+ * @param timeoutMs Timeout in milliseconds
+ * @return OkHttpClient instance
+ */
+ public static OkHttpClient createHttpClient(long timeoutMs) {
+ ConnectionPoolConfig defaultConfig =
+ new ConnectionPoolConfig(
+ 20, // maxIdleConnections
+ 300_000, // keepAliveDurationMs (5 minutes)
Review Comment:
The underscores in numeric literals (e.g., 300_000, 10_000) are valid Java syntax since Java 7, used to improve readability of large numbers. This is a common practice in Java codebases including Flink.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org