davsclaus commented on code in PR #27686:
URL: https://github.com/apache/camel/pull/27686#discussion_r4266926744
##########
components/camel-kamelet/src/main/java/org/apache/camel/component/kamelet/Kamelet.java:
##########
@@ -16,31 +16,39 @@
*/
package org.apache.camel.component.kamelet;
+import java.net.URISyntaxException;
import java.util.Collection;
import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Properties;
import java.util.function.Predicate;
import org.apache.camel.CamelContext;
import org.apache.camel.builder.NoErrorHandlerBuilder;
+import org.apache.camel.model.EnrichDefinition;
import org.apache.camel.model.ModelCamelContext;
+import org.apache.camel.model.PollEnrichDefinition;
import org.apache.camel.model.ProcessorDefinition;
import org.apache.camel.model.ProcessorDefinitionHelper;
import org.apache.camel.model.RouteDefinition;
import org.apache.camel.model.RouteTemplateDefinition;
import org.apache.camel.model.ToDefinition;
+import org.apache.camel.model.ToDynamicDefinition;
+import org.apache.camel.model.WireTapDefinition;
import org.apache.camel.spi.PropertiesComponent;
import org.apache.camel.spi.UuidGenerator;
import org.apache.camel.support.CamelContextHelper;
import org.apache.camel.support.SimpleUuidGenerator;
import org.apache.camel.util.IOHelper;
import org.apache.camel.util.ObjectHelper;
import org.apache.camel.util.StringHelper;
+import org.apache.camel.util.URISupport;
import static
org.apache.camel.model.ProcessorDefinitionHelper.filterTypeInOutputs;
public final class Kamelet {
+
Review Comment:
This blank line is not on `main` and is unrelated to the feature. Please
remove it.
```suggestion
```
##########
components/camel-kamelet/src/main/java/org/apache/camel/component/kamelet/Kamelet.java:
##########
@@ -184,6 +206,215 @@ 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;
+ }
+
+ /**
+ * 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.
+ * <p>
+ * RAW(...) and RAW{...} have limitations when their closing delimiter is
followed by an ampersand. Avoid RAW{...}
+ * for values starting with '{' or ending with '}', as they may be
interpreted as property placeholders.
+ */
+ 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 + "}";
+ }
+ return "RAW(" + value + ")";
Review Comment:
Here the value has `)&` and also starts with `{`, ends with `}` or contains
`}&`, so neither `RAW(...)` nor `RAW{...}` can hold it. `RAW(...)` cuts it at
`)&` and the rest becomes an extra parameter (the case in the `@Disabled`
test). Failing at startup is clearer than a cut value. The message should not
contain the value, as it may be a secret:
```suggestion
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 a source.* or sink.* option cannot
contain \")&\" when it also starts with '{',"
+ " ends with '}' or
contains \"}&\"");
}
return "RAW(" + value + ")";
```
`appendParametersToURI` is called inside the `try`/`catch` that wraps the
error in "Failed to apply advanced sink options ...", so the user also sees
which endpoint it is.
`sinkOptionsPreserveJsonRawValueWithClosingParenthesisAndAmpersand` can then
drop `@Disabled` and assert this error with `assertThrows` and
`exceptionChainContains`, like the other rejection tests.
##########
components/camel-kamelet/src/test/java/org/apache/camel/component/kamelet/KameletAdvancedOptionsTest.java:
##########
@@ -0,0 +1,1144 @@
+/*
+ * 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.Disabled;
+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("block=true"),
+ () -> "Expected block=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("block=true"),
+ () -> "Expected block=true to override block=false: " +
inputUri);
+ assertFalse(inputUri.contains("block=false"),
+ () -> "The old block=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?block=false&timeout={{timeout}}")
+ .to("kamelet:sink");
+
+ from("direct:trigger")
+
.to("kamelet:source-with-placeholder/test?source.block=true");
+ }
+ });
+
+ Properties properties = new Properties();
+ properties.put("timeout", "1000");
+
testContext.getPropertiesComponent().setInitialProperties(properties);
+ testContext.start();
+
+ RouteDefinition expandedRoute =
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("block=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?source.block=true");
+ }
+ });
+
+ Properties properties = new Properties();
+ properties.put("secret", "1000");
+
testContext.getPropertiesComponent().setInitialProperties(properties);
+ testContext.start();
+
+ RouteDefinition expandedRoute =
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("block=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?sink.retain-first=1");
+ }
+ });
+
+ Properties properties = new Properties();
+ properties.put("topic", "placeholder-sink");
+
testContext.getPropertiesComponent().setInitialProperties(properties);
+ testContext.start();
+
+ RouteDefinition expandedRoute =
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?sink.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 =
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 sourceKameletRejectsSinkOptions() throws Exception {
+ try (CamelContext testContext = new DefaultCamelContext()) {
+ testContext.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ routeTemplate("source-with-invalid-options")
+ .from("direct:source")
+ .to("kamelet:sink");
+
+ from("direct:trigger")
+
.to("kamelet:source-with-invalid-options/test?sink.block=true");
+ }
+ });
+
+ Exception exception = assertThrows(Exception.class,
testContext::start);
+
+ assertTrue(hasIllegalArgumentException(exception),
+ () -> "Expected an IllegalArgumentException for sink.* on
a source Kamelet, but got: "
+ + exception);
+ }
+ }
+
+ @Test
+ void sinkKameletRejectsSourceOptions() throws Exception {
+ try (CamelContext testContext = new DefaultCamelContext()) {
+ testContext.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ routeTemplate("sink-with-invalid-options")
+ .from("kamelet:source")
+ .to("mock:result");
+
+ from("direct:trigger")
+
.kamelet("sink-with-invalid-options?source.block=true");
+ }
+ });
+
+ Exception ex = assertThrows(Exception.class, testContext::start);
+ assertTrue(hasIllegalArgumentException(ex),
+ "Expected IllegalArgumentException for source.* on a sink
Kamelet");
+ }
+ }
+
+ @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?sink.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?sink.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?sink.retainFirst=1&sink.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?sink.block=true");
+ }
+ });
+
+ Exception exception = assertThrows(Exception.class,
testContext::start);
+
+ assertTrue(hasIllegalArgumentException(exception),
+ () -> "Expected an IllegalArgumentException, but got: " +
exception);
+
+ assertTrue(
+ exceptionChainContains(exception,
+ "sink.* advanced options 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?sink.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?sink.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?sink.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 sink.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.sink.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);
+
+ String result = (String) method.invoke(null, uri, "other");
+
+ assertEquals(
+ "kamelet:my-source?param=RAW(value&with&ersands)",
+ result);
+ }
+
+ @Test
+ void sinkOptionsRejectAmbiguousLogAndStaticDestinations() throws Exception
{
+ try (DefaultCamelContext testContext = new DefaultCamelContext()) {
+ testContext.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ routeTemplate("sink-with-log")
+ .from("kamelet:source")
+ .to("log:before-sink")
+ .to("mock:actual-sink?retain-first=0");
+
+ from("direct:trigger")
+ .kamelet("sink-with-log?sink.retain-first=1");
+ }
+ });
+
+ Exception exception = assertThrows(Exception.class,
testContext::start);
+
+ assertTrue(hasIllegalArgumentException(exception),
+ () -> "Expected an IllegalArgumentException for multiple
static destinations, but got: "
+ + exception);
+ assertTrue(exceptionChainContains(exception, "multiple top-level
static to() destinations"),
+ () -> "Expected a clear ambiguity error, but got: " +
exception);
+ }
+ }
+
+ @Test
+ void sinkOptionsPreserveOriginalCamelCaseSpelling() throws Exception {
+ try (DefaultCamelContext testContext = new DefaultCamelContext()) {
+ testContext.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ routeTemplate("spelling-sink")
+ .from("kamelet:source")
+ .to("http://localhost/unused-page-size";);
+
+ from("direct:trigger")
+ .kamelet("spelling-sink?sink.pageSize=10");
+ }
+ });
+
+ testContext.start();
+
+ RouteDefinition expandedRoute =
findExpandedRouteWithOutput(testContext, "http://localhost/unused-page-size";);
+ ToDefinition sink = findToDefinition(expandedRoute,
"http://localhost/unused-page-size";);
+
+ assertNotNull(sink, "Expected sink endpoint");
+ String uri = sink.getEndpointUri();
+ assertTrue(uri.contains("pageSize=10"),
+ () -> "Expected original camelCase spelling to be
preserved: " + uri);
+ assertFalse(uri.contains("page-size="),
+ () -> "Parameter name must not be rewritten to kebab-case:
" + uri);
+
+ Map<String, Object> query = parseUriQuery(uri);
+ assertEquals("10", String.valueOf(query.get("pageSize")));
+ assertFalse(query.containsKey("page-size"),
+ () -> "Kebab-case alias must not be forwarded: " + query);
+ }
+ }
+
+ @Test
+ void sinkOptionsPreserveRawValuesWithUriSensitiveCharacters() throws
Exception {
+ try (DefaultCamelContext testContext = new DefaultCamelContext()) {
+ testContext.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ routeTemplate("password-sink")
+ .from("kamelet:source")
+ .to("http://localhost/unused-password";);
+
+ from("direct:trigger")
+
.kamelet("password-sink?sink.password=RAW(p@ss&word)");
+ }
+ });
+
+ testContext.start();
+
+ RouteDefinition expandedRoute =
findExpandedRouteWithOutput(testContext, "http://localhost/unused-password";);
+ ToDefinition sink = findToDefinition(expandedRoute,
"http://localhost/unused-password";);
+
+ assertNotNull(sink, "Expected sink endpoint");
+ String uri = sink.getEndpointUri();
+ Map<String, Object> query = parseUriQuery(uri);
+
+ assertEquals("p@ss&word", rawOrPlain(query.get("password")),
+ () -> "Expected the password value to remain intact: " +
uri);
+ assertFalse(query.containsKey("word"),
+ () -> "Ampersand in the password must not create a
spurious parameter: " + uri);
+
+ HttpEndpoint http = findHttpEndpoint(testContext,
"/unused-password");
+ assertNotNull(http, "Expected the HTTP sink endpoint to be
created");
+ String rawQuery = http.getHttpUri().getRawQuery();
+ assertNotNull(rawQuery, "Expected HTTP URI query parameters");
+ assertFalse(rawQuery.contains("&word"),
+ () -> "The HTTP endpoint must not receive a spurious word
parameter: " + http.getHttpUri());
+ Map<String, Object> httpQuery = URISupport.parseQuery(rawQuery);
+ assertEquals("p@ss&word",
urlDecode(rawOrPlain(httpQuery.get("password"))),
+ () -> "Expected the HTTP endpoint to receive the password
as one value: " + http.getHttpUri());
+ assertFalse(httpQuery.containsKey("word"),
+ () -> "The HTTP endpoint must not receive a spurious word
parameter: " + http.getHttpUri());
+ }
+ }
+
+ @Test
+ void advancedSinkOptionsSkipNestedKameletWhenSelectingMainSink() throws
Exception {
+ try (DefaultCamelContext testContext = new DefaultCamelContext()) {
+ testContext.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ routeTemplate("inner-sink")
+ .from("kamelet:source")
+ .to("mock:inner-sink");
+
+ routeTemplate("nested-and-main")
+ .from("kamelet:source")
+ .to("kamelet:inner-sink")
+ .to("mock:main-sink?retain-first=0");
+
+ from("direct:trigger")
+ .kamelet("nested-and-main?sink.retain-first=1");
+ }
+ });
+
+ testContext.start();
+
+ RouteDefinition expandedRoute =
findExpandedRouteWithOutput(testContext, "mock:main-sink");
+ ToDefinition mainSink = findToDefinition(expandedRoute,
"mock:main-sink");
+ ToDefinition nested = findToDefinition(expandedRoute,
"kamelet:inner-sink");
+
+ assertNotNull(mainSink, "Expected main sink endpoint");
+ assertTrue(mainSink.getEndpointUri().contains("retain-first=1"),
+ () -> "Expected advanced option on the non-Kamelet sink: "
+ mainSink.getEndpointUri());
+
+ assertNotNull(nested, "Expected nested Kamelet destination");
+ assertFalse(nested.getEndpointUri().contains("retain-first="),
+ () -> "Nested kamelet: destination must not receive sink
options: " + nested.getEndpointUri());
+ }
+ }
+
+ @Test
+ void advancedSinkOptionsRejectNestedKameletOnlyDestinations() throws
Exception {
+ try (DefaultCamelContext testContext = new DefaultCamelContext()) {
+ testContext.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ routeTemplate("inner-sink")
+ .from("kamelet:source")
+ .to("mock:inner-sink");
+
+ routeTemplate("nested-kamelet-sink")
+ .from("kamelet:source")
+ .to("kamelet:inner-sink");
+
+ from("direct:trigger")
+
.kamelet("nested-kamelet-sink?sink.retain-first=1");
+ }
+ });
+
+ Exception exception = assertThrows(Exception.class,
testContext::start);
+ assertTrue(hasIllegalArgumentException(exception),
+ () -> "Expected an IllegalArgumentException, but got: " +
exception);
+ assertTrue(exceptionChainContains(exception, "no top-level static
to() destination"),
+ () -> "Expected no-static-sink error, but got: " +
exception);
+ }
+ }
+
+ @Test
+ void sinkOptionsPreservePercentEncodedSecretThroughRouteTemplate() throws
Exception {
+ try (DefaultCamelContext testContext = new DefaultCamelContext()) {
+ testContext.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ routeTemplate("percent-password-sink")
+ .from("kamelet:source")
+ .to("http://localhost/unused-percent-secret";);
+
+ from("direct:trigger")
+
.kamelet("percent-password-sink?sink.password=RAW(se+cret%41)");
+ }
+ });
+
+ testContext.start();
+
+ RouteDefinition expandedRoute = findExpandedRouteWithOutput(
+ testContext, "http://localhost/unused-percent-secret";);
+ ToDefinition sink = findToDefinition(expandedRoute,
"http://localhost/unused-percent-secret";);
+
+ assertNotNull(sink, "Expected sink endpoint");
+ String uri = sink.getEndpointUri();
+ Map<String, Object> query = parseUriQuery(uri);
+ String password = rawOrPlain(query.get("password"));
+
+ assertEquals("se+cret%41", password,
+ () -> "Expected the original literal password to remain
intact: " + uri);
+ assertNotEquals("se cretA", password,
+ () -> "+ and % must not be URI-decoded: " + uri);
+ assertNotEquals("se cret", password,
+ () -> "+ must not be decoded as a space: " + uri);
+ assertNotEquals("secretA", password,
+ () -> "%41 must not be decoded as A: " + uri);
+
+ HttpEndpoint http = findHttpEndpoint(testContext,
"/unused-percent-secret");
+ assertNotNull(http, "Expected the HTTP sink endpoint to be
created");
+ String rawQuery = http.getHttpUri().getRawQuery();
+ assertNotNull(rawQuery, "Expected HTTP URI query parameters");
+ Map<String, Object> httpQuery = URISupport.parseQuery(rawQuery);
+ String httpPassword =
urlDecode(rawOrPlain(httpQuery.get("password")));
+ assertEquals("se+cret%41", httpPassword,
+ () -> "Expected the HTTP endpoint to receive the original
literal password: "
+ + http.getHttpUri());
+ assertNotEquals("se cretA", httpPassword,
+ () -> "+ and % must not be URI-decoded on the HTTP
endpoint: " + http.getHttpUri());
+ }
+ }
+
+ @Test
+ void sinkOptionsPreserveJsonRawValueWithoutPropertyPlaceholder() throws
Exception {
+ try (DefaultCamelContext testContext = new DefaultCamelContext()) {
+ testContext.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ routeTemplate("json-filter-sink")
+ .from("kamelet:source")
+ .to("http://localhost/unused-json-filter";);
+
+ from("direct:trigger")
+
.kamelet("json-filter-sink?sink.filter=RAW({\"a\":\"b&c\"})");
+ }
+ });
+
+ testContext.start();
+
+ RouteDefinition expandedRoute = findExpandedRouteWithOutput(
+ testContext, "http://localhost/unused-json-filter";);
+ ToDefinition sink = findToDefinition(expandedRoute,
"http://localhost/unused-json-filter";);
+
+ assertNotNull(sink, "Expected sink endpoint");
+ String uri = sink.getEndpointUri();
+ assertFalse(uri.contains("RAW{{"),
+ () -> "JSON must not be wrapped as RAW{{...}} or it
becomes a property placeholder: "
+ + uri);
+ assertTrue(uri.contains("RAW("),
+ () -> "Expected RAW(...) wrapping for the JSON value: " +
uri);
+
+ Map<String, Object> query = parseUriQuery(uri);
+ assertEquals("{\"a\":\"b&c\"}", rawOrPlain(query.get("filter")),
+ () -> "Expected the JSON value to remain intact: " + uri);
+ assertFalse(query.containsKey("c\"}"),
+ () -> "Ampersand in the JSON must not create a spurious
parameter: " + uri);
+ }
+ }
+
+ @Disabled("Requires a shared URIScanner fix; RAW values containing )&
inside quoted JSON are truncated during query parsing")
+ @Test
+ void sinkOptionsPreserveJsonRawValueWithClosingParenthesisAndAmpersand()
throws Exception {
+ try (DefaultCamelContext testContext = new DefaultCamelContext()) {
+ testContext.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ routeTemplate("json-edge-case-sink")
+ .from("kamelet:source")
+ .to("http://localhost/unused-json-edge-case";);
+
+ from("direct:trigger-json-edge-case")
+
.kamelet("json-edge-case-sink?sink.filter=RAW({\"filter\":\"(x)&y\"})");
+
+ }
+ });
+
+ testContext.start();
+
+ RouteDefinition expandedRoute = findExpandedRouteWithOutput(
+ testContext, "http://localhost/unused-json-edge-case";);
+ ToDefinition sink = findToDefinition(
+ expandedRoute, "http://localhost/unused-json-edge-case";);
+
+ assertNotNull(sink, "Expected sink endpoint for JSON edge case");
+
+ String uri = sink.getEndpointUri();
+ assertFalse(uri.contains("RAW{{"),
+ () -> "JSON must not be wrapped as RAW{{...}}: " + uri);
+ Map<String, Object> query = parseUriQuery(uri);
+
+ assertEquals("{\"filter\":\"(x)&y\"}",
rawOrPlain(query.get("filter")),
+ () -> "Expected JSON containing )& to remain intact: " +
uri);
+ assertFalse(query.containsKey("y}"),
+ () -> "Ampersand in the JSON must not create a spurious
parameter: " + uri);
+
+ HttpEndpoint http = findHttpEndpoint(testContext,
"/unused-json-edge-case");
+ assertNotNull(http, "Expected the HTTP sink endpoint to be
created");
+ Map<String, Object> httpQuery =
URISupport.parseQuery(http.getHttpUri().getRawQuery());
+ assertEquals("{\"filter\":\"(x)&y\"}",
urlDecode(rawOrPlain(httpQuery.get("filter"))),
+ () -> "Expected the HTTP endpoint to receive the complete
JSON: " + http.getHttpUri());
+ assertFalse(httpQuery.containsKey("y}"),
+ () -> "The HTTP endpoint must not receive a spurious y}
parameter: " + http.getHttpUri());
+ }
+ }
+
+ @Test
+ void sinkOptionsPreserveMultipleAmpersands() throws Exception {
+ assertHttpSinkOption("/unused-multi-amp", "sink.password=RAW(a&b&c)",
"password", "a&b&c", "b", "c");
+ }
+
+ @Test
+ void sinkOptionsPreserveUnquotedCloseParenAmpersand() throws Exception {
+ assertHttpSinkOption("/unused-unquoted-paren-amp",
"sink.password=RAW{p@ss)&word}", "password", "p@ss)&word",
+ "word)");
+ }
+
+ @Test
+ void sinkOptionsPreservePlusEqualsAndQuestionMark() throws Exception {
+ assertHttpSinkOption("/unused-plus-eq-q", "sink.filter=RAW(a+b=c?d)",
"filter", "a+b=c?d", new String[0]);
+ }
Review Comment:
My separator example was wrong. Camel normalizes the `kamelet:` endpoint
URI, which sorts its query parameters, so with
`sink.showAll=true&sink.level=INFO` the overriding `level` is applied first and
the old bug does not show. The overriding option has to sort last. Reading the
code (I did not run it), the `3dd7cac9` version of `appendParametersToURI`
gives `log:override&level=INFO&showAll=true` for this test, and the current
version gives `log:override?level=INFO&showAll=true`:
```java
@Test
void sinkOptionsOverrideTheOnlyExistingParameterAsTheLastOption() throws
Exception {
try (DefaultCamelContext testContext = new DefaultCamelContext()) {
testContext.addRoutes(new RouteBuilder() {
@Override
public void configure() {
routeTemplate("log-override-sink")
.from("kamelet:source")
.to("log:override?showAll=false");
from("direct:trigger")
.kamelet("log-override-sink?sink.level=INFO&sink.showAll=true");
}
});
testContext.start();
ToDefinition sink = findToDefinition(
findExpandedRouteWithOutput(testContext,
"log:override"), "log:override");
assertNotNull(sink, "Expected sink endpoint");
String uri = sink.getEndpointUri();
assertTrue(uri.startsWith("log:override?"), () -> "Expected ?
before the first option: " + uri);
Map<String, Object> query = parseUriQuery(uri);
assertEquals(2, query.size(), () -> uri);
assertEquals("INFO", query.get("level"), () -> uri);
assertEquals("true", query.get("showAll"), () -> uri);
}
}
```
--
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]
