Skip to content

UNOMI-954: Make Camel route refresh resilient to transient failures #779

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
sergehuber merged 3 commits into from
Jun 25, 2026
Merged

UNOMI-954: Make Camel route refresh resilient to transient failures #779

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
eb05611
UNOMI-954: Make Camel route refresh resilient to transient failures
sergehuber Jun 21, 2026
1cf71f6
UNOMI-954: Harden route refresh against null/malformed endpoints and …
sergehuber Jun 25, 2026
f1ad3a1
UNOMI-954: Type-safe retry key and visible status on route creation g…
sergehuber Jun 25, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions extensions/router/router-api/src/main/java/org/apache/unomi/router/api/RouterConstants.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down
11 changes: 11 additions & 0 deletions .../src/main/java/org/apache/unomi/router/api/services/ImportExportConfigurationService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
83 changes: 67 additions & 16 deletions ...er/router-core/src/main/java/org/apache/unomi/router/core/context/RouterCamelContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,7 @@
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
Expand Down Expand Up @@ -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<RetryKey, Integer> 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;
Expand Down Expand Up @@ -163,15 +170,29 @@ public void run() {
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();
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) {
Expand All @@ -187,15 +208,29 @@ public void run() {
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();
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);
}
Comment thread
sergehuber marked this conversation as resolved.
} 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) {
Expand Down Expand Up @@ -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 {
Expand All @@ -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())) {
Expand All @@ -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())) {
Expand Down
18 changes: 15 additions & 3 deletions ...src/main/java/org/apache/unomi/router/core/route/ProfileImportFromSourceRouteBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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());
}
}
}
Expand Down
6 changes: 6 additions & 0 deletions ...ervice/src/main/java/org/apache/unomi/router/services/ExportConfigurationServiceImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -109,4 +109,10 @@ public Map<String, Map<String, RouterConstants.CONFIG_CAMEL_REFRESH>> 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);
}
}
6 changes: 6 additions & 0 deletions ...ervice/src/main/java/org/apache/unomi/router/services/ImportConfigurationServiceImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -106,4 +106,10 @@ public Map<String, Map<String, RouterConstants.CONFIG_CAMEL_REFRESH>> 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);
}
}
31 changes: 16 additions & 15 deletions itests/src/test/java/org/apache/unomi/itests/ProfileImportActorsIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Loading