This is an automated email from the ASF dual-hosted git repository. sergehuber pushed a commit to branch UNOMI-954-camel-route-refresh-resilience in repository https://gitbox.apache.org/repos/asf/unomi.git
commit eb056116cd8af89a87c472eeb8519bb901e59f12 Author: Serge Huber <[email protected]> AuthorDate: Sun Jun 21 20:22:54 2026 +0200 UNOMI-954: Make Camel route refresh resilient to transient failures The snap-and-clear in consumeConfigsToBeRefresh() removed a config from the queue before its route was built, so any failure permanently dropped it with no retry. Fix: - Add requeueForRefresh() to ImportExportConfigurationService so the timer can re-schedule a failed config for the next tick, with a cap of 5 attempts before giving up. - updateProfileImportReaderRoute() now throws instead of silently returning when the config cannot be loaded, allowing the catch block to trigger the re-queue. - Add initialDelay=0 to the Camel file endpoint so the route polls immediately on start instead of waiting the default 1 second. - Harden ProfileImportActorsIT: gate on route start via keepTrying, force refreshIndex(Profile.class) before each ES search attempt. --- .../services/ImportExportConfigurationService.java | 11 +++++ .../router/core/context/RouterCamelContext.java | 54 ++++++++++++++++------ .../route/ProfileImportFromSourceRouteBuilder.java | 3 +- .../services/ExportConfigurationServiceImpl.java | 6 +++ .../services/ImportConfigurationServiceImpl.java | 6 +++ .../apache/unomi/itests/ProfileImportActorsIT.java | 31 +++++++------ 6 files changed, 81 insertions(+), 30 deletions(-) diff --git a/extensions/router/router-api/src/main/java/org/apache/unomi/router/api/services/ImportExportConfigurationService.java b/extensions/router/router-api/src/main/java/org/apache/unomi/router/api/services/ImportExportConfigurationService.java index bb20ba2d0..ba6d321b9 100644 --- a/extensions/router/router-api/src/main/java/org/apache/unomi/router/api/services/ImportExportConfigurationService.java +++ b/extensions/router/router-api/src/main/java/org/apache/unomi/router/api/services/ImportExportConfigurationService.java @@ -98,4 +98,15 @@ public interface ImportExportConfigurationService<T> { * @return map of tenantId to map of configId per operation to be done in camel */ Map<String, Map<String, RouterConstants.CONFIG_CAMEL_REFRESH>> consumeConfigsToBeRefresh(); + + /** + * Re-queues a configuration for a Camel route refresh. Used by the timer when a route + * creation attempt fails, so the config is retried on the next timer tick rather than + * being permanently dropped by the snap-and-clear in {@link #consumeConfigsToBeRefresh()}. + * + * @param tenantId the tenant that owns the configuration + * @param configId the configuration identifier to re-queue + * @param refreshType the refresh operation to retry + */ + void requeueForRefresh(String tenantId, String configId, RouterConstants.CONFIG_CAMEL_REFRESH refreshType); } diff --git a/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/context/RouterCamelContext.java b/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/context/RouterCamelContext.java index d8c59857a..6e714b174 100644 --- a/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/context/RouterCamelContext.java +++ b/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/context/RouterCamelContext.java @@ -49,6 +49,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.*; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; /** @@ -98,6 +99,8 @@ public class RouterCamelContext implements IRouterCamelContext { private ScheduledTask scheduledTask; private Integer configsRefreshInterval = 1000; + private static final int MAX_ROUTE_CREATION_RETRIES = 5; + private final Map<String, Integer> routeCreationRetryCount = new ConcurrentHashMap<>(); public void setExecHistorySize(String execHistorySize) { this.execHistorySize = execHistorySize; @@ -163,15 +166,27 @@ public class RouterCamelContext implements IRouterCamelContext { contextManager.executeAsTenant(tenantId, () -> { try { for (Map.Entry<String, RouterConstants.CONFIG_CAMEL_REFRESH> importConfigToRefresh : tenantImportConfigsToRefresh.getValue().entrySet()) { + String configId = importConfigToRefresh.getKey(); + RouterConstants.CONFIG_CAMEL_REFRESH refreshType = importConfigToRefresh.getValue(); try { - if (importConfigToRefresh.getValue().equals(RouterConstants.CONFIG_CAMEL_REFRESH.UPDATED)) { - updateProfileImportReaderRoute(importConfigToRefresh.getKey(), true); - } else if (importConfigToRefresh.getValue().equals(RouterConstants.CONFIG_CAMEL_REFRESH.REMOVED)) { - killExistingRoute(importConfigToRefresh.getKey(), true); + if (refreshType.equals(RouterConstants.CONFIG_CAMEL_REFRESH.UPDATED)) { + updateProfileImportReaderRoute(configId, true); + routeCreationRetryCount.remove(configId); + } else if (refreshType.equals(RouterConstants.CONFIG_CAMEL_REFRESH.REMOVED)) { + killExistingRoute(configId, true); + routeCreationRetryCount.remove(configId); } } catch (Exception e) { - LOGGER.error("Unexpected error while refreshing({}) camel route: {}", importConfigToRefresh.getValue(), - importConfigToRefresh.getKey(), e); + int attempt = routeCreationRetryCount.merge(configId, 1, Integer::sum); + if (attempt <= MAX_ROUTE_CREATION_RETRIES) { + LOGGER.error("Refreshing({}) camel route {} failed (attempt {}/{}) — will retry on next tick", + refreshType, configId, attempt, MAX_ROUTE_CREATION_RETRIES, e); + importConfigurationService.requeueForRefresh(tenantId, configId, refreshType); + } else { + LOGGER.error("Refreshing({}) camel route {} failed after {} attempts — giving up", + refreshType, configId, MAX_ROUTE_CREATION_RETRIES, e); + routeCreationRetryCount.remove(configId); + } } } } catch (Exception e) { @@ -187,15 +202,27 @@ public class RouterCamelContext implements IRouterCamelContext { contextManager.executeAsTenant(tenantId, () -> { try { for (Map.Entry<String, RouterConstants.CONFIG_CAMEL_REFRESH> exportConfigToRefresh : tenantExportConfigsToRefresh.getValue().entrySet()) { + String configId = exportConfigToRefresh.getKey(); + RouterConstants.CONFIG_CAMEL_REFRESH refreshType = exportConfigToRefresh.getValue(); try { - if (exportConfigToRefresh.getValue().equals(RouterConstants.CONFIG_CAMEL_REFRESH.UPDATED)) { - updateProfileExportReaderRoute(exportConfigToRefresh.getKey(), true); - } else if (exportConfigToRefresh.getValue().equals(RouterConstants.CONFIG_CAMEL_REFRESH.REMOVED)) { - killExistingRoute(exportConfigToRefresh.getKey(), true); + if (refreshType.equals(RouterConstants.CONFIG_CAMEL_REFRESH.UPDATED)) { + updateProfileExportReaderRoute(configId, true); + routeCreationRetryCount.remove(configId); + } else if (refreshType.equals(RouterConstants.CONFIG_CAMEL_REFRESH.REMOVED)) { + killExistingRoute(configId, true); + routeCreationRetryCount.remove(configId); } } catch (Exception e) { - LOGGER.error("Unexpected error while refreshing({}) camel route: {}", exportConfigToRefresh.getValue(), - exportConfigToRefresh.getKey(), e); + int attempt = routeCreationRetryCount.merge(configId, 1, Integer::sum); + if (attempt <= MAX_ROUTE_CREATION_RETRIES) { + LOGGER.error("Refreshing({}) camel route {} failed (attempt {}/{}) — will retry on next tick", + refreshType, configId, attempt, MAX_ROUTE_CREATION_RETRIES, e); + exportConfigurationService.requeueForRefresh(tenantId, configId, refreshType); + } else { + LOGGER.error("Refreshing({}) camel route {} failed after {} attempts — giving up", + refreshType, configId, MAX_ROUTE_CREATION_RETRIES, e); + routeCreationRetryCount.remove(configId); + } } } } catch (Exception e) { @@ -314,8 +341,7 @@ public class RouterCamelContext implements IRouterCamelContext { ImportConfiguration importConfiguration = importConfigurationService.load(configId); if (importConfiguration == null) { - LOGGER.warn("Cannot update profile import reader route, config: {} not found", configId); - return; + throw new IllegalStateException("Cannot update profile import reader route, config: " + configId + " not found — will be retried"); } if (RouterConstants.IMPORT_EXPORT_CONFIG_TYPE_RECURRENT.equals(importConfiguration.getConfigType())) { diff --git a/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/route/ProfileImportFromSourceRouteBuilder.java b/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/route/ProfileImportFromSourceRouteBuilder.java index 9a68e3797..f75abfb1c 100644 --- a/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/route/ProfileImportFromSourceRouteBuilder.java +++ b/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/route/ProfileImportFromSourceRouteBuilder.java @@ -137,7 +137,8 @@ public class ProfileImportFromSourceRouteBuilder extends RouterAbstractRouteBuil lineSplitProcessor.setProfilePropertyTypes(profileService.getTargetPropertyTypes("profiles")); String endpoint = (String) importConfiguration.getProperties().get("source"); - endpoint += "&moveFailed=.error"; + // Poll immediately when the route starts rather than waiting the default 1-second initialDelay. + endpoint += "&initialDelay=0&moveFailed=.error"; if (StringUtils.isNotBlank(endpoint) && allowedEndpoints.contains(endpoint.substring(0, endpoint.indexOf(':')))) { ProcessorDefinition prDef = from(endpoint) diff --git a/extensions/router/router-service/src/main/java/org/apache/unomi/router/services/ExportConfigurationServiceImpl.java b/extensions/router/router-service/src/main/java/org/apache/unomi/router/services/ExportConfigurationServiceImpl.java index 1b7a3f5d2..fbb01bd8f 100644 --- a/extensions/router/router-service/src/main/java/org/apache/unomi/router/services/ExportConfigurationServiceImpl.java +++ b/extensions/router/router-service/src/main/java/org/apache/unomi/router/services/ExportConfigurationServiceImpl.java @@ -109,4 +109,10 @@ public class ExportConfigurationServiceImpl implements ImportExportConfiguration camelConfigsToRefresh.clear(); return result; } + + @Override + public void requeueForRefresh(String tenantId, String configId, RouterConstants.CONFIG_CAMEL_REFRESH refreshType) { + camelConfigsToRefresh.computeIfAbsent(tenantId, k -> new ConcurrentHashMap<>()) + .putIfAbsent(configId, refreshType); + } } diff --git a/extensions/router/router-service/src/main/java/org/apache/unomi/router/services/ImportConfigurationServiceImpl.java b/extensions/router/router-service/src/main/java/org/apache/unomi/router/services/ImportConfigurationServiceImpl.java index b599f88bf..c0dfae3b4 100644 --- a/extensions/router/router-service/src/main/java/org/apache/unomi/router/services/ImportConfigurationServiceImpl.java +++ b/extensions/router/router-service/src/main/java/org/apache/unomi/router/services/ImportConfigurationServiceImpl.java @@ -106,4 +106,10 @@ public class ImportConfigurationServiceImpl implements ImportExportConfiguration camelConfigsToRefresh.clear(); return result; } + + @Override + public void requeueForRefresh(String tenantId, String configId, RouterConstants.CONFIG_CAMEL_REFRESH refreshType) { + camelConfigsToRefresh.computeIfAbsent(tenantId, k -> new ConcurrentHashMap<>()) + .putIfAbsent(configId, refreshType); + } } diff --git a/itests/src/test/java/org/apache/unomi/itests/ProfileImportActorsIT.java b/itests/src/test/java/org/apache/unomi/itests/ProfileImportActorsIT.java index 6ae186655..5ae6f595c 100644 --- a/itests/src/test/java/org/apache/unomi/itests/ProfileImportActorsIT.java +++ b/itests/src/test/java/org/apache/unomi/itests/ProfileImportActorsIT.java @@ -94,22 +94,23 @@ public class ProfileImportActorsIT extends BaseIT { ImportConfiguration savedImportConfigActors = importConfigurationService.save(importConfigActors, true); keepTrying("Failed waiting for actors import configuration to be saved", () -> importConfigurationService.load(importConfigActors.getItemId()), Objects::nonNull, DEFAULT_TRYING_TIMEOUT, DEFAULT_TRYING_TRIES); - // Wait for Camel route to be created and started (the timer runs every 1 second to process config refreshes) - // This gives us visibility into what Camel is doing instead of just waiting for results - // Using official Camel API: getRouteController().getRouteStatus() and Management API for statistics - boolean routeStarted = waitForCamelRouteStarted(itemId, 1000, 5); - if (routeStarted) { - String routeInfo = getCamelRouteInfo(itemId); - System.out.println("==== Camel Route Status: " + routeInfo + " ===="); - } else { - System.out.println("==== Camel Route '" + itemId + "' was not started within timeout ===="); - System.out.println("==== All Camel routes with status: " + getAllCamelRoutesWithStatus() + " ===="); - } - - //Wait for data to be processed + // Gate: wait for the Camel route to be created and started before polling for results. + // The timer fires every ~1 second to pick up config changes, so 15 retries (15 s) is generous. + keepTrying("Camel route '" + itemId + "' did not start — timer may not have fired or route creation failed", + () -> isCamelRouteStarted(itemId), + started -> started, + 1000, 15); + System.out.println("==== Camel Route Status: " + getCamelRouteInfo(itemId) + " ===="); + + // Wait for data to be processed. Force an ES index refresh before each search attempt so + // profiles written by UnomiStorageProcessor are visible to the Search API immediately. keepTrying("Failed waiting for actors initial import to complete", - () -> profileService.findProfilesByPropertyValue("properties.city", "hollywood", 0, 10, null), (p) -> p.getTotalSize() == 6, - 1000, 200); + () -> { + persistenceService.refreshIndex(Profile.class); + return profileService.findProfilesByPropertyValue("properties.city", "hollywood", 0, 10, null); + }, + (p) -> p.getTotalSize() == 6, + 1000, 30); // Refresh the persistence index to ensure the saved configuration is queryable in getAll() // This addresses the flakiness where getAll() returns 0 items due to index refresh delay
