From eb056116cd8af89a87c472eeb8519bb901e59f12 Mon Sep 17 00:00:00 2001 From: Serge Huber Date: Sun, 21 Jun 2026 20:22:54 +0200 Subject: [PATCH 1/3] 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. --- .../ImportExportConfigurationService.java | 11 ++++ .../core/context/RouterCamelContext.java | 54 ++++++++++++++----- .../ProfileImportFromSourceRouteBuilder.java | 3 +- .../ExportConfigurationServiceImpl.java | 6 +++ .../ImportConfigurationServiceImpl.java | 6 +++ .../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 { * @return map of tenantId to map of configId per operation to be done in camel */ Map> 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.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 routeCreationRetryCount = new ConcurrentHashMap<>(); public void setExecHistorySize(String execHistorySize) { this.execHistorySize = execHistorySize; @@ -163,15 +166,27 @@ public void run() { contextManager.executeAsTenant(tenantId, () -> { try { for (Map.Entry 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 void run() { contextManager.executeAsTenant(tenantId, () -> { try { for (Map.Entry 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 void updateProfileImportReaderRoute(String configId, boolean fireEvent) t 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 void configure() throws Exception { 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 Map> consumeCon 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 Map> consumeCon 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 void testImportActors() throws InterruptedException { 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 From 1cf71f6a15c9b6d8789468bb5ab1672b00b041f0 Mon Sep 17 00:00:00 2001 From: Serge Huber Date: Thu, 25 Jun 2026 08:53:50 +0200 Subject: [PATCH 2/3] UNOMI-954: Harden route refresh against null/malformed endpoints and missing export configs Guard against blank or schemeless source endpoints in ProfileImportFromSourceRouteBuilder, which previously crashed with an NPE or StringIndexOutOfBoundsException instead of skipping the route. Make updateProfileExportReaderRoute() throw on a missing config like its import counterpart already does, so the export refresh loop correctly triggers the retry/requeue logic instead of silently giving up. Also namespace the route-creation retry counter by direction and tenant to avoid key collisions between import/export configs and across tenants. --- .../core/context/RouterCamelContext.java | 25 +++++++++++-------- .../ProfileImportFromSourceRouteBuilder.java | 15 +++++++++-- 2 files changed, 28 insertions(+), 12 deletions(-) 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 6e714b174..29ec03c06 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 @@ -102,6 +102,10 @@ public class RouterCamelContext implements IRouterCamelContext { private static final int MAX_ROUTE_CREATION_RETRIES = 5; private final Map routeCreationRetryCount = new ConcurrentHashMap<>(); + private static String retryKey(String direction, String tenantId, String configId) { + return direction + ":" + tenantId + ":" + configId; + } + public void setExecHistorySize(String execHistorySize) { this.execHistorySize = execHistorySize; } @@ -168,16 +172,17 @@ public void run() { for (Map.Entry importConfigToRefresh : tenantImportConfigsToRefresh.getValue().entrySet()) { String configId = importConfigToRefresh.getKey(); RouterConstants.CONFIG_CAMEL_REFRESH refreshType = importConfigToRefresh.getValue(); + String retryKey = retryKey("import", tenantId, configId); try { if (refreshType.equals(RouterConstants.CONFIG_CAMEL_REFRESH.UPDATED)) { updateProfileImportReaderRoute(configId, true); - routeCreationRetryCount.remove(configId); + routeCreationRetryCount.remove(retryKey); } else if (refreshType.equals(RouterConstants.CONFIG_CAMEL_REFRESH.REMOVED)) { killExistingRoute(configId, true); - routeCreationRetryCount.remove(configId); + routeCreationRetryCount.remove(retryKey); } } catch (Exception e) { - int attempt = routeCreationRetryCount.merge(configId, 1, Integer::sum); + int attempt = routeCreationRetryCount.merge(retryKey, 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); @@ -185,7 +190,7 @@ public void run() { } else { LOGGER.error("Refreshing({}) camel route {} failed after {} attempts — giving up", refreshType, configId, MAX_ROUTE_CREATION_RETRIES, e); - routeCreationRetryCount.remove(configId); + routeCreationRetryCount.remove(retryKey); } } } @@ -204,16 +209,17 @@ public void run() { for (Map.Entry exportConfigToRefresh : tenantExportConfigsToRefresh.getValue().entrySet()) { String configId = exportConfigToRefresh.getKey(); RouterConstants.CONFIG_CAMEL_REFRESH refreshType = exportConfigToRefresh.getValue(); + String retryKey = retryKey("export", tenantId, configId); try { if (refreshType.equals(RouterConstants.CONFIG_CAMEL_REFRESH.UPDATED)) { updateProfileExportReaderRoute(configId, true); - routeCreationRetryCount.remove(configId); + routeCreationRetryCount.remove(retryKey); } else if (refreshType.equals(RouterConstants.CONFIG_CAMEL_REFRESH.REMOVED)) { killExistingRoute(configId, true); - routeCreationRetryCount.remove(configId); + routeCreationRetryCount.remove(retryKey); } } catch (Exception e) { - int attempt = routeCreationRetryCount.merge(configId, 1, Integer::sum); + int attempt = routeCreationRetryCount.merge(retryKey, 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); @@ -221,7 +227,7 @@ public void run() { } else { LOGGER.error("Refreshing({}) camel route {} failed after {} attempts — giving up", refreshType, configId, MAX_ROUTE_CREATION_RETRIES, e); - routeCreationRetryCount.remove(configId); + routeCreationRetryCount.remove(retryKey); } } } @@ -365,8 +371,7 @@ public void updateProfileExportReaderRoute(String configId, boolean fireEvent) t ExportConfiguration exportConfiguration = exportConfigurationService.load(configId); if (exportConfiguration == null) { - LOGGER.warn("Cannot update profile export reader route, config: {} not found", configId); - return; + throw new IllegalStateException("Cannot update profile export reader route, config: " + configId + " not found — will be retried"); } if (RouterConstants.IMPORT_EXPORT_CONFIG_TYPE_RECURRENT.equals(exportConfiguration.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 f75abfb1c..845f023c1 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,10 +137,21 @@ public void configure() throws Exception { lineSplitProcessor.setProfilePropertyTypes(profileService.getTargetPropertyTypes("profiles")); String endpoint = (String) importConfiguration.getProperties().get("source"); + if (StringUtils.isBlank(endpoint)) { + LOGGER.error("No source endpoint configured, route {} will be skipped.", importConfiguration.getItemId()); + continue; + } // 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(':')))) { + int schemeSeparatorIndex = endpoint.indexOf(':'); + if (schemeSeparatorIndex < 0) { + LOGGER.error("Endpoint {} has no scheme, route {} will be skipped.", endpoint, importConfiguration.getItemId()); + continue; + } + String endpointScheme = endpoint.substring(0, schemeSeparatorIndex); + + if (allowedEndpoints.contains(endpointScheme)) { ProcessorDefinition prDef = from(endpoint) .routeId(importConfiguration.getItemId())// This allow identification of the route for manual start/stop .autoStartup(importConfiguration.isActive())// Auto-start if the import configuration is set active @@ -177,7 +188,7 @@ public void process(Exchange exchange) throws Exception { prDef.to((String) getEndpointURI(RouterConstants.DIRECTION_FROM, RouterConstants.DIRECT_IMPORT_DEPOSIT_BUFFER)); } } else { - LOGGER.error("Endpoint scheme {} is not allowed, route {} will be skipped.", endpoint.substring(0, endpoint.indexOf(':')), importConfiguration.getItemId()); + LOGGER.error("Endpoint scheme {} is not allowed, route {} will be skipped.", endpointScheme, importConfiguration.getItemId()); } } } From f1ad3a117c3dab185d69b8186ea700699fe2c8eb Mon Sep 17 00:00:00 2001 From: Serge Huber Date: Thu, 25 Jun 2026 09:43:33 +0200 Subject: [PATCH 3/3] UNOMI-954: Type-safe retry key and visible status on route creation give-up Replace the colon-delimited String retry key with a RetryKey record so tenant/config IDs can never collide across import/export or tenants. Persist CONFIG_STATUS_ROUTE_CREATION_FAILED on the configuration once retries are exhausted, instead of only logging it, so the failure is visible through the existing config APIs rather than only in logs. --- .../unomi/router/api/RouterConstants.java | 1 + .../core/context/RouterCamelContext.java | 30 +++++++++++++++---- 2 files changed, 26 insertions(+), 5 deletions(-) diff --git a/extensions/router/router-api/src/main/java/org/apache/unomi/router/api/RouterConstants.java b/extensions/router/router-api/src/main/java/org/apache/unomi/router/api/RouterConstants.java index 0b17be825..5576296d8 100644 --- a/extensions/router/router-api/src/main/java/org/apache/unomi/router/api/RouterConstants.java +++ b/extensions/router/router-api/src/main/java/org/apache/unomi/router/api/RouterConstants.java @@ -32,6 +32,7 @@ enum CONFIG_CAMEL_REFRESH { String CONFIG_STATUS_COMPLETE_ERRORS = "ERRORS"; String CONFIG_STATUS_COMPLETE_SUCCESS = "SUCCESS"; String CONFIG_STATUS_COMPLETE_WITH_ERRORS = "WITH_ERRORS"; + String CONFIG_STATUS_ROUTE_CREATION_FAILED = "ROUTE_CREATION_FAILED"; String IMPORT_EXPORT_CONFIG_TYPE_RECURRENT = "recurrent"; String IMPORT_EXPORT_CONFIG_TYPE_ONESHOT = "oneshot"; 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 29ec03c06..090d6927e 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 @@ -100,10 +100,10 @@ public class RouterCamelContext implements IRouterCamelContext { private Integer configsRefreshInterval = 1000; private static final int MAX_ROUTE_CREATION_RETRIES = 5; - private final Map routeCreationRetryCount = new ConcurrentHashMap<>(); + private final Map routeCreationRetryCount = new ConcurrentHashMap<>(); - private static String retryKey(String direction, String tenantId, String configId) { - return direction + ":" + tenantId + ":" + configId; + /** Identifies a route refresh attempt being retried, scoped by direction and tenant so import/export configs and different tenants can never collide. */ + private record RetryKey(String direction, String tenantId, String configId) { } public void setExecHistorySize(String execHistorySize) { @@ -172,7 +172,7 @@ public void run() { for (Map.Entry importConfigToRefresh : tenantImportConfigsToRefresh.getValue().entrySet()) { String configId = importConfigToRefresh.getKey(); RouterConstants.CONFIG_CAMEL_REFRESH refreshType = importConfigToRefresh.getValue(); - String retryKey = retryKey("import", tenantId, configId); + RetryKey retryKey = new RetryKey("import", tenantId, configId); try { if (refreshType.equals(RouterConstants.CONFIG_CAMEL_REFRESH.UPDATED)) { updateProfileImportReaderRoute(configId, true); @@ -191,6 +191,7 @@ public void run() { LOGGER.error("Refreshing({}) camel route {} failed after {} attempts — giving up", refreshType, configId, MAX_ROUTE_CREATION_RETRIES, e); routeCreationRetryCount.remove(retryKey); + markImportRouteCreationFailed(configId); } } } @@ -209,7 +210,7 @@ public void run() { for (Map.Entry exportConfigToRefresh : tenantExportConfigsToRefresh.getValue().entrySet()) { String configId = exportConfigToRefresh.getKey(); RouterConstants.CONFIG_CAMEL_REFRESH refreshType = exportConfigToRefresh.getValue(); - String retryKey = retryKey("export", tenantId, configId); + RetryKey retryKey = new RetryKey("export", tenantId, configId); try { if (refreshType.equals(RouterConstants.CONFIG_CAMEL_REFRESH.UPDATED)) { updateProfileExportReaderRoute(configId, true); @@ -228,6 +229,7 @@ public void run() { LOGGER.error("Refreshing({}) camel route {} failed after {} attempts — giving up", refreshType, configId, MAX_ROUTE_CREATION_RETRIES, e); routeCreationRetryCount.remove(retryKey); + markExportRouteCreationFailed(configId); } } } @@ -327,6 +329,24 @@ public boolean isEnabled(EventObject event) { camelContext.start(); } + /** Persists a visible failure status on the configuration once route creation has exhausted all retries, instead of only logging it. */ + private void markImportRouteCreationFailed(String configId) { + ImportConfiguration config = importConfigurationService.load(configId); + if (config != null) { + config.setStatus(RouterConstants.CONFIG_STATUS_ROUTE_CREATION_FAILED); + importConfigurationService.save(config, false); + } + } + + /** Persists a visible failure status on the configuration once route creation has exhausted all retries, instead of only logging it. */ + private void markExportRouteCreationFailed(String configId) { + ExportConfiguration config = exportConfigurationService.load(configId); + if (config != null) { + config.setStatus(RouterConstants.CONFIG_STATUS_ROUTE_CREATION_FAILED); + exportConfigurationService.save(config, false); + } + } + /** {@inheritDoc} */ @Override public void killExistingRoute(String routeId, boolean fireEvent) throws Exception {