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-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..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 @@ -49,6 +49,7 @@ import org.slf4j.LoggerFactory; import java.util.*; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; /** @@ -98,6 +99,12 @@ 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<>(); + + /** 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) { this.execHistorySize = execHistorySize; @@ -163,15 +170,29 @@ 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(); + RetryKey retryKey = new RetryKey("import", tenantId, configId); 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(retryKey); + } else if (refreshType.equals(RouterConstants.CONFIG_CAMEL_REFRESH.REMOVED)) { + killExistingRoute(configId, true); + routeCreationRetryCount.remove(retryKey); } } catch (Exception e) { - LOGGER.error("Unexpected error while refreshing({}) camel route: {}", importConfigToRefresh.getValue(), - importConfigToRefresh.getKey(), e); + 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); + 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(retryKey); + markImportRouteCreationFailed(configId); + } } } } catch (Exception e) { @@ -187,15 +208,29 @@ 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(); + RetryKey retryKey = new RetryKey("export", tenantId, configId); 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(retryKey); + } else if (refreshType.equals(RouterConstants.CONFIG_CAMEL_REFRESH.REMOVED)) { + killExistingRoute(configId, true); + routeCreationRetryCount.remove(retryKey); } } catch (Exception e) { - LOGGER.error("Unexpected error while refreshing({}) camel route: {}", exportConfigToRefresh.getValue(), - exportConfigToRefresh.getKey(), e); + 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); + 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(retryKey); + markExportRouteCreationFailed(configId); + } } } } catch (Exception e) { @@ -294,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 { @@ -314,8 +367,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())) { @@ -339,8 +391,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 9a68e3797..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,9 +137,21 @@ public void configure() throws Exception { lineSplitProcessor.setProfilePropertyTypes(profileService.getTargetPropertyTypes("profiles")); String endpoint = (String) importConfiguration.getProperties().get("source"); - endpoint += "&moveFailed=.error"; + 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"; + + 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 (StringUtils.isNotBlank(endpoint) && allowedEndpoints.contains(endpoint.substring(0, endpoint.indexOf(':')))) { + 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 @@ -176,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()); } } } 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