davsclaus commented on code in PR #27832:
URL: https://github.com/apache/camel/pull/27832#discussion_r4270883523
##########
docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc:
##########
@@ -3143,6 +3143,12 @@ Since Camel 4.19, a transacted exchange (JTA or Spring
transacted routes) that w
now committed again. Only the forced shutdown after the timeout marks the
in-flight transacted exchanges for rollback;
with `shutdownNowOnTimeout=false` there is no forced shutdown, so an exchange
completing after the timeout is committed.
+=== camel-kamelet - optional `advancedOptions` forwards extra URI parameters
+
+The kamelet component has an opt-in `advancedOptions` flag (default `false`)
on the component and the endpoint.
+When enabled, extra URI parameters that are not Kamelet template properties or
reserved kamelet options are forwarded
+to the wrapped source or sink endpoint. Existing routes that omit the flag are
unchanged.
Review Comment:
4.23.0 is already released, so this note can't go in the 4_23 guide. With
`advancedOptions` off by default nothing changes for existing users, and the
upgrade guide is for migration only, not for new features. So please remove
this section rather than move it to 4_24. The component doc already describes
the feature.
##########
components/camel-kamelet/src/main/docs/kamelet-component.adoc:
##########
@@ -163,6 +163,48 @@ The usage of this Kamelet into a Camel route is going to
be the same as any othe
WARNING: beware of any potential circular reference you may introduce when
using chain of Kamelets, in which case, the runtime will likely be idle
consuming a high amount of resources.
+=== Advanced component endpoint options
+
+Kamelets expose a defined set of parameters, but the underlying Camel
component may support additional endpoint options.
+You can forward those options by enabling `advancedOptions` (disabled by
default).
+
+When enabled, extra URI parameters that are not Kamelet template properties
and not reserved Kamelet options
+(`routeId`, `templateId`, `location`, `advancedOptions`, and the other kamelet
endpoint options) are appended to the
+endpoint wrapped by the Kamelet. No `source.` or `sink.` prefix is required.
Review Comment:
The `source.`/`sink.` prefixes never shipped, so a user reading this page
won't know what they are. Suggest dropping the sentence:
```suggestion
endpoint wrapped by the Kamelet.
```
##########
components/camel-kamelet/src/main/docs/kamelet-component.adoc:
##########
@@ -163,6 +163,48 @@ The usage of this Kamelet into a Camel route is going to
be the same as any othe
WARNING: beware of any potential circular reference you may introduce when
using chain of Kamelets, in which case, the runtime will likely be idle
consuming a high amount of resources.
+=== Advanced component endpoint options
+
+Kamelets expose a defined set of parameters, but the underlying Camel
component may support additional endpoint options.
+You can forward those options by enabling `advancedOptions` (disabled by
default).
+
+When enabled, extra URI parameters that are not Kamelet template properties
and not reserved Kamelet options
+(`routeId`, `templateId`, `location`, `advancedOptions`, and the other kamelet
endpoint options) are appended to the
+endpoint wrapped by the Kamelet. No `source.` or `sink.` prefix is required.
+
+Enable the feature on the component, or on a single endpoint. The endpoint
setting overrides the component setting.
+
+[source,properties]
+----
+camel.component.kamelet.advanced-options=true
+----
+
+[source,java]
+----
+from("kamelet:aws-s3-source?advancedOptions=true&bucket=x&pageSize=50")
+ .to("log:info");
+----
+
+In this example, `bucket` is a template property of the Kamelet and stays a
template parameter.
+`pageSize=50` is forwarded to the wrapped source endpoint. If that endpoint
URI already contains the same option,
Review Comment:
Matching text for the example above:
```suggestion
In this example, `bucketNameOrArn` is a template property of the Kamelet and
stays a template parameter.
`maxConnections=100` is forwarded to the wrapped `aws2-s3` endpoint. If that
endpoint URI already contains the same option,
```
##########
components/camel-kamelet/src/main/java/org/apache/camel/component/kamelet/Kamelet.java:
##########
@@ -184,6 +207,264 @@ public static void
extractKameletEnvironmentVariables(Map<String, Object> proper
}
}
+ /**
+ * Reads the original advanced options map stored under a single internal
key.
+ */
+ private static Map<String, Object> extractAdvancedOptions(Map<String,
Object> parameters, String key) {
+ Map<String, Object> options = new LinkedHashMap<>();
+ if (parameters == null) {
+ return options;
+ }
+
+ Object value = parameters.get(key);
+ if (value instanceof Map<?, ?> map) {
+ for (Map.Entry<?, ?> entry : map.entrySet()) {
+ if (entry.getKey() != null) {
+ options.put(entry.getKey().toString(), entry.getValue());
+ }
+ }
+ }
+
+ return options;
+ }
+
+ /**
+ * Drops Kamelet template properties from the advanced-option candidates,
treating camelCase and kebab-case names as
+ * aliases the same way {@code DefaultModel} does for route templates.
+ */
+ private static Map<String, Object> excludeTemplateParameters(
+ Map<String, Object> candidates, RouteTemplateDefinition template) {
+ if (candidates == null || candidates.isEmpty()) {
+ return new LinkedHashMap<>();
+ }
+
+ Set<String> templateNames = templateParameterNames(template);
+ if (templateNames.isEmpty()) {
+ return candidates;
+ }
+
+ Map<String, Object> filtered = new LinkedHashMap<>();
+ for (Map.Entry<String, Object> entry : candidates.entrySet()) {
+ String name = entry.getKey();
+ if (name != null && !isTemplateParameter(name, templateNames)) {
+ filtered.put(name, entry.getValue());
+ }
+ }
+ return filtered;
+ }
+
+ private static Set<String> templateParameterNames(RouteTemplateDefinition
template) {
+ Set<String> names = new HashSet<>();
+ if (template == null || template.getTemplateParameters() == null) {
+ return names;
+ }
+ for (RouteTemplateParameterDefinition param :
template.getTemplateParameters()) {
+ if (param.getName() != null) {
+ names.add(param.getName());
+ names.add(StringHelper.dashToCamelCase(param.getName()));
+ names.add(StringHelper.camelCaseToDash(param.getName()));
+ }
+ }
+ return names;
+ }
+
+ private static boolean isTemplateParameter(String name, Set<String>
templateNames) {
+ return templateNames.contains(name)
+ || templateNames.contains(StringHelper.dashToCamelCase(name))
+ || templateNames.contains(StringHelper.camelCaseToDash(name));
+ }
+
+ /**
+ * Preserves the user-supplied option name. When both camelCase and
kebab-case forms are present, the kebab-case
+ * form is kept so the same option is not appended twice.
+ */
+ private static Map<String, Object> normalizeParameterNames(Map<String,
Object> parameters) {
+ Map<String, Object> normalized = new LinkedHashMap<>();
+ Map<String, String> seenOriginal = new HashMap<>();
+ Map<String, Boolean> seenIsCanonical = 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);
+
+ String existingOriginal = seenOriginal.get(name);
+ if (existingOriginal == null) {
+ seenOriginal.put(name, originalName);
+ seenIsCanonical.put(name, isCanonical);
+ normalized.put(originalName, entry.getValue());
+ } else {
+ boolean existingIsCanonical = seenIsCanonical.get(name);
+ if (isCanonical && !existingIsCanonical) {
+ normalized.remove(existingOriginal);
+ seenOriginal.put(name, originalName);
+ seenIsCanonical.put(name, isCanonical);
+ normalized.put(originalName, entry.getValue());
+ }
+ }
+ }
+
+ 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;
+
+ // First remove all parameters that will be overridden.
+ for (String name : parameters.keySet()) {
+ result = removeQueryParameter(result, name);
+ }
+
+ // Then append the options to the preserved URI.
+ StringBuilder sb = new StringBuilder(result);
+ char sep = result.indexOf('?') < 0 ? '?' : '&';
+
+ for (Map.Entry<String, Object> entry : parameters.entrySet()) {
+ sb.append(sep).append(entry.getKey());
+ if (entry.getValue() != null) {
+
sb.append('=').append(wrapRawIfNeeded(String.valueOf(entry.getValue())));
+ }
+ sep = '&';
+ }
+
+ return sb.toString();
+ }
+
+ /**
+ * Re-wraps values that contain URI-sensitive characters. RAW wrappers may
already have been stripped by
+ * {@link URISupport#resolveRawParameterValues(Map)} before options reach
this method.
+ */
+ private static String wrapRawIfNeeded(String value) {
+ if (value == null || value.isEmpty() || isRawToken(value)) {
+ return value;
+ }
+ if (value.contains("&") || value.contains("=") || value.contains("?")
+ || value.contains("+") || value.contains("%")) {
+ if (value.contains(")&") && !value.startsWith("{") &&
!value.endsWith("}")
+ && !value.contains("}&")) {
+ return "RAW{" + value + "}";
+ }
+ if (value.contains(")&")) {
+ // fits in neither RAW(...) nor RAW{...}; keep the value out
of the message, it may be a secret
+ throw new IllegalArgumentException(
+ "The value of advanced options cannot contain \")&\"
when it also starts with '{',"
+ + " ends with '}' or
contains \"}&\"");
+ }
+ return "RAW(" + value + ")";
+ }
+
+ return value;
+ }
+
+ private static boolean isRawToken(String value) {
+ return value != null
+ && ((value.startsWith("RAW(") && value.endsWith(")"))
+ || (value.startsWith("RAW{") && value.endsWith("}")));
+ }
+
+ /**
+ * 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);
Review Comment:
Non-blocking. This runs for every advanced option, even when the option is
not in the URI, and it always rebuilds the whole query. `parseQuery`
URL-decodes the values that are not `RAW(...)`, and `createQueryString(params,
false)` writes them back without encoding. So a template URI with an encoded
value changes: `to("http://host/path?filter=a%26b";)` becomes `filter=a&b` and
is split into two parameters, and `%2B` becomes `+`.
None of the catalog Kamelets has a `%xx` or `+` in its endpoint URIs, so
this only affects custom Kamelets. Still, returning `uri` unchanged when no
parameter matches `normalizedName` would leave every URI that has nothing to
override as it was. The doc sentence about placeholders and RAW being preserved
would then hold for these URIs too.
##########
components/camel-kamelet/src/main/docs/kamelet-component.adoc:
##########
@@ -163,6 +163,48 @@ The usage of this Kamelet into a Camel route is going to
be the same as any othe
WARNING: beware of any potential circular reference you may introduce when
using chain of Kamelets, in which case, the runtime will likely be idle
consuming a high amount of resources.
+=== Advanced component endpoint options
+
+Kamelets expose a defined set of parameters, but the underlying Camel
component may support additional endpoint options.
+You can forward those options by enabling `advancedOptions` (disabled by
default).
+
+When enabled, extra URI parameters that are not Kamelet template properties
and not reserved Kamelet options
+(`routeId`, `templateId`, `location`, `advancedOptions`, and the other kamelet
endpoint options) are appended to the
+endpoint wrapped by the Kamelet. No `source.` or `sink.` prefix is required.
+
+Enable the feature on the component, or on a single endpoint. The endpoint
setting overrides the component setting.
+
+[source,properties]
+----
+camel.component.kamelet.advanced-options=true
+----
+
+[source,java]
+----
+from("kamelet:aws-s3-source?advancedOptions=true&bucket=x&pageSize=50")
Review Comment:
`aws-s3-source` has no `bucket` property (it is `bucketNameOrArn`), and
`aws2-s3` has no `pageSize` option. Both would be forwarded to the S3 endpoint
and startup would fail. My example on #27686 had the same mistake, sorry.
`maxConnections` is an `aws2-s3` option that the Kamelet does not expose:
```suggestion
from("kamelet:aws-s3-source?advancedOptions=true&bucketNameOrArn=x&maxConnections=100")
```
##########
components/camel-kamelet/src/test/java/org/apache/camel/component/kamelet/KameletAdvancedOptionsTest.java:
##########
@@ -0,0 +1,1352 @@
+/*
+ * 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.
+ */
+package org.apache.camel.component.kamelet;
+
+import java.net.URLDecoder;
+import java.nio.charset.StandardCharsets;
+import java.util.Map;
+import java.util.Properties;
+import java.util.regex.Pattern;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.http.HttpEndpoint;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.model.ModelCamelContext;
+import org.apache.camel.model.ProcessorDefinition;
+import org.apache.camel.model.RouteDefinition;
+import org.apache.camel.model.ToDefinition;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.apache.camel.util.URISupport;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class KameletAdvancedOptionsTest extends CamelTestSupport {
+
+ @Test
+ void advancedSourceOptionsAreApplied() {
+ RouteDefinition route = getRoute("sourceTrigger");
+ assertNotNull(route, "Expected the source caller route to be created");
+
+ String inputUri = findExpandedInputUri("direct:source?");
+ assertTrue(inputUri.contains("synchronous=true"),
+ () -> "Expected synchronous=true on the expanded source
endpoint, but was: " + inputUri);
+ }
+
+ @Test
+ void advancedSourceOptionsOverrideExistingUriParameters() {
+ RouteDefinition route = getRoute("sourceOverrideTrigger");
+ assertNotNull(route, "Expected the source override caller route to be
created");
+
+ String inputUri = findExpandedInputUri("direct:source-with-options?");
+ assertTrue(inputUri.contains("synchronous=true"),
+ () -> "Expected synchronous=true to override
synchronous=false: " + inputUri);
+ assertFalse(inputUri.contains("synchronous=false"),
+ () -> "The old synchronous=false value must not remain: " +
inputUri);
+ }
+
+ @Test
+ void advancedSourceOptionsPreservePropertyPlaceholders() throws Exception {
+ try (DefaultCamelContext testContext = new DefaultCamelContext()) {
+ testContext.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ routeTemplate("source-with-placeholder")
+
.from("direct:source?synchronous=false&timeout={{timeout}}")
+ .to("kamelet:sink");
+
+ from("direct:trigger")
+
.to("kamelet:source-with-placeholder/test?advancedOptions=true&synchronous=true");
+ }
+ });
+
+ Properties properties = new Properties();
+ properties.put("timeout", "1000");
+
testContext.getPropertiesComponent().setInitialProperties(properties);
+ testContext.start();
+
+ RouteDefinition expandedRoute = ((ModelCamelContext)
testContext).getRouteDefinitions().stream()
+ .filter(route -> Boolean.TRUE.equals(route.isKamelet()))
+ .filter(route -> route.getInput() != null)
+ .filter(route -> route.getInput().getEndpointUri() != null
+ &&
route.getInput().getEndpointUri().startsWith("direct:source?"))
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("Expanded source
route not found"));
+
+ String uri = expandedRoute.getInput().getEndpointUri();
+ assertTrue(uri.contains("synchronous=true"),
+ () -> "Expected advanced option, but was: " + uri);
+ assertTrue(uri.contains("timeout={{timeout}}"),
+ () -> "Expected the property placeholder to remain intact:
" + uri);
+ }
+ }
+
+ @Test
+ void advancedSourceOptionsPreserveRawExpressions() throws Exception {
+ try (DefaultCamelContext testContext = new DefaultCamelContext()) {
+ testContext.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ routeTemplate("source-with-raw")
+ .from("direct:source?timeout=RAW({{secret}})")
+ .to("kamelet:sink");
+
+ from("direct:trigger")
+
.to("kamelet:source-with-raw/test?advancedOptions=true&synchronous=true");
+ }
+ });
+
+ Properties properties = new Properties();
+ properties.put("secret", "1000");
+
testContext.getPropertiesComponent().setInitialProperties(properties);
+ testContext.start();
+
+ RouteDefinition expandedRoute = ((ModelCamelContext)
testContext).getRouteDefinitions().stream()
+ .filter(route -> Boolean.TRUE.equals(route.isKamelet()))
+ .filter(route -> route.getInput() != null)
+ .filter(route -> route.getInput().getEndpointUri() != null
+ &&
route.getInput().getEndpointUri().startsWith("direct:source?"))
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("Expanded source
route not found"));
+
+ String uri = expandedRoute.getInput().getEndpointUri();
+ assertTrue(uri.contains("synchronous=true"),
+ () -> "Expected advanced option, but was: " + uri);
+ assertTrue(uri.contains("timeout=RAW({{secret}})"),
+ () -> "Expected the RAW expression to remain intact: " +
uri);
+ }
+ }
+
+ @Test
+ void advancedSinkOptionsPreservePropertyPlaceholders() throws Exception {
+ try (DefaultCamelContext testContext = new DefaultCamelContext()) {
+ testContext.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ routeTemplate("sink-with-placeholder")
+ .from("kamelet:source")
+ .to("mock:{{topic}}?retain-first=0");
+
+ from("direct:trigger")
+
.kamelet("sink-with-placeholder?advancedOptions=true&retain-first=1");
+ }
+ });
+
+ Properties properties = new Properties();
+ properties.put("topic", "placeholder-sink");
+
testContext.getPropertiesComponent().setInitialProperties(properties);
+ testContext.start();
+
+ RouteDefinition expandedRoute = ((ModelCamelContext)
testContext).getRouteDefinitions().stream()
+ .filter(route -> Boolean.TRUE.equals(route.isKamelet()))
+ .filter(route -> findToDefinition(route, "mock:{{topic}}")
!= null)
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("Expanded sink route
not found"));
+
+ ToDefinition sink = findToDefinition(expandedRoute,
"mock:{{topic}}");
+ assertNotNull(sink, "Expected the placeholder sink endpoint");
+
+ String uri = sink.getEndpointUri();
+ assertTrue(uri.contains("{{topic}}"),
+ () -> "Expected the property placeholder to remain intact:
" + uri);
+ assertTrue(uri.contains("retain-first=1"),
+ () -> "Expected the advanced option on the sink endpoint:
" + uri);
+ assertFalse(uri.contains("%7B%7Btopic%7D%7D"),
+ () -> "The placeholder must not be URL-encoded: " + uri);
+ }
+ }
+
+ @Test
+ void advancedSinkOptionsPreserveRawExpressions() throws Exception {
+ try (DefaultCamelContext testContext = new DefaultCamelContext()) {
+ testContext.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ routeTemplate("sink-with-raw")
+ .from("kamelet:source")
+ .to("mock:{{bucket}}?secretKey=RAW({{secret}})");
+
+ from("direct:trigger")
+
.kamelet("sink-with-raw?advancedOptions=true&retain-first=1");
+ }
+ });
+
+ Properties properties = new Properties();
+ properties.put("bucket", "placeholder-bucket");
+ properties.put("secret", "secret-value");
+
testContext.getPropertiesComponent().setInitialProperties(properties);
+ testContext.start();
+
+ RouteDefinition expandedRoute = ((ModelCamelContext)
testContext).getRouteDefinitions().stream()
+ .filter(route -> Boolean.TRUE.equals(route.isKamelet()))
+ .filter(route -> route.getOutputs() != null)
+ .filter(route -> route.getOutputs().stream()
+ .anyMatch(output -> output instanceof ToDefinition
+ && ((ToDefinition)
output).getEndpointUri() != null
+ && ((ToDefinition)
output).getEndpointUri().startsWith("mock:{{bucket}}")))
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("Expanded RAW sink
route not found"));
+
+ ToDefinition sink = findToDefinition(expandedRoute,
"mock:{{bucket}}");
+ assertNotNull(sink, "Expected the RAW sink endpoint");
+
+ String uri = sink.getEndpointUri();
+ assertTrue(uri.contains("{{bucket}}"),
+ () -> "Expected the bucket placeholder to remain intact: "
+ uri);
+ assertTrue(uri.contains("secretKey=RAW({{secret}})"),
+ () -> "Expected the RAW expression to remain intact: " +
uri);
+ assertTrue(uri.contains("retain-first=1"),
+ () -> "Expected the advanced option on the sink endpoint:
" + uri);
+ assertFalse(uri.contains("%7B%7B"),
+ () -> "Property placeholders must not be URL-encoded: " +
uri);
+ }
+ }
+
+ @Test
+ void advancedSinkOptionsAreAppliedAndMessageIsDelivered() throws Exception
{
+ MockEndpoint sink = getMockEndpoint("mock:advanced-sink-result");
+ sink.expectedBodiesReceived("hello");
+
+ template.sendBody("direct:sink-trigger", "hello");
+
+ sink.assertIsSatisfied();
+ assertEquals(1, sink.getRetainFirst(),
+ "Expected retain-first=1 on the configured sink endpoint");
+ assertEquals(5, sink.getRetainLast(),
+ "Expected retain-last=5 to be preserved on the configured sink
endpoint");
+ }
+
+ @Test
+ void noAdvancedOptionsRetainsExistingBehavior() throws Exception {
+ MockEndpoint sink = getMockEndpoint("mock:basic-sink-result");
+ sink.expectedBodiesReceived("hello");
+
+ template.sendBody("direct:basic-trigger", "hello");
+
+ sink.assertIsSatisfied();
+ assertEquals(-1, sink.getRetainFirst(),
+ "The basic sink should retain the default retainFirst value");
+ assertEquals(-1, sink.getRetainLast(),
+ "The basic sink should retain the default retainLast value");
+ }
+
+ @Test
+ void extraParametersWithoutAdvancedOptionsAreNotForwarded() throws
Exception {
+ try (DefaultCamelContext testContext = new DefaultCamelContext()) {
+ testContext.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ routeTemplate("sink-without-flag")
+ .from("kamelet:source")
+ .to("mock:no-flag-sink?retain-first=0");
+
+ from("direct:trigger")
+ .kamelet("sink-without-flag?retain-first=1");
+ }
+ });
+
+ testContext.start();
+
+ ToDefinition sink = findToDefinition(
+ findExpandedRouteWithOutput(testContext,
"mock:no-flag-sink"), "mock:no-flag-sink");
+ assertNotNull(sink, "Expected sink endpoint");
+ String uri = sink.getEndpointUri();
+ assertTrue(uri.contains("retain-first=0"),
+ () -> "Without advancedOptions the original value must
remain: " + uri);
+ assertFalse(uri.contains("retain-first=1"),
+ () -> "Without advancedOptions extra URI parameters must
not be forwarded: " + uri);
+ }
+ }
+
+ @Test
+ void componentAdvancedOptionsForwardsWithoutEndpointFlag() throws
Exception {
+ try (DefaultCamelContext testContext = new DefaultCamelContext()) {
+ testContext.getComponent("kamelet",
KameletComponent.class).setAdvancedOptions(true);
+ testContext.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ routeTemplate("component-sink")
+ .from("kamelet:source")
+ .to("mock:component-sink?retain-first=0");
+
+ from("direct:trigger")
+ .kamelet("component-sink?retain-first=1");
+ }
+ });
+
+ testContext.start();
+
+ ToDefinition sink = findToDefinition(
+ findExpandedRouteWithOutput(testContext,
"mock:component-sink"), "mock:component-sink");
+ assertNotNull(sink, "Expected sink endpoint");
+ assertTrue(sink.getEndpointUri().contains("retain-first=1"),
+ () -> "Component advancedOptions=true should forward extra
parameters: "
+ + sink.getEndpointUri());
+ }
+ }
+
+ @Test
+ void endpointAdvancedOptionsFalseOverridesComponent() throws Exception {
+ try (DefaultCamelContext testContext = new DefaultCamelContext()) {
+ testContext.getComponent("kamelet",
KameletComponent.class).setAdvancedOptions(true);
+ testContext.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ routeTemplate("endpoint-override-sink")
+ .from("kamelet:source")
+ .to("mock:endpoint-override-sink?retain-first=0");
+
+ from("direct:trigger")
+
.kamelet("endpoint-override-sink?advancedOptions=false&retain-first=1");
+ }
+ });
+
+ testContext.start();
+
+ ToDefinition sink = findToDefinition(
+ findExpandedRouteWithOutput(testContext,
"mock:endpoint-override-sink"),
+ "mock:endpoint-override-sink");
+ assertNotNull(sink, "Expected sink endpoint");
+ String uri = sink.getEndpointUri();
+ assertTrue(uri.contains("retain-first=0"),
+ () -> "Endpoint advancedOptions=false must keep the
original value: " + uri);
+ assertFalse(uri.contains("retain-first=1"),
+ () -> "Endpoint advancedOptions=false must override the
component: " + uri);
+ }
+ }
+
+ @Test
+ void templatePropertiesAreNotForwardedOnSource() throws Exception {
+ try (DefaultCamelContext testContext = new DefaultCamelContext()) {
+ testContext.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ routeTemplate("source-with-param")
+ .templateParameter("name")
+ .from("direct:{{name}}?synchronous=false")
+ .to("kamelet:sink");
+
+ from("direct:trigger")
+
.to("kamelet:source-with-param?advancedOptions=true&name=foo&synchronous=true");
+ }
+ });
+
+ testContext.start();
+
+ String uri = ((ModelCamelContext)
testContext).getRouteDefinitions().stream()
+ .filter(route -> Boolean.TRUE.equals(route.isKamelet()))
+ .filter(route -> route.getInput() != null)
+ .map(route -> route.getInput().getEndpointUri())
+ .filter(endpointUri -> endpointUri != null &&
endpointUri.contains("synchronous=true"))
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("Expanded source
route not found"));
+
+ assertTrue(uri.startsWith("direct:foo") ||
uri.startsWith("direct:{{name}}"),
+ () -> "Expected the template property to be used as the
source path: " + uri);
+ assertFalse(uri.contains("name="),
+ () -> "Template property name must not be forwarded as a
query parameter: " + uri);
+ }
+ }
+
+ @Test
+ void templatePropertiesAreNotForwardedOnSink() throws Exception {
+ try (DefaultCamelContext testContext = new DefaultCamelContext()) {
+ testContext.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ routeTemplate("sink-with-param")
+ .templateParameter("topic")
+ .from("kamelet:source")
+ .to("mock:{{topic}}?retain-first=0");
+
+ from("direct:trigger")
+
.kamelet("sink-with-param?advancedOptions=true&topic=out&retain-first=1");
+ }
+ });
+
+ testContext.start();
+
+ ToDefinition sink = findToDefinition(
+ findExpandedRouteWithOutput(testContext,
"mock:{{topic}}"), "mock:{{topic}}");
+ assertNotNull(sink, "Expected sink endpoint");
+ String uri = sink.getEndpointUri();
+ assertTrue(uri.contains("retain-first=1"),
+ () -> "Expected the advanced option on the sink endpoint:
" + uri);
+ Map<String, Object> query = parseUriQuery(uri);
+ assertFalse(query.containsKey("topic"),
+ () -> "Template property topic must not be forwarded as a
query parameter: " + uri);
+ }
+ }
+
+ @Test
+ void reservedOptionsAndSwitchAreNotForwarded() throws Exception {
+ try (DefaultCamelContext testContext = new DefaultCamelContext()) {
+ testContext.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ routeTemplate("reserved-sink")
+ .from("kamelet:source")
+ .to("mock:reserved-sink?retain-first=0");
+
+ from("direct:trigger")
+
.kamelet("reserved-sink?advancedOptions=true&retain-first=1&location=classpath:kamelets");
+ }
+ });
+
+ testContext.start();
+
+ ToDefinition sink = findToDefinition(
+ findExpandedRouteWithOutput(testContext,
"mock:reserved-sink"), "mock:reserved-sink");
+ assertNotNull(sink, "Expected sink endpoint");
+ String uri = sink.getEndpointUri();
+ assertTrue(uri.contains("retain-first=1"),
+ () -> "Expected the advanced option on the sink endpoint:
" + uri);
+ assertFalse(uri.contains("advancedOptions="),
+ () -> "The advancedOptions switch must not be forwarded: "
+ uri);
+ assertFalse(uri.contains("location="),
+ () -> "Reserved option location must not be forwarded: " +
uri);
+ }
+ }
+
+ @Test
+ void actionWithoutWrappedEndpointFailsWhenAdvancedOptionsSupplied() throws
Exception {
+ try (DefaultCamelContext testContext = new DefaultCamelContext()) {
+ testContext.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ routeTemplate("set-body-action")
+ .from("kamelet:source")
+ .setBody().constant("x");
+
+ from("direct:trigger")
+
.kamelet("set-body-action?advancedOptions=true&retain-first=1");
+ }
+ });
+
+ Exception exception = assertThrows(Exception.class,
testContext::start);
+ assertTrue(hasIllegalArgumentException(exception),
+ () -> "Expected an IllegalArgumentException for an action
without a wrapped endpoint, but got: "
+ + exception);
+ assertTrue(exceptionChainContains(exception, "no top-level static
to() destination"),
+ () -> "Expected a clear no-wrapped-endpoint error, but
got: " + exception);
+ }
+ }
+
+ @Test
+ void sourceKameletForwardsUnprefixedOptions() throws Exception {
+ try (CamelContext testContext = new DefaultCamelContext()) {
+ testContext.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ routeTemplate("source-with-unprefixed-options")
+ .from("direct:source")
+ .to("kamelet:sink");
+
+ from("direct:trigger")
+
.to("kamelet:source-with-unprefixed-options/test?advancedOptions=true&synchronous=true");
+ }
+ });
+
+ testContext.start();
+
+ RouteDefinition expandedRoute = ((ModelCamelContext)
testContext).getRouteDefinitions().stream()
+ .filter(route -> Boolean.TRUE.equals(route.isKamelet()))
+ .filter(route -> route.getInput() != null)
+ .filter(route -> route.getInput().getEndpointUri() != null
+ &&
route.getInput().getEndpointUri().startsWith("direct:source?"))
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("Expanded source
route not found"));
+
+ String uri = expandedRoute.getInput().getEndpointUri();
+ assertTrue(uri.contains("synchronous=true"),
+ () -> "Expected advanced option on source endpoint: " +
uri);
+ }
+ }
+
+ @Test
+ void sinkKameletForwardsUnprefixedOptions() throws Exception {
+ try (CamelContext testContext = new DefaultCamelContext()) {
+ testContext.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ routeTemplate("sink-with-unprefixed-options")
+ .from("kamelet:source")
+ .to("mock:result");
+
+ from("direct:trigger")
+
.kamelet("sink-with-unprefixed-options?advancedOptions=true&retain-first=1");
+ }
+ });
+
+ testContext.start();
+
+ RouteDefinition expandedRoute = ((ModelCamelContext)
testContext).getRouteDefinitions().stream()
+ .filter(route -> Boolean.TRUE.equals(route.isKamelet()))
+ .filter(route -> findToDefinition(route, "mock:result") !=
null)
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("Expanded sink route
not found"));
+
+ ToDefinition sink = findToDefinition(expandedRoute, "mock:result");
+ assertNotNull(sink, "Expected the sink endpoint");
+
+ String uri = sink.getEndpointUri();
+ assertTrue(uri.contains("retain-first=1"),
+ () -> "Expected advanced option on sink endpoint: " + uri);
+ }
+ }
+
+ @Test
+ void advancedSinkOptionsRejectMultipleTopLevelStaticDestinations() throws
Exception {
+ try (DefaultCamelContext testContext = new DefaultCamelContext()) {
+ testContext.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ routeTemplate("mock-multi-sink")
+ .from("kamelet:source")
+ .to("mock:first-advanced-sink?retain-first=0")
+ .to("mock:second-advanced-sink?retain-first=0");
+
+ from("direct:multi-sink-trigger")
+
.kamelet("mock-multi-sink?advancedOptions=true&retain-first=1");
+ }
+ });
+
+ Exception exception = assertThrows(Exception.class,
testContext::start);
+
+ assertTrue(hasIllegalArgumentException(exception),
+ () -> "Expected an IllegalArgumentException for multiple
destinations, but got: " + exception);
+ assertTrue(exceptionChainContains(exception, "multiple top-level
static to() destinations"),
+ () -> "Expected a clear ambiguity error, but got: " +
exception);
+ }
+ }
+
+ @Test
+ void advancedSinkOptionsRejectNestedChoiceDestinations() throws Exception {
+ try (DefaultCamelContext testContext = new DefaultCamelContext()) {
+ testContext.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ routeTemplate("choice-sink")
+ .from("kamelet:source")
+ .choice()
+ .when(header("selected").isEqualTo(true))
+ .to("mock:choice-selected?retain-first=0")
+ .otherwise()
+ .to("mock:choice-default?retain-first=0")
+ .end();
+
+ from("direct:trigger")
+
.kamelet("choice-sink?advancedOptions=true&retain-first=1");
+ }
+ });
+
+ Exception exception = assertThrows(Exception.class,
testContext::start);
+
+ assertTrue(hasIllegalArgumentException(exception),
+ () -> "Expected an IllegalArgumentException for nested
destinations, but got: " + exception);
+ assertTrue(exceptionChainContains(exception, "no top-level static
to() destination"),
+ () -> "Expected a clear no-static-sink error, but got: " +
exception);
+ }
+ }
+
+ @Test
+ void sinkOptionsNormalizeDuplicateCamelCaseAndKebabCaseNames() throws
Exception {
+ try (DefaultCamelContext testContext = new DefaultCamelContext()) {
+ testContext.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ routeTemplate("sink-with-duplicate-options")
+ .from("kamelet:source")
+ .to("mock:duplicate-options?retain-first=0");
+
+ from("direct:trigger")
+ .kamelet(
+
"sink-with-duplicate-options?advancedOptions=true&retainFirst=1&retain-first=2");
+ }
+ });
+
+ testContext.start();
+
+ RouteDefinition expandedRoute = findExpandedRouteWithOutput(
+ testContext, "mock:duplicate-options");
+ ToDefinition sink = findToDefinition(expandedRoute,
"mock:duplicate-options");
+
+ assertNotNull(sink, "Expected duplicate-options sink endpoint");
+
+ String uri = sink.getEndpointUri();
+ assertEquals(1, countOccurrences(uri, "retain-first="),
+ () -> "Expected only one retain-first option: " + uri);
+ assertTrue(uri.matches(".*retain-first=[^&]+.*"),
+ () -> "Expected a canonical retain-first option: " + uri);
+ assertFalse(uri.contains("retainFirst="),
+ () -> "The camelCase alias should not remain in the URI: "
+ uri);
+ }
+ }
+
+ @Test
+ void testSinkOptionsWithDynamicOutputOnly() throws Exception {
+ try (CamelContext testContext = new DefaultCamelContext()) {
+ testContext.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ routeTemplate("dynamic-sink")
+ .from("kamelet:source")
+ .toD("log:${header.destination}");
+
+ from("direct:trigger")
+
.kamelet("dynamic-sink?advancedOptions=true&pageSize=10");
+ }
+ });
+
+ Exception exception = assertThrows(Exception.class,
testContext::start);
+
+ assertTrue(hasIllegalArgumentException(exception),
+ () -> "Expected an IllegalArgumentException, but got: " +
exception);
+
+ assertTrue(
+ exceptionChainContains(exception,
+ "advancedOptions cannot be applied"),
+ () -> "Expected a clear explanation for dynamic endpoints,
but got: " + exception);
+ }
+ }
+
+ @Test
+ void testSinkOptionsWithFixedAndDynamicOutputs() throws Exception {
+ try (CamelContext testContext = new DefaultCamelContext()) {
+ testContext.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ routeTemplate("mixed-sink")
+ .from("kamelet:source")
+ .to("mock:fixed-sink?retain-first=0")
+ .toD("mock:${header.destination}");
+
+ from("direct:trigger")
+
.kamelet("mixed-sink?advancedOptions=true&retain-first=1");
+ }
+ });
+
+ Exception exception = assertThrows(Exception.class,
testContext::start);
+
+ assertTrue(hasIllegalArgumentException(exception),
+ () -> "Expected an IllegalArgumentException for mixed
static/dynamic destinations, but got: "
+ + exception);
+ assertTrue(exceptionChainContains(exception, "contains a dynamic
destination"),
+ () -> "Expected a clear dynamic-destination error, but
got: " + exception);
+ }
+ }
+
+ @Test
+ void testSinkOptionsWithEnrichOnly() throws Exception {
+ assertDynamicSinkOptionsRejected("enrich", "enrich");
+ }
+
+ @Test
+ void testSinkOptionsWithPollEnrichOnly() throws Exception {
+ assertDynamicSinkOptionsRejected("pollEnrich", "pollEnrich");
+ }
+
+ @Test
+ void wireTapIsNotTreatedAsTheMainSinkDestination() throws Exception {
+ try (DefaultCamelContext testContext = new DefaultCamelContext()) {
+ testContext.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ routeTemplate("wiretap-sink")
+ .from("kamelet:source")
+ .wireTap("mock:side-effect")
+ .to("mock:main-sink?retain-first=0");
+
+ from("direct:trigger")
+
.kamelet("wiretap-sink?advancedOptions=true&retain-first=1");
+ }
+ });
+
+ testContext.start();
+
+ RouteDefinition expandedRoute =
findExpandedRouteWithOutput(testContext, "mock:main-sink");
+ ToDefinition mainSink = findToDefinition(expandedRoute,
"mock:main-sink");
+
+ assertNotNull(mainSink, "Expected main sink endpoint");
+ assertTrue(mainSink.getEndpointUri().contains("retain-first=1"),
+ () -> "Expected advanced option on main sink: " +
mainSink.getEndpointUri());
+
+ boolean wireTapHasAdvancedOption =
expandedRoute.getOutputs().stream()
+ .filter(output ->
output.getClass().getSimpleName().contains("WireTap"))
+ .anyMatch(output ->
output.toString().contains("retain-first=1"));
+
+ assertFalse(wireTapHasAdvancedOption,
+ "The wireTap destination should not receive sink options");
+ }
+ }
+
+ @Test
+ void testDynamicOutputWithoutSinkOptions() throws Exception {
+ try (CamelContext testContext = new DefaultCamelContext()) {
+ testContext.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ routeTemplate("dynamic-sink")
+ .from("kamelet:source")
+ .toD("mock:${header.destination}");
+
+ from("direct:trigger")
+ .kamelet("dynamic-sink");
+ }
+ });
+
+ testContext.start();
+
+ MockEndpoint dynamicSink =
testContext.getEndpoint("mock:dynamic-sink", MockEndpoint.class);
+ dynamicSink.expectedBodiesReceived("hello");
+
+ testContext.createProducerTemplate().sendBodyAndHeader(
+ "direct:trigger", "hello", "destination", "dynamic-sink");
+
+ dynamicSink.assertIsSatisfied();
+ }
+ }
+
+ @Test
+ void routeScopedPropertyDoesNotClashWithAdvancedOptions() throws Exception
{
+ try (DefaultCamelContext testContext = new DefaultCamelContext()) {
+ testContext.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ routeTemplate("sink-template")
+ .from("kamelet:source")
+ .to("mock:result?retain-first=0");
+
+ from("direct:trigger")
+
.kamelet("sink-template/testRoute?advancedOptions=true&retain-first=1");
+ }
+ });
+
+ Properties properties = new Properties();
+ // This is a route-scoped property that should NOT be treated as
an advanced option
+
properties.put("camel.kamelet.sink-template.testRoute.source.retainFirst",
"999");
+
testContext.getPropertiesComponent().setInitialProperties(properties);
+ testContext.start();
+
+ RouteDefinition expandedRoute =
findExpandedRouteWithOutput(testContext, "mock:result");
+ ToDefinition sink = findToDefinition(expandedRoute, "mock:result");
+
+ assertNotNull(sink, "Expected sink endpoint");
+ String uri = sink.getEndpointUri();
+ assertTrue(uri.contains("retain-first=1"),
+ () -> "Expected URI parameter retain-first=1 from advanced
option: " + uri);
+ assertFalse(uri.contains("retain-first=999"),
+ () -> "Route-scoped property should not be treated as
advanced option: " + uri);
+ }
+ }
+
+ @Test
+ void routeScopedPropertyCannotInjectInternalPrefix() throws Exception {
+ try (DefaultCamelContext testContext = new DefaultCamelContext()) {
+ testContext.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ routeTemplate("sink-template")
+ .from("kamelet:source")
+ .to("mock:result?retain-first=0");
+
+ from("direct:trigger")
+ .kamelet("sink-template/testRoute");
+ }
+ });
+
+ Properties properties = new Properties();
+ // Attempt to inject via route-scoped property - should NOT work
+
properties.put("camel.kamelet.sink-template.testRoute.internal.advanced.options.retain-first",
"999");
+
testContext.getPropertiesComponent().setInitialProperties(properties);
+ testContext.start();
+
+ RouteDefinition expandedRoute =
findExpandedRouteWithOutput(testContext, "mock:result");
+ ToDefinition sink = findToDefinition(expandedRoute, "mock:result");
+
+ assertNotNull(sink, "Expected sink endpoint");
+ String uri = sink.getEndpointUri();
+ // The internal prefix should be ignored when coming from
route-scoped properties
+ assertFalse(uri.contains("retain-first=999"),
+ () -> "Route-scoped property with internal prefix should
be ignored: " + uri);
+ assertTrue(uri.contains("retain-first=0"),
+ () -> "Original value should remain: " + uri);
+ }
+ }
+
+ @Test
+ void removeQueryParameterPreservesRawAmpersands() throws Exception {
+ String uri =
"kamelet:my-source?param=RAW(value&with&ersands)&other=123";
+
+ var method = Kamelet.class.getDeclaredMethod(
+ "removeQueryParameter", String.class, String.class);
+ method.setAccessible(true);
Review Comment:
Non-blocking. Instead of calling the private method through reflection,
`removeQueryParameter` could be package-private (the test is in the same
package), and the test calls it directly. Then a rename shows up at compile
time, not as a `NoSuchMethodException`.
--
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]
