This is an automated email from the ASF dual-hosted git repository.
sergehuber pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/unomi.git
The following commit(s) were added to refs/heads/master by this push:
new 5e7f613ae UNOMI-954: Make Camel route refresh resilient to transient
failures (#779)
5e7f613ae is described below
commit 5e7f613aef2eee2a25e9d373b6b558afd40d3a42
Author: Serge Huber <[email protected]>
AuthorDate: Thu Jun 25 11:22:47 2026 +0200
UNOMI-954: Make Camel route refresh resilient to transient failures (#779)
---
.../apache/unomi/router/api/RouterConstants.java | 1 +
.../services/ImportExportConfigurationService.java | 11 +++
.../router/core/context/RouterCamelContext.java | 83 +++++++++++++++++-----
.../route/ProfileImportFromSourceRouteBuilder.java | 18 ++++-
.../services/ExportConfigurationServiceImpl.java | 6 ++
.../services/ImportConfigurationServiceImpl.java | 6 ++
.../apache/unomi/itests/ProfileImportActorsIT.java | 31 ++++----
7 files changed, 122 insertions(+), 34 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 @@ public interface RouterConstants {
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<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..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.Logger;
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<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;
@@ -163,15 +170,29 @@ 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();
+ 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 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();
+ 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 class RouterCamelContext implements
IRouterCamelContext {
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 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()))
{
@@ -339,8 +391,7 @@ public class RouterCamelContext implements
IRouterCamelContext {
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 class ProfileImportFromSourceRouteBuilder extends
RouterAbstractRouteBuil
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 class ProfileImportFromSourceRouteBuilder extends
RouterAbstractRouteBuil
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 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
