update http by zhaohai666 · Pull Request #384 · apache/rocketmq-connect

zhaohai666 · 2022-12-01T02:40:22Z

The http protocol supports Basic, Api Key and OAuth2 authentication.

RockteMQ-AI · 2026-07-30T10:45:50Z

⚠️ Merge conflict detected

This PR has conflicts with the base branch and cannot be merged. Please rebase or merge the base branch into your branch and resolve the conflicts:

git fetch origin
git checkout update-connect-http
git rebase origin/main
# resolve conflicts, then:
git push --force-with-lease

This is a one-time reminder. Feel free to @mention me for a re-review after conflicts are resolved.

Automated notification by github-manager-bot

RockteMQ-AI

Summary

This PR modifies 25 file(s) with 2415 lines of diff. Changes look reasonable.

Automated review by github-manager-bot

Additional notes (not anchored to a changed line)

[INFO] connectors/rocketmq-connect-http/README.md:1 — Large diff (2415 lines). Consider breaking into smaller, focused PRs for easier review. (line outside diff)

RockteMQ-AI

Summary

PR received and logged for review. This PR requires detailed code review by a maintainer.

Diff size: 2415 lines
Author: zhaohai666 (CONTRIBUTOR)

Automated review by RockteMQ-AI

RockteMQ-AI

Summary

Review of PR #384: update http

Findings: 16 issue(s) identified (4 critical).
CLA: unknown

Please address the inline comments above.

Automated review by github-manager-bot

RockteMQ-AI · 2026-08-23T20:44:18Z

+        oauth2ClientSecret = CheckUtils.checkNullReturnDefault(config.getString(HttpConstant.OAUTH2_CLIENTSECRET_CONSTANT));
+        oauth2HttpMethod = CheckUtils.checkNullReturnDefault(config.getString(HttpConstant.OAUTH2_HTTP_METHOD_CONSTANT));
+        proxyType = CheckUtils.checkNullReturnDefault(config.getString(HttpConstant.PROXY_TYPE_CONSTANT));
+        proxyHost = CheckUtils.checkNullReturnDefault(config.getString(HttpConstant.PROXY_HOST_CONSTANT));


scheduledExecutorService is never initialized. The field is declared but only set via setScheduledExecutorService(), which is never called internally. In start(), scheduledExecutorService.scheduleAtFixedRate(...) will throw NullPointerException, preventing the task from starting entirely.

RockteMQ-AI · 2026-08-23T20:44:18Z

+            if (httpCallback.isFailed()) {
+                throw new RetriableException(httpCallback.getMsg());
+            }
        } catch (Exception e) {


The catch(Exception e) block wraps RetriableException (thrown at lines 97 and 100) in a new RuntimeException. This prevents the connector framework from recognizing retriable failures and retrying them. RetriableException should be caught separately and re-thrown unwrapped, or the catch block should not catch it.

RockteMQ-AI · 2026-08-23T20:44:19Z

+    public void init(ClientConfig config) {
+        try {
+            SSLContextBuilder sslContextBuilder = new SSLContextBuilder().loadTrustMaterial(null, new TrustStrategy() {
+                @Override


SSL/TLS verification is completely disabled: TrustStrategy.isTrusted always returns true (line 77) and NoopHostnameVerifier.verify always returns true (line 253). This makes all HTTPS connections vulnerable to man-in-the-middle attacks. This should at minimum be opt-in via a configuration flag, not the unconditional default.

RockteMQ-AI · 2026-08-23T20:44:19Z

+package org.apache.rocketmq.connect.http.sink.auth;
+
+import com.google.common.collect.Maps;
+import com.sun.org.apache.xerces.internal.impl.dv.util.Base64;


Uses com.sun.org.apache.xercesinternal.impl.dv.util.Base64, an internal JDK class not part of the public API. It may not exist in non-Oracle JDKs or future JDK versions, causing ClassNotFoundException at runtime. Should use java.util.Base64.getEncoder().encodeToString() instead.

RockteMQ-AI · 2026-08-23T20:44:19Z

+                HttpClientContext.create(), this.socksProxyConfig, httpCallback, MDC.get(LOG_SIFT_TAG));
+        Future<String> submit = executorServicePool.submit(httpRequestCallable);
+        String result = submit.get();
+        log.info("ApacheHttpClientImpl | execute｜ success | result : {}", result);


execute() calls submit.get() which blocks indefinitely with no timeout. This makes the countDownLatch-based 30-second timeout in put() non-functional: if any HTTP request hangs and no 'timeout' config is set, submit.get() blocks forever and the latch await is never reached. Use submit.get(timeout, TimeUnit) or Future.cancel() on timeout.

RockteMQ-AI · 2026-08-23T20:44:19Z

+    private static final Logger log = LoggerFactory.getLogger(ApacheHttpClientImpl.class);
+
+    private static ExecutorService executorServicePool = new ThreadPoolExecutor(200, 2000, 600, TimeUnit.SECONDS,
+            new LinkedBlockingDeque<Runnable>(1000), new DefaultThreadFactory("ApacheHttpClientRequestThread"));


executorServicePool is a static field shared across all connector instances but is never shut down in close(). This causes a thread leak (200 core threads) when connectors are stopped or recreated. The pool should be an instance field and shut down in close(), or use a shared lifecycle-managed pool.

RockteMQ-AI · 2026-08-23T20:44:19Z

+            PoolingHttpClientConnectionManager connManager = new PoolingHttpClientConnectionManager(reg,
+                    new FakeDnsResolver());
+            connManager.setMaxTotal(400);
+            connManager.setDefaultMaxPerRoute(500);


Connection pool configuration is illogical: setMaxTotal(400) is less than setDefaultMaxPerRoute(500). Since per-route connections cannot exceed the total, the effective per-route limit is capped at 400. These values should be consistent, with maxTotal >= defaultMaxPerRoute.

RockteMQ-AI · 2026-08-23T20:44:19Z

+                headerMap.put(entry.getKey(), (String) entry.getValue());
+            }
+        }
    }


validate() always checks authType against AuthTypeEnum and throws if no match is found. But authType is marked optional (NO) in the README parameter table. If authType is not configured (null), the stream filter matches no enum and validation throws, contradicting the documented contract.

RockteMQ-AI · 2026-08-23T20:44:19Z

+        clientConfig.setTimeout(timeout);
+        clientConfig.setOauth2HttpMethod(oauth2HttpMethod);
+        clientConfig.setOauth2Endpoint(oauth2Endpoint);
+        clientConfig.setBasicUser(basicUser);


In addHeaderMap, the else branch casts entry.getValue() to String without verifying the actual type. If the JSON header value is an Integer, Boolean, or JSONArray (not a JSONObject), this throws ClassCastException. Should use String.valueOf(entry.getValue()) or check all possible types.

RockteMQ-AI · 2026-08-23T20:44:19Z

 http://${runtime-ip}:${runtime-port}/connectors/${rocketmq-http-sink-connector-name}
-?config={"source-rocketmq":"${runtime-ip}:${runtime-port}","source-cluster":"${broker-cluster}","connector-class":"org.apache.rocketmq.connect.http.sink.HttpSinkConnector","connect-topicname" : "${connect-topicname}","url":"${url}"}
+?config={"source-rocketmq":"${runtime-ip}:${runtime-port}","source-cluster":"${broker-cluster}","connector-class":"HttpSinkConnector",
+"urlPattern":"${urlPattern}","method":"${method}","queryStringParameters":"${queryStringParameters}","headerParameters":"${headerParameters}","bodys":"${bodys}","authType":"${authType}","basicUser":"${basicUser}","basicPassword":"${basicPassword}",


The JSON config example has duplicate keys: proxyPort appears 3 times and proxyUser appears 2 times across lines 18-19. Duplicate JSON keys cause undefined behavior in parsers (only the last value is typically kept). These should be deduplicated to match the parameter table.

zh378814 added 3 commits December 1, 2022 10:36

update http

966b51b

update validate

bab09e9

update result

ba353ba

RockteMQ-AI reviewed Aug 5, 2026

View reviewed changes

RockteMQ-AI reviewed Aug 20, 2026

View reviewed changes

RockteMQ-AI suggested changes Aug 23, 2026

View reviewed changes

Uh oh!

Conversation

zhaohai666 commented Dec 1, 2022

Uh oh!

RockteMQ-AI commented Jul 30, 2026

Uh oh!

RockteMQ-AI left a comment

Choose a reason for hiding this comment

Summary

Additional notes (not anchored to a changed line)

Uh oh!

RockteMQ-AI left a comment

Choose a reason for hiding this comment

Summary

Uh oh!

RockteMQ-AI left a comment

Choose a reason for hiding this comment

Summary

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants