feat(core): default error handler

camel-k-core/api/src/main/java/org/apache/camel/k/SourceType.java

@@ -17,6 +17,8 @@
 package org.apache.camel.k;
 
 public enum SourceType {
-    source,
-    template
+    // Order matters. We want the sources to be loaded following this enum order.
+    errorHandler,
+    template,
+    source
 }

camel-k-core/support/src/main/java/org/apache/camel/k/listener/SourcesConfigurer.java

@@ -16,8 +16,11 @@
  */
 package org.apache.camel.k.listener;
 
+import java.util.Arrays;
+
 import org.apache.camel.k.Runtime;
 import org.apache.camel.k.SourceDefinition;
+import org.apache.camel.k.SourceType;
 import org.apache.camel.k.support.Constants;
 import org.apache.camel.k.support.PropertiesSupport;
 import org.apache.camel.k.support.SourcesSupport;
@@ -64,14 +67,49 @@ protected void accept(Runtime runtime) {
         // property that can't be bound to this configurer.
         //
         PropertiesSupport.bindProperties(
-            runtime.getCamelContext(),
-            this,
-            k -> k.startsWith(CAMEL_K_SOURCES_PREFIX),
-            CAMEL_K_PREFIX);
+                runtime.getCamelContext(),
+                this,
+                k -> k.startsWith(CAMEL_K_SOURCES_PREFIX),
+                CAMEL_K_PREFIX);
+
+        checkUniqueErrorHandler();
+        sortSources();
 
         if (ObjectHelper.isNotEmpty(this.getSources())) {
             SourcesSupport.loadSources(runtime, this.getSources());
         }
     }
 
+    private void checkUniqueErrorHandler() {
+        checkUniqueErrorHandler(this.sources);
+    }
+
+    static void checkUniqueErrorHandler(SourceDefinition[] sources) {
+        long errorHandlers = sources == null ? 0 : Arrays.stream(sources).filter(s -> s.getType() == SourceType.errorHandler).count();
+        if (errorHandlers > 1) {
+            throw new IllegalArgumentException("Expected only one error handler source type, got " + errorHandlers);
+        }
+    }
+
+    private void sortSources() {
+        sortSources(this.getSources());
+    }
+
+    static void sortSources(SourceDefinition[] sources) {
+        if (sources == null) {
+            return;
+        }
+        // We must ensure the source order as defined in SourceType enum
+        Arrays.sort(sources,
+                (a, b) -> {
+                    if (a.getType() == null) {
+                        return SourceType.source.compareTo(b.getType());
+                    } else if (b.getType() == null) {
+                        return a.getType().compareTo(SourceType.source);
+                    } else {
+                        return a.getType().compareTo(b.getType());
+                    }
+                });
+    }
+
 }

camel-k-core/support/src/main/java/org/apache/camel/k/support/SourcesSupport.java

@@ -16,12 +16,14 @@
  */
 package org.apache.camel.k.support;
 
+import java.lang.reflect.Field;
 import java.util.Collection;
 import java.util.List;
 
 import org.apache.camel.ExtendedCamelContext;
 import org.apache.camel.RoutesBuilder;
 import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.builder.ErrorHandlerBuilder;
 import org.apache.camel.builder.RouteBuilder;
 import org.apache.camel.builder.RouteBuilderLifecycleStrategy;
 import org.apache.camel.k.Runtime;
@@ -62,7 +64,7 @@ protected void accept(Runtime runtime) {
     }
 
     public static void loadSources(Runtime runtime, String... routes) {
-        for (String route: routes) {
+        for (String route : routes) {
             if (ObjectHelper.isEmpty(route)) {
                 continue;
             }
@@ -78,7 +80,7 @@ public static void loadSources(Runtime runtime, String... routes) {
     }
 
     public static void loadSources(Runtime runtime, SourceDefinition... definitions) {
-        for (SourceDefinition definition: definitions) {
+        for (SourceDefinition definition : definitions) {
             LOGGER.info("Loading routes from: {}", definition);
 
             load(runtime, Sources.fromDefinition(definition));
@@ -128,21 +130,64 @@ public void afterConfigure(RouteBuilder builder) {
                     }
                 });
                 break;
+            case errorHandler:
+                if (!source.getInterceptors().isEmpty()) {
+                    LOGGER.warn("Interceptors associated to the route template {} will be ignored", source.getName());
+                }
+
+                interceptors = List.of(new RouteBuilderLifecycleStrategy() {
+                    @Override
+                    public void afterConfigure(RouteBuilder builder) {
+                        List<RouteDefinition> routes = builder.getRouteCollection().getRoutes();
+                        List<RouteTemplateDefinition> templates = builder.getRouteTemplateCollection().getRouteTemplates();
+
+                        if (routes.size() > 0) {
+                            throw new IllegalArgumentException("There should be no route definition, got " + routes.size());
+                        }
+                        if (!templates.isEmpty()) {
+                            throw new IllegalArgumentException("There should not be any template, got " + templates.size());
+                        }
+
+                        if (hasErrorHandlerBuilder(builder)) {
+                            LOGGER.debug("Setting default error handler builder factory as {}", builder.getErrorHandlerBuilder());
+                            runtime.getCamelContext().adapt(ExtendedCamelContext.class).setErrorHandlerFactory(builder.getErrorHandlerBuilder());
+                        }
+                    }
+                });
+                break;
             default:
                 throw new IllegalArgumentException("Unknown source type: " + source.getType());
         }
 
         try {
             final Resource resource = Sources.asResource(runtime.getCamelContext(), source);
-            final ExtendedCamelContext ecc =  runtime.getCamelContext(ExtendedCamelContext.class);
+            final ExtendedCamelContext ecc = runtime.getCamelContext(ExtendedCamelContext.class);
             final Collection<RoutesBuilder> builders = ecc.getRoutesLoader().findRoutesBuilders(resource);
 
             builders.stream()
-                .map(RouteBuilder.class::cast)
-                .peek(b -> interceptors.forEach(b::addLifecycleInterceptor))
-                .forEach(runtime::addRoutes);
+                    .map(RouteBuilder.class::cast)
+                    .peek(b -> interceptors.forEach(b::addLifecycleInterceptor))
+                    .forEach(runtime::addRoutes);
         } catch (Exception e) {
             throw RuntimeCamelException.wrapRuntimeCamelException(e);
         }
     }
+
+    static boolean hasErrorHandlerBuilder(RouteBuilder builder) {
+        //return builder.hasErrorHandlerBuilder();
+        // TODO We need to replace the following workaround with the statement above once we switch to camel-3.10.0 or above
+        try {
+            Field f = RouteBuilder.class.getSuperclass().getDeclaredField("errorHandlerBuilder");
+            f.setAccessible(true);
+            ErrorHandlerBuilder privateErrorHandlerBuilder = (ErrorHandlerBuilder) f.get(builder);
+            return privateErrorHandlerBuilder != null;
+        } catch (Exception e) {
+            throw new IllegalArgumentException("Something went wrong while checking the error handler builder", e);
+        }
+    }
+
+    public static void loadErrorHandlerSource(Runtime runtime, SourceDefinition errorHandlerSourceDefinition) {
+        LOGGER.info("Loading error handler from: {}", errorHandlerSourceDefinition);
+        load(runtime, Sources.fromDefinition(errorHandlerSourceDefinition));
+    }
 }

camel-k-core/support/src/test/java/org/apache/camel/k/listener/SourceConfigurerTest.java

@@ -0,0 +1,119 @@
+/*
+ * 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.k.listener;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.k.support.PropertiesSupport;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import static org.apache.camel.k.test.CamelKTestSupport.asProperties;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+public class SourceConfigurerTest {
+    @Test
+    public void shouldLoadMultipleSources() {
+        CamelContext context = new DefaultCamelContext();
+        context.getPropertiesComponent().setInitialProperties(asProperties(
+                "camel.k.sources[0].name", "sourceName0",
+                "camel.k.sources[0].location", "classpath:MyTemplate1.java",
+                "camel.k.sources[0].type", "template",
+                "camel.k.sources[1].name", "err1",
+                "camel.k.sources[1].location", "classpath:MyTemplate2.java",
+                "camel.k.sources[2].name", "err2",
+                "camel.k.sources[2].location", "classpath:Err2.java",
+                "camel.k.sources[2].type", "errorHandler"
+        ));
+
+        SourcesConfigurer configuration = new SourcesConfigurer();
+
+        PropertiesSupport.bindProperties(
+                context,
+                configuration,
+                k -> k.startsWith(SourcesConfigurer.CAMEL_K_SOURCES_PREFIX),
+                SourcesConfigurer.CAMEL_K_PREFIX);
+
+        assertThat(configuration.getSources().length).isEqualTo(3);
+    }
+
+    @Test
+    public void shouldFailOnMultipleErrorHandlers() {
+        CamelContext context = new DefaultCamelContext();
+        context.getPropertiesComponent().setInitialProperties(asProperties(
+                "camel.k.sources[0].name", "sourceName0",
+                "camel.k.sources[0].location", "classpath:MyTemplate1.java",
+                "camel.k.sources[0].type", "template",
+                "camel.k.sources[1].name", "err1",
+                "camel.k.sources[1].location", "classpath:Err1.java",
+                "camel.k.sources[1].type", "errorHandler",
+                "camel.k.sources[2].name", "err2",
+                "camel.k.sources[2].location", "classpath:Err2.java",
+                "camel.k.sources[2].type", "errorHandler"
+        ));
+
+        SourcesConfigurer configuration = new SourcesConfigurer();
+
+        PropertiesSupport.bindProperties(
+                context,
+                configuration,
+                k -> k.startsWith(SourcesConfigurer.CAMEL_K_SOURCES_PREFIX),
+                SourcesConfigurer.CAMEL_K_PREFIX);
+
+        assertThat(configuration.getSources().length).isEqualTo(3);
+        Assertions.assertThrows(IllegalArgumentException.class, () -> {
+            SourcesConfigurer.checkUniqueErrorHandler(configuration.getSources());
+        }, "java.lang.IllegalArgumentException: Expected only one error handler source type, got 2");
+    }
+
+    @Test
+    public void shouldOrderSourcesByType() {
+        CamelContext context = new DefaultCamelContext();
+        context.getPropertiesComponent().setInitialProperties(asProperties(
+                "camel.k.sources[0].name", "template1",
+                "camel.k.sources[0].type", "template",
+                "camel.k.sources[1].name", "source1",
+                "camel.k.sources[2].name", "source2",
+                "camel.k.sources[2].type", "source",
+                "camel.k.sources[3].name", "errorHandler1",
+                "camel.k.sources[3].type", "errorHandler"
+        ));
+
+        SourcesConfigurer configuration = new SourcesConfigurer();
+
+        PropertiesSupport.bindProperties(
+                context,
+                configuration,
+                k -> k.startsWith(SourcesConfigurer.CAMEL_K_SOURCES_PREFIX),
+                SourcesConfigurer.CAMEL_K_PREFIX);
+        SourcesConfigurer.sortSources(configuration.getSources());
+
+        assertThat(configuration.getSources().length).isEqualTo(4);
+        assertThat(configuration.getSources()[0].getName()).isEqualTo("errorHandler1");
+        assertThat(configuration.getSources()[1].getName()).isEqualTo("template1");
+        // Order for the same type does not matter
+        assertThat(configuration.getSources()[2].getName()).contains("source");
+        assertThat(configuration.getSources()[3].getName()).contains("source");
+    }
+
+    @Test
+    public void shouldNotFailOnEmptySources() {
+        SourcesConfigurer.sortSources(null);
+        SourcesConfigurer.checkUniqueErrorHandler(null);
+    }
+}