From c55e79f25c6b65bdd73e40d26cf048c58a6e31fc Mon Sep 17 00:00:00 2001 From: smallhibiscus <844981280> Date: Fri, 17 Jul 2026 16:36:37 +0800 Subject: [PATCH] [fix](logstash) Add HTTP timeouts and SO_KEEPALIVE to prevent pipeline hang (1.2.1 -> 1.2.2) ### What problem does this PR solve? Issue Number: close #xxx Related PR: #63181 Problem Summary: logstash-output-doris 1.2.0 hung forever on TCP half-open connections because Future.get() and HttpAsyncClient had no timeouts. #63181 migrated to HttpClient4 sync with NoConnectionReuseStrategy, which removes the Future.get() hang and avoids keep-alive connection reuse, but RequestConfig still lacked connect/socket timeouts and SO_KEEPALIVE was not enabled. Without socketTimeout, @client.execute() can still block forever when the peer dies mid-request. This patch (1.2.1 -> 1.2.2): - set connectTimeout / connectionRequestTimeout / socketTimeout on RequestConfig - enable SO_KEEPALIVE via SocketConfig - expose connect_timeout_ms, connection_request_timeout_ms, socket_timeout_ms as configurable options (defaults: 60s / 60s / 600s) ### Release note logstash-output-doris 1.2.2: add HTTP connect/socket timeouts and SO_KEEPALIVE to prevent worker threads hanging forever on TCP half-open connections. New optional configs: connect_timeout_ms, connection_request_timeout_ms, socket_timeout_ms. ### Check List (For Author) - Test: - Manual test / No need to test (config-only HttpClient setup; timeout behavior matches Apache HttpClient4 RequestConfig / SocketConfig) - Behavior changed: Yes (HTTP requests now fail with timeout instead of hanging) - Does this need documentation: No (README updated in this PR) Co-authored-by: Cursor --- extension/logstash/README.md | 24 ++++++++++++++ .../logstash/lib/logstash/outputs/doris.rb | 33 +++++++++++++++++-- .../logstash/logstash-output-doris.gemspec | 2 +- 3 files changed, 56 insertions(+), 3 deletions(-) diff --git a/extension/logstash/README.md b/extension/logstash/README.md index c29d4c1ca40b8c..3c25987c9fa549 100644 --- a/extension/logstash/README.md +++ b/extension/logstash/README.md @@ -39,6 +39,30 @@ jruby -S gem build logstash-output-doris.gemspec Produces `logstash-output-doris--java.gem`. +## HTTP timeouts (v1.2.2+) + +To avoid worker threads hanging forever on TCP half-open connections, the +plugin configures HttpClient4 timeouts and enables `SO_KEEPALIVE`: + +| Option | Default | Meaning | +|--------|---------|---------| +| `connect_timeout_ms` | `60000` | TCP connect timeout | +| `connection_request_timeout_ms` | `60000` | Timeout leasing a connection from the pool | +| `socket_timeout_ms` | `600000` | Socket read timeout (response wait) | + +Example: + +``` +output { + doris { + http_hosts => ["http://fe:8030"] + ... + connect_timeout_ms => 60000 + socket_timeout_ms => 600000 + } +} +``` + ## Install The jars are already vendored inside the gem, so the install hook does not diff --git a/extension/logstash/lib/logstash/outputs/doris.rb b/extension/logstash/lib/logstash/outputs/doris.rb index b1e28eb4ae2e6c..25447bae076892 100644 --- a/extension/logstash/lib/logstash/outputs/doris.rb +++ b/extension/logstash/lib/logstash/outputs/doris.rb @@ -41,6 +41,7 @@ class LogStash::Outputs::Doris < LogStash::Outputs::Base java_import 'org.apache.http.impl.NoConnectionReuseStrategy' java_import 'org.apache.http.protocol.HttpRequestExecutor' java_import 'org.apache.http.client.config.RequestConfig' + java_import 'org.apache.http.config.SocketConfig' # support multi thread concurrency for performance # so multi_receive() and function it calls are all stateless and thread safe @@ -88,6 +89,17 @@ class LogStash::Outputs::Doris < LogStash::Outputs::Base # max retry queue size in MB, default is 20% max memory of JVM config :max_retry_queue_mb, :validate => :number, :default => java.lang.Runtime.get_runtime.max_memory / 1024 / 1024 / 5 + # HTTP connect timeout in milliseconds (time to establish TCP connection) + config :connect_timeout_ms, :validate => :number, :default => 60_000 + + # HTTP connection request timeout in milliseconds (time to lease a connection from the pool) + config :connection_request_timeout_ms, :validate => :number, :default => 60_000 + + # HTTP socket / read timeout in milliseconds (time waiting for response data). + # Without this, a TCP half-open connection can block the worker thread forever. + # Default 600s to leave headroom for large stream loads. + config :socket_timeout_ms, :validate => :number, :default => 600_000 + def print_plugin_info() @plugins = Gem::Specification.find_all{|spec| spec.name =~ /logstash-output-doris/ } @plugin_name = @plugins[0].name @@ -122,7 +134,7 @@ def http_query(table) end def register - # HttpClient 4.5.13 sync — same setup as Doris Flink connector (HttpUtil.java) + # HttpClient 4.5.13 sync — same setup as Doris Flink / Kettle connectors (HttpUtil.java) # Key points: # - setRequestExecutor(60s) : long wait for 100-continue, FE may delay 307 under load # - setRedirectStrategy : follow 307 on PUT (default DefaultRedirectStrategy refuses) @@ -130,14 +142,31 @@ def register # - NoConnectionReuseStrategy : one connection per request, dodge keep-alive half-close # - setExpectContinueEnabled(true) : critical -> HC4 waits for 100, FE 307s before body is sent, # entity stays unconsumed, RedirectExec follows successfully + # - RequestConfig timeouts : prevent worker threads from hanging forever on TCP + # half-open connections (connect / pool / socket read) + # - SocketConfig SO_KEEPALIVE : let the kernel probe dead peers even without reuse + request_config = RequestConfig.custom + .setConnectTimeout(@connect_timeout_ms) + .setConnectionRequestTimeout(@connection_request_timeout_ms) + .setSocketTimeout(@socket_timeout_ms) + .setExpectContinueEnabled(true) + .build + socket_config = SocketConfig.custom + .setSoKeepAlive(true) + .build @client = HttpClients.custom .setRequestExecutor(HttpRequestExecutor.new(60_000)) .setRedirectStrategy(DorisRedirectStrategy.new) .setRetryHandler(DefaultHttpRequestRetryHandler.new(0, false)) .setConnectionReuseStrategy(NoConnectionReuseStrategy::INSTANCE) - .setDefaultRequestConfig(RequestConfig.custom.setExpectContinueEnabled(true).build) + .setDefaultRequestConfig(request_config) + .setDefaultSocketConfig(socket_config) .build + @logger.info("http timeouts (ms): connect=#{@connect_timeout_ms}, " \ + "connection_request=#{@connection_request_timeout_ms}, " \ + "socket=#{@socket_timeout_ms}; so_keepalive=true") + @request_headers = make_request_headers @logger.info("request headers: ", @request_headers) diff --git a/extension/logstash/logstash-output-doris.gemspec b/extension/logstash/logstash-output-doris.gemspec index daa23cdb39fa6f..66a55d486f3477 100644 --- a/extension/logstash/logstash-output-doris.gemspec +++ b/extension/logstash/logstash-output-doris.gemspec @@ -18,7 +18,7 @@ under the License. =end Gem::Specification.new do |s| s.name = 'logstash-output-doris' - s.version = '1.2.1' + s.version = '1.2.2' s.author = 'Apache Doris' s.email = 'dev@doris.apache.org' s.homepage = 'http://doris.apache.org'