mirror of
https://github.com/apache/doris.git
synced 2026-11-04 07:31:17 +00:00
bingquanzhao
62dc78fbb0
Replace HttpClient5 async with HttpClient4 sync to fix
CircularRedirectException (1.2.0 -> 1.2.1)
The logstash-output-doris plugin uses Apache HttpClient5 async client to
PUT stream load requests. Against SelectDB Cloud / BYOC FE — which
returns '307 + Connection: close' on stream load — the async client
fails with CircularRedirectException under any meaningful concurrency /
body size.
Root cause:
1. HC5 async does not strictly block body transmission while waiting for
'100 Continue'. When FE returns 307 before issuing 100, the entity
producer has already started writing; FE closing the connection then
yields an IOException mid-transfer.
2. HC5 default exec chain wraps RedirectExec around
AsyncHttpRequestRetryExec. The recoverable IOException triggers an
internal retry that re-enters the same FE -> 307 path, but
RedirectLocations from the first attempt is still populated, so the same
BE URL is detected as 'already visited' and reported as a circular
redirect.
This is a real HC5-vs-HC4 implementation difference, not a configuration
issue. The Doris Flink connector also follows FE 307 to BE in its
default path (autoRedirect=true) and works correctly precisely because
it uses HC4 sync: HC4 honors 'Expect: 100-continue' strictly, so when FE
307s without sending 100, the entity is left unconsumed and HC4's
RedirectExec follows the redirect normally.
This patch aligns the plugin with the Flink connector's HTTP layer:
- bump gem version 1.2.0 -> 1.2.1
- httpclient5 5.4.2 (async) -> httpclient 4.5.13 (sync)
- SimpleRequestBuilder -> HttpPut + ByteArrayEntity (repeatable)
- HttpAsyncClients defaults -> HttpClients with:
* setRequestExecutor(HttpRequestExecutor(60s))
* setRedirectStrategy(DorisRedirectStrategy) (isRedirectable=true,
strip userinfo, normalize empty query)
* setRetryHandler(DefaultHttpRequestRetryHandler(0, false))
* setConnectionReuseStrategy(NoConnectionReuseStrategy.INSTANCE)
* RequestConfig.setExpectContinueEnabled(true)
- Async future plumbing in TableEvents replaced with sync
response_code / response_body / response_error fields.
- Stringify both key and value at request.addHeader call site: HC4's
addHeader(String, String) is strict on types whereas HC5 had a
permissive (String, Object) overload; user configs commonly carry
Float / Integer values like 'max_filter_ratio => 1.0'.
- Drop 's.requirements << jar ...' from gemspec: with JARs vendored
under
lib/, the maven lookup at install time is unnecessary and forced users
to set JARS_SKIP=true for offline installs.
Pipeline configuration, retry queue, save_on_failure, group_commit,
label generation, header handling - all unchanged.
Verified on a SelectDB BYOC cluster mirroring the reported production
shape (16 workers x 10000 batch x 200,000 events):
- Before: 100% requests fail with CircularRedirectException
- After: 20/20 stream loads Status=Success, 200,000/200,000 rows
ingested, 0 HTTP-layer errors.
51 lines
2.1 KiB
Ruby
51 lines
2.1 KiB
Ruby
=begin
|
|
Licensed to the Apache Software Foundation (ASF) under one
|
|
or more contributor license agreements. See the NOTICE file
|
|
distributed with this work for additional information
|
|
regarding copyright ownership. The ASF licenses this file
|
|
to you under the Apache License, Version 2.0 (the
|
|
"License"); you may not use this file except in compliance
|
|
with the License. You may obtain a copy of the License at
|
|
|
|
http://www.apache.org/licenses/LICENSE-2.0
|
|
|
|
Unless required by applicable law or agreed to in writing,
|
|
software distributed under the License is distributed on an
|
|
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
|
|
KIND, either express or implied. See the License for the
|
|
specific language governing permissions and limitations
|
|
under the License.
|
|
=end
|
|
Gem::Specification.new do |s|
|
|
s.name = 'logstash-output-doris'
|
|
s.version = '1.2.1'
|
|
s.author = 'Apache Doris'
|
|
s.email = 'dev@doris.apache.org'
|
|
s.homepage = 'http://doris.apache.org'
|
|
s.licenses = ['Apache-2.0']
|
|
s.summary = "This output lets you `PUT` messages in a batched fashion to Doris HTTP endpoint"
|
|
s.description = "This gem is a logstash plugin required to be installed on top of the Logstash core pipeline using $LS_HOME/bin/plugin install gemname. This gem is not a stand-alone program"
|
|
s.require_paths = ["lib"]
|
|
|
|
s.platform = 'java'
|
|
|
|
# Files
|
|
s.files = Dir['lib/**/*','spec/**/*','*.gemspec','*.md','Gemfile','LICENSE' ]
|
|
|
|
# Tests
|
|
s.test_files = s.files.grep(%r{^(test|spec|features)/})
|
|
|
|
# Special flag to let us know this is actually a logstash plugin
|
|
s.metadata = { "logstash_plugin" => "true", "logstash_group" => "output" }
|
|
|
|
# Gem dependencies
|
|
s.add_runtime_dependency "logstash-core-plugin-api", ">= 1.60", "<= 2.99"
|
|
s.add_runtime_dependency 'jar-dependencies', ">= 0.0.2" # depends on logstash version
|
|
|
|
s.add_development_dependency 'logstash-devutils', '~> 1.3'
|
|
s.add_development_dependency 'sinatra', '~> 1.4'
|
|
s.add_development_dependency 'webrick', '~> 1.9'
|
|
|
|
s.requirements << 'jar org.apache.httpcomponents, httpclient, 4.5.13'
|
|
end
|