This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 506dc13b6758 CAMEL-25811: camel-keycloak - the consumer reads every
event since the previous poll (#28202)
506dc13b6758 is described below
commit 506dc13b6758506c67bc762f73a3bea1809f5acb
Author: allthingssecurity <[email protected]>
AuthorDate: Thu Oct 22 00:38:18 2026 +0530
CAMEL-25811: camel-keycloak - the consumer reads every event since the
previous poll (#28202)
* CAMEL-25811: camel-keycloak - the consumer reads every event since the
previous poll
Keycloak returns the most recent events first. The consumer asked for the
first maxResults events (default 100) and made the newest one its
checkpoint, so when more than maxResults user or admin events happened
between two polls (a busy realm, or the first poll after Keycloak was
unreachable) the older ones were never returned again and were lost
silently.
Once a checkpoint is set, a poll now reads the following pages as well,
until a page reaches an event older than the checkpoint or is the last
one; the fingerprints skip the events a page repeats when new events
arrive while the pages are read. The fingerprint cache is limited after
the poll instead of while the events are read. The first poll is
unchanged.
* CAMEL-25811: camel-keycloak - address review: page after a first poll
without events, maxResults wording, doc on large polls
- When a poll succeeds before any event was seen (a new realm, or a
filter that has not matched yet), the checkpoint becomes 0, so the next
polls read every new event page by page instead of only the latest
maxResults.
- The maxResults description says it is the size of the first poll and
the page size of the later polls (component JSON, catalog, DSL).
- The consumer page says that a poll after a long outage can be large:
maxResults limits the page, not the poll.
* CAMEL-25811: camel-keycloak - address review: no events for a response
without content
The admin client returns null for a 2xx response without a content type
(such as 204 No Content); the consumer now polls no events instead of
failing on the null list.
* CAMEL-25811: camel-keycloak - address review: a response without content
is no answer
A response without content on the first page made the consumer take the
realm as empty: lastEventTime became 0, and the next poll paged through
the whole history of the realm and routed it as new events. Such a
response is now no answer: the poll routes nothing and the checkpoint
does not move, so the next poll is still a first poll.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/catalog/components/keycloak.json | 4 +-
.../camel/catalog/docs/keycloak-consumer.adoc | 11 +-
.../apache/camel/component/keycloak/keycloak.json | 4 +-
.../src/main/docs/keycloak-consumer.adoc | 11 +-
.../component/keycloak/KeycloakConfiguration.java | 7 +-
.../camel/component/keycloak/KeycloakConsumer.java | 115 +++++++---
.../KeycloakConsumerMoreThanMaxResultsTest.java | 254 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 8 +
.../dsl/KeycloakComponentBuilderFactory.java | 3 +-
.../dsl/KeycloakEndpointBuilderFactory.java | 18 +-
10 files changed, 390 insertions(+), 45 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/keycloak.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/keycloak.json
index 9464ca5e57f1..91c717d8cfab 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/keycloak.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/keycloak.json
@@ -42,7 +42,7 @@
"introspectionCacheTtl": { "index": 15, "kind": "property", "displayName":
"Introspection Cache Ttl", "group": "common", "label": "", "required": false,
"type": "integer", "javaType": "long", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 60, "configurationClass":
"org.apache.camel.component.keycloak.KeycloakConfiguration",
"configurationField": "configuration", "description": "Time-to-live for cached
introspection results in seconds" },
"ipAddress": { "index": 16, "kind": "property", "displayName": "Ip
Address", "group": "common", "label": "", "required": false, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "autowired": false,
"secret": false, "configurationClass":
"org.apache.camel.component.keycloak.KeycloakConfiguration",
"configurationField": "configuration", "description": "Filter events by IP
address" },
"keycloakClient": { "index": 17, "kind": "property", "displayName":
"Keycloak Client", "group": "common", "label": "", "required": false, "type":
"object", "javaType": "org.keycloak.admin.client.Keycloak", "deprecated":
false, "deprecationNote": "", "autowired": true, "secret": false,
"configurationClass":
"org.apache.camel.component.keycloak.KeycloakConfiguration",
"configurationField": "configuration", "description": "To use an existing
configured Keycloak admin client" },
- "maxResults": { "index": 18, "kind": "property", "displayName": "Max
Results", "group": "common", "label": "", "required": false, "type": "integer",
"javaType": "int", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": 100, "configurationClass":
"org.apache.camel.component.keycloak.KeycloakConfiguration",
"configurationField": "configuration", "description": "Maximum number of events
to retrieve per poll" },
+ "maxResults": { "index": 18, "kind": "property", "displayName": "Max
Results", "group": "common", "label": "", "required": false, "type": "integer",
"javaType": "int", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": 100, "configurationClass":
"org.apache.camel.component.keycloak.KeycloakConfiguration",
"configurationField": "configuration", "description": "The number of events the
first poll retrieves, and the page size of the later polls, which read every e
[...]
"operation": { "index": 19, "kind": "property", "displayName":
"Operation", "group": "common", "label": "", "required": false, "type": "enum",
"javaType": "org.apache.camel.component.keycloak.KeycloakOperations", "enum": [
"createRealm", "deleteRealm", "getRealm", "updateRealm", "createUser",
"deleteUser", "getUser", "updateUser", "listUsers", "searchUsers",
"createRole", "deleteRole", "getRole", "updateRole", "listRoles",
"assignRoleToUser", "removeRoleFromUser", "getUserRoles", "cr [...]
"operationTypes": { "index": 20, "kind": "property", "displayName":
"Operation Types", "group": "common", "label": "", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.keycloak.KeycloakConfiguration",
"configurationField": "configuration", "description": "Filter admin events by
operation types (comma-separated list, e.g., CREATE,UPDATE,DELETE)" },
"password": { "index": 21, "kind": "property", "displayName": "Password",
"group": "common", "label": "", "required": false, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "autowired": false,
"secret": true, "security": "secret", "configurationClass":
"org.apache.camel.component.keycloak.KeycloakConfiguration",
"configurationField": "configuration", "description": "Keycloak password" },
@@ -146,7 +146,7 @@
"introspectionCacheTtl": { "index": 15, "kind": "parameter",
"displayName": "Introspection Cache Ttl", "group": "common", "label": "",
"required": false, "type": "integer", "javaType": "long", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": 60, "configurationClass":
"org.apache.camel.component.keycloak.KeycloakConfiguration",
"configurationField": "configuration", "description": "Time-to-live for cached
introspection results in seconds" },
"ipAddress": { "index": 16, "kind": "parameter", "displayName": "Ip
Address", "group": "common", "label": "", "required": false, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "autowired": false,
"secret": false, "configurationClass":
"org.apache.camel.component.keycloak.KeycloakConfiguration",
"configurationField": "configuration", "description": "Filter events by IP
address" },
"keycloakClient": { "index": 17, "kind": "parameter", "displayName":
"Keycloak Client", "group": "common", "label": "", "required": false, "type":
"object", "javaType": "org.keycloak.admin.client.Keycloak", "deprecated":
false, "deprecationNote": "", "autowired": true, "secret": false,
"configurationClass":
"org.apache.camel.component.keycloak.KeycloakConfiguration",
"configurationField": "configuration", "description": "To use an existing
configured Keycloak admin client" },
- "maxResults": { "index": 18, "kind": "parameter", "displayName": "Max
Results", "group": "common", "label": "", "required": false, "type": "integer",
"javaType": "int", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": 100, "configurationClass":
"org.apache.camel.component.keycloak.KeycloakConfiguration",
"configurationField": "configuration", "description": "Maximum number of events
to retrieve per poll" },
+ "maxResults": { "index": 18, "kind": "parameter", "displayName": "Max
Results", "group": "common", "label": "", "required": false, "type": "integer",
"javaType": "int", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": 100, "configurationClass":
"org.apache.camel.component.keycloak.KeycloakConfiguration",
"configurationField": "configuration", "description": "The number of events the
first poll retrieves, and the page size of the later polls, which read every
[...]
"operation": { "index": 19, "kind": "parameter", "displayName":
"Operation", "group": "common", "label": "", "required": false, "type": "enum",
"javaType": "org.apache.camel.component.keycloak.KeycloakOperations", "enum": [
"createRealm", "deleteRealm", "getRealm", "updateRealm", "createUser",
"deleteUser", "getUser", "updateUser", "listUsers", "searchUsers",
"createRole", "deleteRole", "getRole", "updateRole", "listRoles",
"assignRoleToUser", "removeRoleFromUser", "getUserRoles", "c [...]
"operationTypes": { "index": 20, "kind": "parameter", "displayName":
"Operation Types", "group": "common", "label": "", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.keycloak.KeycloakConfiguration",
"configurationField": "configuration", "description": "Filter admin events by
operation types (comma-separated list, e.g., CREATE,UPDATE,DELETE)" },
"password": { "index": 21, "kind": "parameter", "displayName": "Password",
"group": "common", "label": "", "required": false, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "autowired": false,
"secret": true, "security": "secret", "configurationClass":
"org.apache.camel.component.keycloak.KeycloakConfiguration",
"configurationField": "configuration", "description": "Keycloak password" },
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/keycloak-consumer.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/keycloak-consumer.adoc
index f2f0bb098fa4..7764059ee2e0 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/keycloak-consumer.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/keycloak-consumer.adoc
@@ -14,6 +14,15 @@ The consumer supports two types of events:
The consumer uses a polling mechanism with fingerprint-based deduplication to
ensure events are not processed multiple times.
+Keycloak returns the most recent events first. The first poll reads the latest
`maxResults` events; every later poll
+reads all the events since the previous poll, page by page (`maxResults` is
then the page size), so no event is skipped
+when more than `maxResults` events happened between two polls. When the first
poll finds no event (a new realm, or a
+filter that has not matched yet), the later polls read every event as well.
+
+A poll after a long outage of a busy realm can therefore be large: it reads
all the events that Keycloak still keeps
+(see the event expiration of the realm), and creates their exchanges before it
routes them. `maxResults` does not limit
+the size of such a poll, only the size of each page.
+
== Configuration
The consumer supports the same authentication methods as the producer (access
token, refresh token, username/password, or client credentials):
@@ -664,7 +673,7 @@ The consumer supports the following configuration options:
| `realm` | | The Keycloak realm to consume events from (required)
| `eventType` | events | Type of events to consume: `events` or `admin-events`
-| `maxResults` | 100 | Maximum number of events to retrieve per poll
+| `maxResults` | 100 | Maximum number of events of the first poll, and the
page size of the later polls, which read every event since the previous poll
| `first` | 0 | Offset for pagination (first result index)
| `initialDelay` | 1000 | Delay before first poll (milliseconds)
| `delay` | 500 | Delay between polls (milliseconds)
diff --git
a/components/camel-keycloak/src/generated/resources/META-INF/org/apache/camel/component/keycloak/keycloak.json
b/components/camel-keycloak/src/generated/resources/META-INF/org/apache/camel/component/keycloak/keycloak.json
index 9464ca5e57f1..91c717d8cfab 100644
---
a/components/camel-keycloak/src/generated/resources/META-INF/org/apache/camel/component/keycloak/keycloak.json
+++
b/components/camel-keycloak/src/generated/resources/META-INF/org/apache/camel/component/keycloak/keycloak.json
@@ -42,7 +42,7 @@
"introspectionCacheTtl": { "index": 15, "kind": "property", "displayName":
"Introspection Cache Ttl", "group": "common", "label": "", "required": false,
"type": "integer", "javaType": "long", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 60, "configurationClass":
"org.apache.camel.component.keycloak.KeycloakConfiguration",
"configurationField": "configuration", "description": "Time-to-live for cached
introspection results in seconds" },
"ipAddress": { "index": 16, "kind": "property", "displayName": "Ip
Address", "group": "common", "label": "", "required": false, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "autowired": false,
"secret": false, "configurationClass":
"org.apache.camel.component.keycloak.KeycloakConfiguration",
"configurationField": "configuration", "description": "Filter events by IP
address" },
"keycloakClient": { "index": 17, "kind": "property", "displayName":
"Keycloak Client", "group": "common", "label": "", "required": false, "type":
"object", "javaType": "org.keycloak.admin.client.Keycloak", "deprecated":
false, "deprecationNote": "", "autowired": true, "secret": false,
"configurationClass":
"org.apache.camel.component.keycloak.KeycloakConfiguration",
"configurationField": "configuration", "description": "To use an existing
configured Keycloak admin client" },
- "maxResults": { "index": 18, "kind": "property", "displayName": "Max
Results", "group": "common", "label": "", "required": false, "type": "integer",
"javaType": "int", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": 100, "configurationClass":
"org.apache.camel.component.keycloak.KeycloakConfiguration",
"configurationField": "configuration", "description": "Maximum number of events
to retrieve per poll" },
+ "maxResults": { "index": 18, "kind": "property", "displayName": "Max
Results", "group": "common", "label": "", "required": false, "type": "integer",
"javaType": "int", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": 100, "configurationClass":
"org.apache.camel.component.keycloak.KeycloakConfiguration",
"configurationField": "configuration", "description": "The number of events the
first poll retrieves, and the page size of the later polls, which read every e
[...]
"operation": { "index": 19, "kind": "property", "displayName":
"Operation", "group": "common", "label": "", "required": false, "type": "enum",
"javaType": "org.apache.camel.component.keycloak.KeycloakOperations", "enum": [
"createRealm", "deleteRealm", "getRealm", "updateRealm", "createUser",
"deleteUser", "getUser", "updateUser", "listUsers", "searchUsers",
"createRole", "deleteRole", "getRole", "updateRole", "listRoles",
"assignRoleToUser", "removeRoleFromUser", "getUserRoles", "cr [...]
"operationTypes": { "index": 20, "kind": "property", "displayName":
"Operation Types", "group": "common", "label": "", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.keycloak.KeycloakConfiguration",
"configurationField": "configuration", "description": "Filter admin events by
operation types (comma-separated list, e.g., CREATE,UPDATE,DELETE)" },
"password": { "index": 21, "kind": "property", "displayName": "Password",
"group": "common", "label": "", "required": false, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "autowired": false,
"secret": true, "security": "secret", "configurationClass":
"org.apache.camel.component.keycloak.KeycloakConfiguration",
"configurationField": "configuration", "description": "Keycloak password" },
@@ -146,7 +146,7 @@
"introspectionCacheTtl": { "index": 15, "kind": "parameter",
"displayName": "Introspection Cache Ttl", "group": "common", "label": "",
"required": false, "type": "integer", "javaType": "long", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": 60, "configurationClass":
"org.apache.camel.component.keycloak.KeycloakConfiguration",
"configurationField": "configuration", "description": "Time-to-live for cached
introspection results in seconds" },
"ipAddress": { "index": 16, "kind": "parameter", "displayName": "Ip
Address", "group": "common", "label": "", "required": false, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "autowired": false,
"secret": false, "configurationClass":
"org.apache.camel.component.keycloak.KeycloakConfiguration",
"configurationField": "configuration", "description": "Filter events by IP
address" },
"keycloakClient": { "index": 17, "kind": "parameter", "displayName":
"Keycloak Client", "group": "common", "label": "", "required": false, "type":
"object", "javaType": "org.keycloak.admin.client.Keycloak", "deprecated":
false, "deprecationNote": "", "autowired": true, "secret": false,
"configurationClass":
"org.apache.camel.component.keycloak.KeycloakConfiguration",
"configurationField": "configuration", "description": "To use an existing
configured Keycloak admin client" },
- "maxResults": { "index": 18, "kind": "parameter", "displayName": "Max
Results", "group": "common", "label": "", "required": false, "type": "integer",
"javaType": "int", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": 100, "configurationClass":
"org.apache.camel.component.keycloak.KeycloakConfiguration",
"configurationField": "configuration", "description": "Maximum number of events
to retrieve per poll" },
+ "maxResults": { "index": 18, "kind": "parameter", "displayName": "Max
Results", "group": "common", "label": "", "required": false, "type": "integer",
"javaType": "int", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": 100, "configurationClass":
"org.apache.camel.component.keycloak.KeycloakConfiguration",
"configurationField": "configuration", "description": "The number of events the
first poll retrieves, and the page size of the later polls, which read every
[...]
"operation": { "index": 19, "kind": "parameter", "displayName":
"Operation", "group": "common", "label": "", "required": false, "type": "enum",
"javaType": "org.apache.camel.component.keycloak.KeycloakOperations", "enum": [
"createRealm", "deleteRealm", "getRealm", "updateRealm", "createUser",
"deleteUser", "getUser", "updateUser", "listUsers", "searchUsers",
"createRole", "deleteRole", "getRole", "updateRole", "listRoles",
"assignRoleToUser", "removeRoleFromUser", "getUserRoles", "c [...]
"operationTypes": { "index": 20, "kind": "parameter", "displayName":
"Operation Types", "group": "common", "label": "", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.keycloak.KeycloakConfiguration",
"configurationField": "configuration", "description": "Filter admin events by
operation types (comma-separated list, e.g., CREATE,UPDATE,DELETE)" },
"password": { "index": 21, "kind": "parameter", "displayName": "Password",
"group": "common", "label": "", "required": false, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "autowired": false,
"secret": true, "security": "secret", "configurationClass":
"org.apache.camel.component.keycloak.KeycloakConfiguration",
"configurationField": "configuration", "description": "Keycloak password" },
diff --git a/components/camel-keycloak/src/main/docs/keycloak-consumer.adoc
b/components/camel-keycloak/src/main/docs/keycloak-consumer.adoc
index f2f0bb098fa4..7764059ee2e0 100644
--- a/components/camel-keycloak/src/main/docs/keycloak-consumer.adoc
+++ b/components/camel-keycloak/src/main/docs/keycloak-consumer.adoc
@@ -14,6 +14,15 @@ The consumer supports two types of events:
The consumer uses a polling mechanism with fingerprint-based deduplication to
ensure events are not processed multiple times.
+Keycloak returns the most recent events first. The first poll reads the latest
`maxResults` events; every later poll
+reads all the events since the previous poll, page by page (`maxResults` is
then the page size), so no event is skipped
+when more than `maxResults` events happened between two polls. When the first
poll finds no event (a new realm, or a
+filter that has not matched yet), the later polls read every event as well.
+
+A poll after a long outage of a busy realm can therefore be large: it reads
all the events that Keycloak still keeps
+(see the event expiration of the realm), and creates their exchanges before it
routes them. `maxResults` does not limit
+the size of such a poll, only the size of each page.
+
== Configuration
The consumer supports the same authentication methods as the producer (access
token, refresh token, username/password, or client credentials):
@@ -664,7 +673,7 @@ The consumer supports the following configuration options:
| `realm` | | The Keycloak realm to consume events from (required)
| `eventType` | events | Type of events to consume: `events` or `admin-events`
-| `maxResults` | 100 | Maximum number of events to retrieve per poll
+| `maxResults` | 100 | Maximum number of events of the first poll, and the
page size of the later polls, which read every event since the previous poll
| `first` | 0 | Offset for pagination (first result index)
| `initialDelay` | 1000 | Delay before first poll (milliseconds)
| `delay` | 500 | Delay between polls (milliseconds)
diff --git
a/components/camel-keycloak/src/main/java/org/apache/camel/component/keycloak/KeycloakConfiguration.java
b/components/camel-keycloak/src/main/java/org/apache/camel/component/keycloak/KeycloakConfiguration.java
index 52a8b1bdca8e..db6d5bbd1a8e 100644
---
a/components/camel-keycloak/src/main/java/org/apache/camel/component/keycloak/KeycloakConfiguration.java
+++
b/components/camel-keycloak/src/main/java/org/apache/camel/component/keycloak/KeycloakConfiguration.java
@@ -71,7 +71,9 @@ public class KeycloakConfiguration implements Cloneable {
@UriParam(description = "Type of events to consume: events or
admin-events", defaultValue = "events")
private String eventType = "events";
- @UriParam(description = "Maximum number of events to retrieve per poll",
defaultValue = "100")
+ @UriParam(description = "The number of events the first poll retrieves,
and the page size of the later polls,"
+ + " which read every event since the previous
poll",
+ defaultValue = "100")
private int maxResults = 100;
@UriParam(description = "Offset for pagination (first result index)",
defaultValue = "0")
@@ -281,7 +283,8 @@ public class KeycloakConfiguration implements Cloneable {
}
/**
- * Maximum number of events to retrieve per poll. Default is 100.
+ * The number of events the first poll retrieves, and the page size of the
later polls, which read every event since
+ * the previous poll. Default is 100.
*/
public void setMaxResults(int maxResults) {
this.maxResults = maxResults;
diff --git
a/components/camel-keycloak/src/main/java/org/apache/camel/component/keycloak/KeycloakConsumer.java
b/components/camel-keycloak/src/main/java/org/apache/camel/component/keycloak/KeycloakConsumer.java
index bcda15834994..063f81a31d00 100644
---
a/components/camel-keycloak/src/main/java/org/apache/camel/component/keycloak/KeycloakConsumer.java
+++
b/components/camel-keycloak/src/main/java/org/apache/camel/component/keycloak/KeycloakConsumer.java
@@ -22,6 +22,8 @@ import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
import java.util.Set;
+import java.util.function.BiFunction;
+import java.util.function.ToLongFunction;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
@@ -119,16 +121,20 @@ public class KeycloakConsumer extends
ScheduledBatchPollingConsumer {
// Don't use dateFrom filter to avoid issues with inclusive/exclusive
semantics
// Instead rely on client-side fingerprint-based deduplication
List<String> eventTypes = parseCommaSeparatedList(config.getTypes());
- List<EventRepresentation> events =
keycloakClient.realm(realm).getEvents(
+ List<EventRepresentation> events = fetchNewEvents((first, max) ->
keycloakClient.realm(realm).getEvents(
eventTypes, // types
config.getClient(), // client
config.getUser(), // user
config.getDateFrom(), // dateFrom
config.getDateTo(), // dateTo
config.getIpAddress(), // ipAddress
- config.getFirst(), // first
- config.getMaxResults() // max
- );
+ first, // first
+ max // max
+ ), EventRepresentation::getTime);
+ if (events == null) {
+ // no answer: the next poll is still a first poll, and the
checkpoint does not move
+ return new LinkedList<>();
+ }
Queue<Exchange> queue = new LinkedList<>();
long highestEventTime = lastEventTime != null ? lastEventTime : 0;
@@ -164,12 +170,6 @@ public class KeycloakConsumer extends
ScheduledBatchPollingConsumer {
if (eventTime > highestEventTime) {
highestEventTime = eventTime;
}
-
- // Prevent unbounded memory growth
- if (processedEventFingerprints.size() >
MAX_FINGERPRINT_CACHE_SIZE) {
- processedEventFingerprints.clear();
- LOG.debug("Cleared fingerprint cache due to size limit");
- }
}
// Update lastEventTime if we processed events with newer timestamps
@@ -177,11 +177,64 @@ public class KeycloakConsumer extends
ScheduledBatchPollingConsumer {
forgetOlderFingerprints(highestEventTime);
lastEventTime = highestEventTime;
LOG.debug("Updated last event time to: {}", lastEventTime);
+ } else if (lastEventTime == null && events.isEmpty()) {
+ // no event yet (a new realm, or a filter that has not matched
yet): every event from now on is new, so
+ // the next polls read all of them page by page, and not only the
latest maxResults
+ lastEventTime = 0L;
}
+ limitFingerprintCache();
return queue;
}
+ /**
+ * Reads the events of a poll. Keycloak returns the most recent events
first, and the newest event of a poll becomes
+ * the checkpoint, so the events that do not fit into the first {@code
maxResults} (the older new ones, for example
+ * after Keycloak was unreachable for a while) would never be returned
again. Once a checkpoint is set, the
+ * following pages are read as well, until a page reaches an event older
than the checkpoint or is the last one.
+ * When new events arrive while the pages are read, the next page repeats
events of the previous one; the
+ * fingerprints skip them.
+ *
+ * @return the events, or {@code null} when the first page is a response
without content (the client returns
+ * {@code null} for it): that is no answer, and not a realm
without events, so the caller must neither route
+ * anything nor move the checkpoint
+ */
+ private <T> List<T> fetchNewEvents(BiFunction<Integer, Integer, List<T>>
fetch, ToLongFunction<T> eventTime) {
+ KeycloakConfiguration config = getEndpoint().getConfiguration();
+ int first = config.getFirst();
+ int max = config.getMaxResults();
+ List<T> page = fetch.apply(first, max);
+ if (page == null) {
+ // the client returns null for a response without content
+ return null;
+ }
+ if (lastEventTime == null || max <= 0 || page.size() < max) {
+ return page;
+ }
+ List<T> answer = new ArrayList<>(page);
+ while (page.size() >= max &&
eventTime.applyAsLong(page.get(page.size() - 1)) >= lastEventTime
+ && isRunAllowed()) {
+ first += max;
+ page = fetch.apply(first, max);
+ if (page == null) {
+ break;
+ }
+ answer.addAll(page);
+ }
+ return answer;
+ }
+
+ /**
+ * Prevents unbounded memory growth. Done after a poll, when only the
fingerprints at the checkpoint are left, and
+ * not while the events of a poll are read: those fingerprints are needed
until the end of the poll.
+ */
+ private void limitFingerprintCache() {
+ if (processedEventFingerprints.size() > MAX_FINGERPRINT_CACHE_SIZE) {
+ processedEventFingerprints.clear();
+ LOG.debug("Cleared fingerprint cache due to size limit");
+ }
+ }
+
/**
* Forgets the fingerprints of the events older than the given time, to
free memory. The fingerprints of the events
* at that time are kept: the next poll returns those events again (the
query has no lower time bound, dateFrom is a
@@ -214,18 +267,23 @@ public class KeycloakConsumer extends
ScheduledBatchPollingConsumer {
// Don't use dateFrom filter to avoid issues with inclusive/exclusive
semantics
// Instead rely on client-side fingerprint-based deduplication
List<String> adminOperationTypes =
parseCommaSeparatedList(config.getOperationTypes());
- List<AdminEventRepresentation> adminEvents =
keycloakClient.realm(realm).getAdminEvents(
- adminOperationTypes, // operationTypes
- config.getAuthRealmFilter(), // authRealm
- config.getAuthClient(), // authClient
- config.getAuthUser(), // authUser
- config.getAuthIpAddress(), // authIpAddress
- config.getResourcePath(), // resourcePath
- config.getDateFrom(), // dateFrom
- config.getDateTo(), // dateTo
- config.getFirst(), // first
- config.getMaxResults() // max
- );
+ List<AdminEventRepresentation> adminEvents
+ = fetchNewEvents((first, max) ->
keycloakClient.realm(realm).getAdminEvents(
+ adminOperationTypes, // operationTypes
+ config.getAuthRealmFilter(), // authRealm
+ config.getAuthClient(), // authClient
+ config.getAuthUser(), // authUser
+ config.getAuthIpAddress(), // authIpAddress
+ config.getResourcePath(), // resourcePath
+ config.getDateFrom(), // dateFrom
+ config.getDateTo(), // dateTo
+ first, // first
+ max // max
+ ), AdminEventRepresentation::getTime);
+ if (adminEvents == null) {
+ // no answer: the next poll is still a first poll, and the
checkpoint does not move
+ return new LinkedList<>();
+ }
Queue<Exchange> queue = new LinkedList<>();
long highestEventTime = lastEventTime != null ? lastEventTime : 0;
@@ -261,14 +319,6 @@ public class KeycloakConsumer extends
ScheduledBatchPollingConsumer {
if (eventTime > highestEventTime) {
highestEventTime = eventTime;
}
-
- // Prevent unbounded memory growth
- if (processedEventFingerprints.size() >
MAX_FINGERPRINT_CACHE_SIZE) {
- // Keep only recent fingerprints when cache gets too large
- // This is a simple strategy - in production you'd want a more
sophisticated LRU cache
- processedEventFingerprints.clear();
- LOG.debug("Cleared fingerprint cache due to size limit");
- }
}
// Update lastEventTime if we processed events with newer timestamps
@@ -276,7 +326,12 @@ public class KeycloakConsumer extends
ScheduledBatchPollingConsumer {
forgetOlderFingerprints(highestEventTime);
lastEventTime = highestEventTime;
LOG.debug("Updated last admin event time to: {}", lastEventTime);
+ } else if (lastEventTime == null && adminEvents.isEmpty()) {
+ // no event yet (a new realm, or a filter that has not matched
yet): every event from now on is new, so
+ // the next polls read all of them page by page, and not only the
latest maxResults
+ lastEventTime = 0L;
}
+ limitFingerprintCache();
return queue;
}
diff --git
a/components/camel-keycloak/src/test/java/org/apache/camel/component/keycloak/KeycloakConsumerMoreThanMaxResultsTest.java
b/components/camel-keycloak/src/test/java/org/apache/camel/component/keycloak/KeycloakConsumerMoreThanMaxResultsTest.java
new file mode 100644
index 000000000000..baa0062b64d6
--- /dev/null
+++
b/components/camel-keycloak/src/test/java/org/apache/camel/component/keycloak/KeycloakConsumerMoreThanMaxResultsTest.java
@@ -0,0 +1,254 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.keycloak;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.camel.BindToRegistry;
+import org.apache.camel.Exchange;
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+import org.keycloak.admin.client.Keycloak;
+import org.keycloak.admin.client.resource.RealmResource;
+import org.keycloak.representations.idm.AdminEventRepresentation;
+import org.keycloak.representations.idm.EventRepresentation;
+import org.mockito.Mockito;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.when;
+
+/**
+ * More events than maxResults between two polls: Keycloak returns the most
recent events first, so the events that do
+ * not fit into the first maxResults are the older ones, and they must still
be routed.
+ */
+class KeycloakConsumerMoreThanMaxResultsTest extends CamelTestSupport {
+
+ @BindToRegistry("keycloakClient")
+ private final Keycloak keycloakClient = Mockito.mock(Keycloak.class);
+
+ private final RealmResource realmResource =
Mockito.mock(RealmResource.class);
+
+ // the events in the realm, the most recent first as Keycloak returns them
+ private final List<EventRepresentation> events = new ArrayList<>();
+ private final List<AdminEventRepresentation> adminEvents = new
ArrayList<>();
+
+ @Override
+ protected void doPreSetup() {
+ when(keycloakClient.realm(anyString())).thenReturn(realmResource);
+ // honour first and max like Keycloak does
+ when(realmResource.getEvents(any(), any(), any(), any(), any(), any(),
anyInt(), anyInt()))
+ .thenAnswer(invocation -> page(events,
invocation.getArgument(6), invocation.getArgument(7)));
+ when(realmResource.getAdminEvents(any(), any(), any(), any(), any(),
any(), any(), any(), anyInt(), anyInt()))
+ .thenAnswer(invocation -> page(adminEvents,
invocation.getArgument(8), invocation.getArgument(9)));
+ }
+
+ private static <T> List<T> page(List<T> all, Integer first, Integer max) {
+ int from = Math.min(first != null ? first : 0, all.size());
+ int to = max != null ? Math.min(from + max, all.size()) : all.size();
+ return new ArrayList<>(all.subList(from, to));
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("keycloak:events?keycloakClient=#keycloakClient&realm=test-realm&eventType=events"
+ + "&maxResults=2&startScheduler=false")
+ .routeId("events")
+ .to("mock:events");
+
+
from("keycloak:admin-events?keycloakClient=#keycloakClient&realm=test-realm&eventType=admin-events"
+ + "&maxResults=2&startScheduler=false")
+ .routeId("admin-events")
+ .to("mock:admin-events");
+ }
+ };
+ }
+
+ @Test
+ void routesTheEventsBeyondMaxResults() throws Exception {
+ events.add(event(1000, "user-1"));
+ assertEquals(1, poll("events"));
+
+ // three new events between two polls, one more than maxResults
+ events.add(0, event(2000, "user-2"));
+ events.add(0, event(3000, "user-3"));
+ events.add(0, event(4000, "user-4"));
+ poll("events");
+ poll("events");
+
+ MockEndpoint mock = getMockEndpoint("mock:events");
+ mock.expectedMessageCount(4);
+ mock.assertIsSatisfied();
+ List<String> userIds = new ArrayList<>();
+ for (Exchange exchange : mock.getReceivedExchanges()) {
+
userIds.add(exchange.getIn().getBody(EventRepresentation.class).getUserId());
+ }
+ assertEquals(List.of("user-1", "user-2", "user-3", "user-4"),
userIds.stream().sorted().toList());
+ }
+
+ @Test
+ void routesTheAdminEventsBeyondMaxResults() throws Exception {
+ adminEvents.add(adminEvent(1000, "users/1"));
+ assertEquals(1, poll("admin-events"));
+
+ // three new admin events between two polls, one more than maxResults
+ adminEvents.add(0, adminEvent(2000, "users/2"));
+ adminEvents.add(0, adminEvent(3000, "users/3"));
+ adminEvents.add(0, adminEvent(4000, "users/4"));
+ poll("admin-events");
+ poll("admin-events");
+
+ MockEndpoint mock = getMockEndpoint("mock:admin-events");
+ mock.expectedMessageCount(4);
+ mock.assertIsSatisfied();
+ List<String> paths = new ArrayList<>();
+ for (Exchange exchange : mock.getReceivedExchanges()) {
+
paths.add(exchange.getIn().getBody(AdminEventRepresentation.class).getResourcePath());
+ }
+ assertEquals(List.of("users/1", "users/2", "users/3", "users/4"),
paths.stream().sorted().toList());
+ }
+
+ @Test
+ void routesTheEventsBeyondMaxResultsAfterAFirstPollWithoutEvents() throws
Exception {
+ // a new realm, or a filter that has not matched yet
+ assertEquals(0, poll("events"));
+
+ events.add(0, event(2000, "user-2"));
+ events.add(0, event(3000, "user-3"));
+ events.add(0, event(4000, "user-4"));
+ poll("events");
+ poll("events");
+
+ MockEndpoint mock = getMockEndpoint("mock:events");
+ mock.expectedMessageCount(3);
+ mock.assertIsSatisfied();
+ List<String> userIds = new ArrayList<>();
+ for (Exchange exchange : mock.getReceivedExchanges()) {
+
userIds.add(exchange.getIn().getBody(EventRepresentation.class).getUserId());
+ }
+ assertEquals(List.of("user-2", "user-3", "user-4"),
userIds.stream().sorted().toList());
+ }
+
+ @Test
+ void routesTheAdminEventsBeyondMaxResultsAfterAFirstPollWithoutEvents()
throws Exception {
+ assertEquals(0, poll("admin-events"));
+
+ adminEvents.add(0, adminEvent(2000, "users/2"));
+ adminEvents.add(0, adminEvent(3000, "users/3"));
+ adminEvents.add(0, adminEvent(4000, "users/4"));
+ poll("admin-events");
+ poll("admin-events");
+
+ MockEndpoint mock = getMockEndpoint("mock:admin-events");
+ mock.expectedMessageCount(3);
+ mock.assertIsSatisfied();
+ List<String> paths = new ArrayList<>();
+ for (Exchange exchange : mock.getReceivedExchanges()) {
+
paths.add(exchange.getIn().getBody(AdminEventRepresentation.class).getResourcePath());
+ }
+ assertEquals(List.of("users/2", "users/3", "users/4"),
paths.stream().sorted().toList());
+ }
+
+ @Test
+ void newEventsArrivingWhilePagingAreRoutedOnce() throws Exception {
+ events.add(event(1000, "user-1"));
+ assertEquals(1, poll("events"));
+
+ events.add(0, event(2000, "user-2"));
+ events.add(0, event(3000, "user-3"));
+ events.add(0, event(4000, "user-4"));
+ events.add(0, event(5000, "user-5"));
+ events.add(0, event(6000, "user-6"));
+ // a new event arrives after the first page was read: the next pages
are shifted by one
+ int[] calls = { 0 };
+ when(realmResource.getEvents(any(), any(), any(), any(), any(), any(),
anyInt(), anyInt()))
+ .thenAnswer(invocation -> {
+ if (calls[0]++ == 1) {
+ events.add(0, event(7000, "user-7"));
+ }
+ return page(events, invocation.getArgument(6),
invocation.getArgument(7));
+ });
+ poll("events");
+ poll("events");
+ poll("events");
+
+ MockEndpoint mock = getMockEndpoint("mock:events");
+ mock.expectedMessageCount(7);
+ mock.assertIsSatisfied();
+ List<String> userIds = new ArrayList<>();
+ for (Exchange exchange : mock.getReceivedExchanges()) {
+
userIds.add(exchange.getIn().getBody(EventRepresentation.class).getUserId());
+ }
+ assertEquals(List.of("user-1", "user-2", "user-3", "user-4", "user-5",
"user-6", "user-7"),
+ userIds.stream().sorted().toList());
+ }
+
+ @Test
+ void aResponseWithoutContentIsAPollWithoutEvents() throws Exception {
+ // the client returns null for a 2xx response without a content type,
such as 204 No Content
+ when(realmResource.getEvents(any(), any(), any(), any(), any(), any(),
anyInt(), anyInt())).thenReturn(null);
+ when(realmResource.getAdminEvents(any(), any(), any(), any(), any(),
any(), any(), any(), anyInt(), anyInt()))
+ .thenReturn(null);
+
+ assertEquals(0, poll("events"));
+ assertEquals(0, poll("admin-events"));
+
+ // that was no answer, not a realm without events: the next poll is
still a first poll, which routes the
+ // latest maxResults events and not the whole history of the realm
+ doPreSetup();
+ events.add(0, event(2000, "user-2"));
+ events.add(0, event(3000, "user-3"));
+ events.add(0, event(4000, "user-4"));
+ adminEvents.add(0, adminEvent(2000, "users/2"));
+ adminEvents.add(0, adminEvent(3000, "users/3"));
+ adminEvents.add(0, adminEvent(4000, "users/4"));
+ assertEquals(2, poll("events"));
+ assertEquals(2, poll("admin-events"));
+ assertEquals(0, poll("events"));
+ assertEquals(0, poll("admin-events"));
+ }
+
+ private int poll(String routeId) throws Exception {
+ return ((KeycloakConsumer)
context.getRoute(routeId).getConsumer()).poll();
+ }
+
+ private static EventRepresentation event(long time, String userId) {
+ EventRepresentation event = new EventRepresentation();
+ event.setTime(time);
+ event.setType("LOGIN");
+ event.setUserId(userId);
+ return event;
+ }
+
+ private static AdminEventRepresentation adminEvent(long time, String
resourcePath) {
+ AdminEventRepresentation event = new AdminEventRepresentation();
+ event.setTime(time);
+ event.setOperationType("CREATE");
+ event.setResourceType("USER");
+ event.setResourcePath(resourcePath);
+ return event;
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index bcae0bed3e0b..3d848a8480c8 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -1449,6 +1449,14 @@ it stayed started without receiving any event. The
events of the time without a
`docker:events` or `docker:stats` consumer now closes its stream: before, the
stream stayed open, and a route that was
stopped and started again received every event once more for each restart.
+=== camel-keycloak - the consumer reads every event since the previous poll
+
+The `keycloak` consumer (`eventType=events` and `eventType=admin-events`)
reads, after its first poll, all the events
+since the previous poll, page by page, with `maxResults` as the page size.
Before, it read only the latest `maxResults`
+events and moved past the older ones, so when more than `maxResults` events
happened between two polls the older ones
+were never consumed. A poll can now route more than `maxResults` events. The
first poll still reads the latest
+`maxResults` events.
+
=== camel-influxdb2 - a write is synchronous and a rejected write fails the
exchange
The producer wrote with the batching write API of the InfluxDB client: the
point was put into a buffer of the client
diff --git
a/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/KeycloakComponentBuilderFactory.java
b/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/KeycloakComponentBuilderFactory.java
index 0e52a5bdd0a1..6aff0f5efc9c 100644
---
a/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/KeycloakComponentBuilderFactory.java
+++
b/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/KeycloakComponentBuilderFactory.java
@@ -338,7 +338,8 @@ public interface KeycloakComponentBuilderFactory {
/**
- * Maximum number of events to retrieve per poll.
+ * The number of events the first poll retrieves, and the page size of
+ * the later polls, which read every event since the previous poll.
*
* The option is a: <code>int</code> type.
*
diff --git
a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/KeycloakEndpointBuilderFactory.java
b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/KeycloakEndpointBuilderFactory.java
index 91451123d1c9..c066d31f71ef 100644
---
a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/KeycloakEndpointBuilderFactory.java
+++
b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/KeycloakEndpointBuilderFactory.java
@@ -355,7 +355,8 @@ public interface KeycloakEndpointBuilderFactory {
return this;
}
/**
- * Maximum number of events to retrieve per poll.
+ * The number of events the first poll retrieves, and the page size of
+ * the later polls, which read every event since the previous poll.
*
* The option is a: <code>int</code> type.
*
@@ -370,7 +371,8 @@ public interface KeycloakEndpointBuilderFactory {
return this;
}
/**
- * Maximum number of events to retrieve per poll.
+ * The number of events the first poll retrieves, and the page size of
+ * the later polls, which read every event since the previous poll.
*
* The option will be converted to a <code>int</code> type.
*
@@ -1575,7 +1577,8 @@ public interface KeycloakEndpointBuilderFactory {
return this;
}
/**
- * Maximum number of events to retrieve per poll.
+ * The number of events the first poll retrieves, and the page size of
+ * the later polls, which read every event since the previous poll.
*
* The option is a: <code>int</code> type.
*
@@ -1590,7 +1593,8 @@ public interface KeycloakEndpointBuilderFactory {
return this;
}
/**
- * Maximum number of events to retrieve per poll.
+ * The number of events the first poll retrieves, and the page size of
+ * the later polls, which read every event since the previous poll.
*
* The option will be converted to a <code>int</code> type.
*
@@ -2199,7 +2203,8 @@ public interface KeycloakEndpointBuilderFactory {
return this;
}
/**
- * Maximum number of events to retrieve per poll.
+ * The number of events the first poll retrieves, and the page size of
+ * the later polls, which read every event since the previous poll.
*
* The option is a: <code>int</code> type.
*
@@ -2214,7 +2219,8 @@ public interface KeycloakEndpointBuilderFactory {
return this;
}
/**
- * Maximum number of events to retrieve per poll.
+ * The number of events the first poll retrieves, and the page size of
+ * the later polls, which read every event since the previous poll.
*
* The option will be converted to a <code>int</code> type.
*
