davsclaus commented on code in PR #27686:
URL: https://github.com/apache/camel/pull/27686#discussion_r4249462413
##########
components/camel-kamelet/src/main/java/org/apache/camel/component/kamelet/Kamelet.java:
##########
@@ -184,6 +205,164 @@ public static void
extractKameletEnvironmentVariables(Map<String, Object> proper
}
}
+ /**
+ * Extracts advanced options without discarding camelCase and kebab-case
aliases before normalization.
+ */
+ private static Map<String, Object> extractAdvancedOptions(Map<String,
Object> parameters, String prefix) {
+ Map<String, Object> options = new LinkedHashMap<>();
+ if (parameters == null) {
+ return options;
+ }
+
+ for (Map.Entry<String, Object> entry : parameters.entrySet()) {
+ String key = entry.getKey();
+ if (key != null && key.startsWith(prefix)) {
+ options.put(key.substring(prefix.length()), entry.getValue());
+ }
+ }
+
+ return options;
+ }
+
+ /**
+ * Normalizes option names so that camelCase and kebab-case aliases do not
cause the same option to be appended more
+ * than once. When both forms are present, the kebab-case form takes
precedence.
+ */
+ private static Map<String, Object> normalizeParameterNames(Map<String,
Object> parameters) {
+ Map<String, Object> normalized = new LinkedHashMap<>();
+ Map<String, Boolean> canonicalKeys = new HashMap<>();
+
+ if (parameters == null) {
+ return normalized;
+ }
+
+ for (Map.Entry<String, Object> entry : parameters.entrySet()) {
+ String originalName = entry.getKey();
+ String name = StringHelper.camelCaseToDash(originalName);
Review Comment:
This always puts the kebab-case name on the endpoint, even when the user
wrote camelCase, so `sink.pageSize=10` becomes `page-size=10`. For lenient
components (`http`, `vertx-http`, ...) unknown parameters are sent to the
remote server, so the server gets a different name than the user wrote. The
aliases come from `DefaultModel.addProperty`, which adds them after
`KameletComponent` has extracted the options. `KameletComponent` still has the
original key in `sinkOptions`/`sourceOptions`, so it would be better to pass
those through as they are (for example as one map value under a single internal
key) instead of rebuilding them from the alias pairs here.
##########
components/camel-kamelet/src/main/java/org/apache/camel/component/kamelet/Kamelet.java:
##########
@@ -184,6 +205,164 @@ public static void
extractKameletEnvironmentVariables(Map<String, Object> proper
}
}
+ /**
+ * Extracts advanced options without discarding camelCase and kebab-case
aliases before normalization.
+ */
+ private static Map<String, Object> extractAdvancedOptions(Map<String,
Object> parameters, String prefix) {
+ Map<String, Object> options = new LinkedHashMap<>();
+ if (parameters == null) {
+ return options;
+ }
+
+ for (Map.Entry<String, Object> entry : parameters.entrySet()) {
+ String key = entry.getKey();
+ if (key != null && key.startsWith(prefix)) {
+ options.put(key.substring(prefix.length()), entry.getValue());
+ }
+ }
+
+ return options;
+ }
+
+ /**
+ * Normalizes option names so that camelCase and kebab-case aliases do not
cause the same option to be appended more
+ * than once. When both forms are present, the kebab-case form takes
precedence.
+ */
+ private static Map<String, Object> normalizeParameterNames(Map<String,
Object> parameters) {
+ Map<String, Object> normalized = new LinkedHashMap<>();
+ Map<String, Boolean> canonicalKeys = new HashMap<>();
+
+ if (parameters == null) {
+ return normalized;
+ }
+
+ for (Map.Entry<String, Object> entry : parameters.entrySet()) {
+ String originalName = entry.getKey();
+ String name = StringHelper.camelCaseToDash(originalName);
+ boolean isCanonical = originalName.equals(name);
+
+ if (!normalized.containsKey(name) || (isCanonical &&
!canonicalKeys.getOrDefault(name, false))) {
+ normalized.put(name, entry.getValue());
+ canonicalKeys.put(name, isCanonical);
+ }
+ }
+
+ return normalized;
+ }
+
+ /**
+ * Appends query parameters without parsing or re-encoding the endpoint
URI. This preserves property placeholders
+ * and RAW expressions that may appear in the URI.
+ */
+ private static String appendParametersToURI(String uri, Map<String,
Object> parameters) {
+ if (parameters == null || parameters.isEmpty()) {
+ return uri;
+ }
+
+ String result = uri;
+ for (Map.Entry<String, Object> entry : parameters.entrySet()) {
+ String name = entry.getKey();
+ String value = String.valueOf(entry.getValue());
+
+ result = removeQueryParameter(result, name);
+ result += result.contains("?") ? "&" : "?";
+ result += name + "=" + value;
+ }
+
+ return result;
+ }
+
+ /**
+ * Removes a query parameter by name, treating camelCase and kebab-case
names as aliases.
+ */
+ private static String removeQueryParameter(String uri, String
parameterName) {
+ int queryIndex = uri.indexOf('?');
+ if (queryIndex < 0) {
+ return uri;
+ }
+
+ String base = uri.substring(0, queryIndex);
+ String query = uri.substring(queryIndex + 1);
+
+ Map<String, Object> params;
+ try {
+ params = URISupport.parseQuery(query);
+ } catch (URISyntaxException e) {
+ throw new IllegalArgumentException("Invalid URI query: " + query,
e);
+ }
+
+ String normalizedName = StringHelper.camelCaseToDash(parameterName);
+ params.keySet().removeIf(
+ name ->
StringHelper.camelCaseToDash(name).equals(normalizedName));
+
+ return params.isEmpty() ? base : base + "?" +
URISupport.createQueryString(params, false);
+ }
+
+ /**
+ * Finds a single statically defined main sink among top-level outputs.
+ *
+ * <p>
+ * Endpoint schemes are deliberately not used to guess whether a
destination is auxiliary. A log or direct endpoint
+ * can be a legitimate destination, and scheme-based exclusions can select
the wrong endpoint. Only top-level static
+ * to() destinations are considered; destinations inside branches and
WireTap are not.
+ *
+ * @throws IllegalArgumentException if more than one candidate exists,
because the intended sink cannot be
+ * determined unambiguously
+ */
+ private static ToDefinition
findMainSink(Collection<ProcessorDefinition<?>> outputs) {
+ if (outputs == null) {
+ return null;
+ }
+
+ ToDefinition mainSink = null;
+
+ for (ProcessorDefinition<?> output : outputs) {
+ if (output instanceof WireTapDefinition) {
+ continue;
+ }
+
+ if (output instanceof ToDefinition to
+ && !isKameletEndpoint(to.getEndpointUri(), SINK_ID)
+ && !isKameletEndpoint(to.getEndpointUri(), SOURCE_ID)) {
Review Comment:
The docs say "Kamelet endpoints ... are excluded", but only
`kamelet:source`/`kamelet:sink` are skipped here. A sink kamelet built on
another kamelet (`.to("kamelet:aws-s3-sink?...")`) would get the `sink.*`
options appended to that kamelet URI, where they turn into template parameters
of the nested kamelet. Please either skip all `kamelet:` URIs or make the docs
describe the actual behaviour, and add a test for it.
##########
components/camel-kamelet/src/main/java/org/apache/camel/component/kamelet/Kamelet.java:
##########
@@ -230,9 +411,77 @@ public static RouteDefinition
templateToRoute(RouteTemplateDefinition in, Map<St
throw new IllegalArgumentException("Camel route " + rid + " input
does not exist.");
}
- // must make the source and sink endpoints are unique by appending the
route id before we create the route from the template
- if (def.getInput().getEndpointUri().startsWith("kamelet:source")
- ||
def.getInput().getEndpointUri().startsWith("kamelet://source")) {
+ // Extract advanced options for source/sink component endpoints.
+ // The internal.advanced.source. and internal.advanced.sink. prefixes
are reserved for these options.
+ Map<String, Object> sourceOptions = normalizeParameterNames(
+ extractAdvancedOptions(parameters,
"internal.advanced.source."));
+ Map<String, Object> sinkOptions = normalizeParameterNames(
+ extractAdvancedOptions(parameters, "internal.advanced.sink."));
+
+ // Determine Kamelet type: source Kamelets have a real component as
input;
+ // sink/action Kamelets have kamelet:source as input and the real
component in outputs.
+ boolean isSourceKamelet =
!isKameletEndpoint(def.getInput().getEndpointUri(), SOURCE_ID);
+
+ if (isSourceKamelet) {
+ // Source Kamelet: the real component endpoint is the input.
+ if (!sourceOptions.isEmpty()) {
+ String uri = def.getInput().getEndpointUri();
+ try {
+ def.getInput().setUri(appendParametersToURI(uri,
sourceOptions));
+ } catch (Exception e) {
+ throw new IllegalArgumentException(
+ "Failed to apply advanced source options to
endpoint URI: " + uri, e);
+ }
+ }
+
+ if (!sinkOptions.isEmpty()) {
+ throw new IllegalArgumentException(
+ "sink.* advanced options are not valid for source
Kamelets (input is '"
+ +
def.getInput().getEndpointUri() + "'). Use source.* options instead.");
+ }
+ } else {
+ // Sink/action Kamelet: the real component endpoint is in the
outputs.
+ if (!sourceOptions.isEmpty()) {
+ throw new IllegalArgumentException(
+ "source.* advanced options are not valid for
sink/action Kamelets (input is '"
+ +
def.getInput().getEndpointUri() + "'). Use sink.* options instead.");
+ }
+
+ if (!sinkOptions.isEmpty()) {
+ // If a dynamic destination exists, the intended static
destination cannot be determined reliably.
+ if (hasUnsupportedDynamicSinkTarget(def.getOutputs())) {
+ throw new IllegalArgumentException(
+ "sink.* advanced options cannot be applied because
this Kamelet contains a dynamic "
+ + "destination (toD,
enrich, or pollEnrich). Configure the options directly "
+ + "in the destination
endpoint URI.");
+ }
+
+ ToDefinition mainSink;
+ try {
+ mainSink = findMainSink(def.getOutputs());
+ } catch (IllegalArgumentException e) {
+ throw e;
+ }
Review Comment:
Catching and rethrowing the same exception does nothing:
```suggestion
ToDefinition mainSink = findMainSink(def.getOutputs());
```
--
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]
