RockteMQ-AI commented on code in PR #535:
URL: https://github.com/apache/rocketmq-connect/pull/535#discussion_r3902561300
##########
connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/HttpSinkTask.java:
##########
@@ -140,6 +146,32 @@ public void put(List<ConnectRecord> records) throws
ConnectException {
}
}
+ /**
+ * Get a formatted url that will replace a placeholder with Extension
Values
+ *
+ * @param url the source url str
+ * @param extensions ConnectRecord Extension Values
+ * @return the formatted url
+ */
+ private String formatUrl(String url, KeyValue extensions) {
Review Comment:
PATTERN.matcher(url).matches() only returns true when the ENTIRE url is a
single `{placeholder}`. For any realistic template like
`http://host/api/{tenant}/orders`, matches() is false and formatUrl() returns
the url unchanged — the feature silently no-ops for embedded placeholders, even
though the replacement loop below is clearly written for them. Use find() (or
drop the gate and just attempt replacement). Also note `\w+` excludes keys
containing `.` or `-`, which are common in connect extension keys, so
`{connect.topic}` would never be detected even after switching to find().
##########
connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/HttpSinkTask.java:
##########
@@ -140,6 +146,32 @@ public void put(List<ConnectRecord> records) throws
ConnectException {
}
}
+ /**
+ * Get a formatted url that will replace a placeholder with Extension
Values
+ *
+ * @param url the source url str
+ * @param extensions ConnectRecord Extension Values
+ * @return the formatted url
+ */
+ private String formatUrl(String url, KeyValue extensions) {
+ if (!PATTERN.matcher(url).matches()) {
+ return url;
+ }
+ if (extensions != null && extensions.keySet() != null) {
+ Set<String> keys = extensions.keySet();
+ String template = url;
+ for (String key : keys) {
+ String value = extensions.getString(key);
+ if (StringUtils.isNotEmpty(value)) {
+ // simple replaced the placeholder
Review Comment:
Extension values are untrusted record data inserted into the URL with no
encoding. A value containing a space, `&`, `#`, `/`, or `?` yields a malformed
URL or silently alters the path/query semantics. URL-encode values (e.g.
URLEncoder.encode(value, StandardCharsets.UTF_8)). Additionally, sequential
String.replace calls let a value that itself contains `{otherKey}` get
re-replaced by a later iteration; a single-pass Matcher.appendReplacement (with
Matcher.quoteReplacement) avoids both problems.
##########
connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/HttpSinkTask.java:
##########
@@ -140,6 +146,32 @@ public void put(List<ConnectRecord> records) throws
ConnectException {
}
}
+ /**
+ * Get a formatted url that will replace a placeholder with Extension
Values
+ *
+ * @param url the source url str
+ * @param extensions ConnectRecord Extension Values
+ * @return the formatted url
+ */
+ private String formatUrl(String url, KeyValue extensions) {
+ if (!PATTERN.matcher(url).matches()) {
+ return url;
+ }
+ if (extensions != null && extensions.keySet() != null) {
+ Set<String> keys = extensions.keySet();
+ String template = url;
+ for (String key : keys) {
+ String value = extensions.getString(key);
+ if (StringUtils.isNotEmpty(value)) {
+ // simple replaced the placeholder
+ template = template.replace("{" + key + "}", value);
+ }
+ }
Review Comment:
If a placeholder has no matching extension key, or the value is null/empty
(skipped at line 164), the raw `{key}` is left in the URL and the request is
sent to an invalid endpoint — surfacing only as a confusing downstream 404/DNS
error with nothing in the logs. Log a warning listing unresolved placeholders
(or throw a ConnectException) so bad records are diagnosable and can be routed
to dead-letter handling.
##########
connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/HttpSinkTask.java:
##########
@@ -48,13 +48,17 @@
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
public class HttpSinkTask extends SinkTask {
private static final Logger log =
LoggerFactory.getLogger(HttpSinkTask.class);
+ // the Regex Pattern like '{number}'
+ protected static final Pattern PATTERN = Pattern.compile("\\{(\\w+)\\}");
Review Comment:
PATTERN is declared protected but is only used inside this class, and the
name is generic enough to collide with subclass fields. Prefer `private static
final Pattern URL_PLACEHOLDER_PATTERN`.
##########
connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/HttpSinkTask.java:
##########
@@ -140,6 +146,32 @@ public void put(List<ConnectRecord> records) throws
ConnectException {
}
}
+ /**
+ * Get a formatted url that will replace a placeholder with Extension
Values
+ *
+ * @param url the source url str
+ * @param extensions ConnectRecord Extension Values
+ * @return the formatted url
+ */
Review Comment:
The PR checklist claims >80% unit-test coverage, but no tests were added in
this PR. formatUrl has several distinct branches worth covering: whole-URL
placeholder, embedded placeholders, missing extension key, null/empty value,
and values with special characters. Consider extracting the logic to a
package-private static helper (or testing through put() with the HTTP call
mocked) so it can be unit-tested.
--
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]
