github-actions[bot] commented on code in PR #65766:
URL: https://github.com/apache/doris/pull/65766#discussion_r3602017064
##########
extension/logstash/lib/logstash/outputs/doris.rb:
##########
@@ -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
Review Comment:
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.
##########
extension/logstash/lib/logstash/outputs/doris.rb:
##########
@@ -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)
Review Comment:
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.
##########
extension/logstash/lib/logstash/outputs/doris.rb:
##########
@@ -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)
+ .setSocketTimeout(@socket_timeout_ms)
Review Comment:
`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.
##########
extension/logstash/lib/logstash/outputs/doris.rb:
##########
@@ -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)
+ .setSocketTimeout(@socket_timeout_ms)
Review Comment:
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.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
