Skip to content

[fix](logstash) Add HTTP timeouts and SO_KEEPALIVE to prevent pipeline hang #65766

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
smallhibiscus wants to merge 1 commit into
base: master
Choose a base branch
from
+56 −3
Open

[fix](logstash) Add HTTP timeouts and SO_KEEPALIVE to prevent pipeline hang #65766

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 24 additions & 0 deletions extension/logstash/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,30 @@ jruby -S gem build logstash-output-doris.gemspec

Produces `logstash-output-doris-<version>-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
Expand Down
33 changes: 31 additions & 2 deletions extension/logstash/lib/logstash/outputs/doris.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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

@github-actions github-actions Bot Jul 17, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

These three public settings accept a wider domain than the Java API they feed. Logstash's :number accepts fractions, and on both README-supported JRuby lines 0.5 is truncated to Java 0; HttpClient interprets zero as an infinite timeout. Negative values likewise leave the finite-timeout guarantee undefined/disabled, while values above 2147483647 raise RangeError during register. The startup log prints the original Ruby value, so it can even claim 0.5 ms while the effective timeout is infinite. Please validate positive whole milliseconds in 1..2147483647, pass an explicit Java int, document the range, and add JRuby-facing boundary tests for all three options.


# 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
Expand Down Expand Up @@ -122,22 +134,39 @@ 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)
# - DefaultHttpRequestRetryHandler(0, false) : NO retries -> avoid spurious CircularRedirect
# - 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)

@github-actions github-actions Bot Jul 17, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please size the connection pool before making pool leases expire after 60 seconds. HttpClients.custom defaults to only 2 simultaneous connections per route (20 total), while this plugin is concurrency :shared and the retry thread uses the same client. With one effective FE/BE route and two valid loads running for more than 60 seconds, a third worker times out here with ConnectionPoolTimeoutException without sending its request; handle_request then requeues it, and request exceptions bypass max_retries. This turns normal shared concurrency into an unbounded retry cycle. Configure maxConnPerRoute/maxConnTotal for the supported worker count (or keep the lease wait consistent with valid in-flight loads) and cover three same-route executions in a test.

.setSocketTimeout(@socket_timeout_ms)

@github-actions github-actions Bot Jul 17, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

setSocketTimeout only configures SO_TIMEOUT for InputStream.read; it does not time the synchronous ByteArrayEntity upload inside @client.execute. If the server returns 100 Continue and then stops consuming a batch larger than the TCP send buffers (for example, a live zero-window peer), the worker blocks in OutputStream.write; SO_KEEPALIVE also cannot abort that live connection. The advertised anti-hang behavior therefore still misses the request-body phase. Please enforce an overall/cancellable request deadline that closes/aborts the request while it is being written, or narrow the README/release claim to response-read hangs.

@github-actions github-actions Bot Jul 17, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

A timeout-driven retry is not safe merely because it reuses the label. The retried HTTP request gets a new Stream Load request ID; if the original transaction is still PREPARE, Doris returns Label Already Exists with ExistingJobStatus: RUNNING. The existing handler treats every label collision as terminal success and discards this batch, but the original can still fail/abort afterward, leaving the rows permanently lost. Please retain and reconcile RUNNING/PRECOMMITTED collisions (poll until FINISHED, or retry the same label if it aborts) and only treat a finished existing job as success. Add a race test where the original is running at the collision and then aborts.

.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)

Expand Down
2 changes: 1 addition & 1 deletion extension/logstash/logstash-output-doris.gemspec
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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'
Expand Down
Loading