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 4e022b88e4c8 CAMEL-25525: camel-keycloak - Do not route the latest
events of a poll a second time (#27676)
4e022b88e4c8 is described below
commit 4e022b88e4c80b8c11274cc982959800634e58c5
Author: allthingssecurity <[email protected]>
AuthorDate: Sun Oct 11 12:31:47 2026 +0530
CAMEL-25525: camel-keycloak - Do not route the latest events of a poll a
second time (#27676)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/catalog/docs/keycloak-consumer.adoc | 3 +-
.../src/main/docs/keycloak-consumer.adoc | 3 +-
.../camel/component/keycloak/KeycloakConsumer.java | 16 ++-
.../keycloak/KeycloakConsumerLatestEventsTest.java | 159 +++++++++++++++++++++
4 files changed, 175 insertions(+), 6 deletions(-)
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 87dca08674b0..9c0484ac670c 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
@@ -908,6 +908,7 @@ Before consuming events, you must enable event logging in
Keycloak:
The consumer uses fingerprint-based deduplication to prevent processing the
same event multiple times:
* Events are uniquely identified by combining timestamp and event-specific
properties
-* Fingerprints are cached per timestamp and cleared when moving to newer
timestamps
+* Fingerprints of the events with the latest timestamp are kept between polls,
because Keycloak returns those events
+again on the next poll; fingerprints of older events are dropped when a newer
timestamp is seen
* Maximum cache size is 1000 fingerprints to prevent memory issues
* This ensures reliable event processing even with high event volumes
diff --git a/components/camel-keycloak/src/main/docs/keycloak-consumer.adoc
b/components/camel-keycloak/src/main/docs/keycloak-consumer.adoc
index 87dca08674b0..9c0484ac670c 100644
--- a/components/camel-keycloak/src/main/docs/keycloak-consumer.adoc
+++ b/components/camel-keycloak/src/main/docs/keycloak-consumer.adoc
@@ -908,6 +908,7 @@ Before consuming events, you must enable event logging in
Keycloak:
The consumer uses fingerprint-based deduplication to prevent processing the
same event multiple times:
* Events are uniquely identified by combining timestamp and event-specific
properties
-* Fingerprints are cached per timestamp and cleared when moving to newer
timestamps
+* Fingerprints of the events with the latest timestamp are kept between polls,
because Keycloak returns those events
+again on the next poll; fingerprints of older events are dropped when a newer
timestamp is seen
* Maximum cache size is 1000 fingerprints to prevent memory issues
* This ensures reliable event processing even with high event volumes
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 f380ae01b936..bcda15834994 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
@@ -174,8 +174,7 @@ public class KeycloakConsumer extends
ScheduledBatchPollingConsumer {
// Update lastEventTime if we processed events with newer timestamps
if (highestEventTime > (lastEventTime != null ? lastEventTime : 0)) {
- // Clear fingerprints when moving to a new timestamp to free memory
- processedEventFingerprints.clear();
+ forgetOlderFingerprints(highestEventTime);
lastEventTime = highestEventTime;
LOG.debug("Updated last event time to: {}", lastEventTime);
}
@@ -183,6 +182,16 @@ public class KeycloakConsumer extends
ScheduledBatchPollingConsumer {
return queue;
}
+ /**
+ * 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
+ * date), and they must not be routed a second time.
+ */
+ private void forgetOlderFingerprints(long eventTime) {
+ String prefix = eventTime + "|";
+ processedEventFingerprints.removeIf(fingerprint ->
!fingerprint.startsWith(prefix));
+ }
+
private String getEventFingerprint(EventRepresentation event) {
// Create a composite key from event properties to uniquely identify it
return event.getTime() + "|"
@@ -264,8 +273,7 @@ public class KeycloakConsumer extends
ScheduledBatchPollingConsumer {
// Update lastEventTime if we processed events with newer timestamps
if (highestEventTime > (lastEventTime != null ? lastEventTime : 0)) {
- // Clear fingerprints when moving to a new timestamp to free memory
- processedEventFingerprints.clear();
+ forgetOlderFingerprints(highestEventTime);
lastEventTime = highestEventTime;
LOG.debug("Updated last admin event time to: {}", lastEventTime);
}
diff --git
a/components/camel-keycloak/src/test/java/org/apache/camel/component/keycloak/KeycloakConsumerLatestEventsTest.java
b/components/camel-keycloak/src/test/java/org/apache/camel/component/keycloak/KeycloakConsumerLatestEventsTest.java
new file mode 100644
index 000000000000..26cfe33f1c31
--- /dev/null
+++
b/components/camel-keycloak/src/test/java/org/apache/camel/component/keycloak/KeycloakConsumerLatestEventsTest.java
@@ -0,0 +1,159 @@
+/*
+ * 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;
+
+/**
+ * Keycloak returns the events of the latest timestamp again on the next poll,
so the consumer must remember that it
+ * routed them when it moves its checkpoint to that timestamp.
+ */
+class KeycloakConsumerLatestEventsTest 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);
+ when(realmResource.getEvents(any(), any(), any(), any(), any(), any(),
anyInt(), anyInt()))
+ .thenAnswer(invocation -> new ArrayList<>(events));
+ when(realmResource.getAdminEvents(any(), any(), any(), any(), any(),
any(), any(), any(), anyInt(), anyInt()))
+ .thenAnswer(invocation -> new ArrayList<>(adminEvents));
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("keycloak:events?keycloakClient=#keycloakClient&realm=test-realm&eventType=events"
+ + "&startScheduler=false")
+ .routeId("events")
+ .to("mock:events");
+
+
from("keycloak:admin-events?keycloakClient=#keycloakClient&realm=test-realm&eventType=admin-events"
+ + "&startScheduler=false")
+ .routeId("admin-events")
+ .to("mock:admin-events");
+ }
+ };
+ }
+
+ @Test
+ void routesEachEventOnce() throws Exception {
+ events.add(event(2000, "LOGOUT", "user-456"));
+ events.add(event(1000, "LOGIN", "user-123"));
+
+ assertEquals(2, poll("events"));
+ assertEquals(0, poll("events"), "the events of the first poll must not
be routed again");
+
+ // a new event, and a second one at the same time as the previous
latest event
+ events.add(0, event(3000, "LOGIN", "user-789"));
+ events.add(1, event(2000, "LOGIN", "user-000"));
+ assertEquals(2, poll("events"));
+ assertEquals(0, poll("events"), "the events of the second poll must
not be routed again");
+
+ MockEndpoint mock = getMockEndpoint("mock:events");
+ mock.expectedMessageCount(4);
+ mock.assertIsSatisfied();
+ assertEquals(List.of("user-456", "user-123", "user-789", "user-000"),
userIds(mock));
+ }
+
+ @Test
+ void routesEachAdminEventOnce() throws Exception {
+ adminEvents.add(adminEvent(2000, "UPDATE", "users/2"));
+ adminEvents.add(adminEvent(1000, "CREATE", "users/1"));
+
+ assertEquals(2, poll("admin-events"));
+ assertEquals(0, poll("admin-events"), "the admin events of the first
poll must not be routed again");
+
+ // a new admin event, and a second one at the same time as the
previous latest admin event
+ adminEvents.add(0, adminEvent(3000, "DELETE", "users/1"));
+ adminEvents.add(1, adminEvent(2000, "CREATE", "users/3"));
+ assertEquals(2, poll("admin-events"));
+ assertEquals(0, poll("admin-events"), "the admin events of the second
poll must not be routed again");
+
+ MockEndpoint mock = getMockEndpoint("mock:admin-events");
+ mock.expectedMessageCount(4);
+ mock.assertIsSatisfied();
+ assertEquals(List.of("users/2", "users/1", "users/1", "users/3"),
resourcePaths(mock));
+ }
+
+ private int poll(String routeId) throws Exception {
+ return ((KeycloakConsumer)
context.getRoute(routeId).getConsumer()).poll();
+ }
+
+ private static List<String> userIds(MockEndpoint mock) {
+ List<String> answer = new ArrayList<>();
+ for (Exchange exchange : mock.getReceivedExchanges()) {
+
answer.add(exchange.getIn().getBody(EventRepresentation.class).getUserId());
+ }
+ return answer;
+ }
+
+ private static List<String> resourcePaths(MockEndpoint mock) {
+ List<String> answer = new ArrayList<>();
+ for (Exchange exchange : mock.getReceivedExchanges()) {
+
answer.add(exchange.getIn().getBody(AdminEventRepresentation.class).getResourcePath());
+ }
+ return answer;
+ }
+
+ private static EventRepresentation event(long time, String type, String
userId) {
+ EventRepresentation event = new EventRepresentation();
+ event.setTime(time);
+ event.setType(type);
+ event.setUserId(userId);
+ return event;
+ }
+
+ private static AdminEventRepresentation adminEvent(long time, String
operationType, String resourcePath) {
+ AdminEventRepresentation event = new AdminEventRepresentation();
+ event.setTime(time);
+ event.setOperationType(operationType);
+ event.setResourceType("USER");
+ event.setResourcePath(resourcePath);
+ return event;
+ }
+}
