featzhang commented on code in PR #27568:
URL: https://github.com/apache/flink/pull/27568#discussion_r3121305949
##########
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) {
Review Comment:
✅ Fixed: Renamed LOCK to CACHE_LOCK for better clarity. Updated in commit
3eb6e7b7f1a.
##########
flink-models/flink-model-triton/src/main/java/org/apache/flink/model/triton/TritonOptions.java:
##########
@@ -165,4 +165,89 @@ private TritonOptions() {
+ "Example: %s",
code("'X-Custom-Header:value,X-Another:value2'"))
.build());
+
+ // ========== Connection Pool Management Options ==========
+
+ public static final ConfigOption<Integer> CONNECTION_POOL_MAX_IDLE =
+ ConfigOptions.key("connection-pool-max-idle")
+ .intType()
+ .defaultValue(20)
+ .withDescription(
+ Description.builder()
+ .text(
+ "Maximum number of idle
connections to keep in the pool. "
+ + "Higher values reduce
connection setup overhead but consume more memory. "
+ + "Recommended: 10-50
depending on parallelism and QPS. "
Review Comment:
✅ Fixed: Added explanation that QPS means "Queries Per Second". Updated in
commit 3eb6e7b7f1a.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
