This is an automated email from the ASF dual-hosted git repository.
gnodet 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 3e0b456f3561 CAMEL-20199: Replace synchronized with ReentrantLock in
OAuth and security components (#25246)
3e0b456f3561 is described below
commit 3e0b456f35617c0b5ce0e153ee4b1e1a26f4d39f
Author: Guillaume Nodet <[email protected]>
AuthorDate: Fri Jul 31 07:35:08 2026 +0200
CAMEL-20199: Replace synchronized with ReentrantLock in OAuth and security
components (#25246)
CAMEL-20199: Replace synchronized with ReentrantLock in OAuth and security
components
Convert synchronized blocks to ReentrantLock in camel-oauth
(DefaultOAuthTokenValidationFactory, JwksCache), camel-keycloak
(KeycloakPublicKeyResolver), and camel-pqc (KeyRotationScheduler)
to eliminate virtual thread pinning.
Co-Authored-By: Claude Opus 4.6 <[email protected]>
---
.../security/KeycloakPublicKeyResolver.java | 41 ++++++++------
.../oauth/DefaultOAuthTokenValidationFactory.java | 10 +++-
.../java/org/apache/camel/oauth/JwksCache.java | 17 ++++--
.../pqc/lifecycle/KeyRotationScheduler.java | 66 ++++++++++++----------
4 files changed, 80 insertions(+), 54 deletions(-)
diff --git
a/components/camel-keycloak/src/main/java/org/apache/camel/component/keycloak/security/KeycloakPublicKeyResolver.java
b/components/camel-keycloak/src/main/java/org/apache/camel/component/keycloak/security/KeycloakPublicKeyResolver.java
index 8381685047b9..745a92b0c69b 100644
---
a/components/camel-keycloak/src/main/java/org/apache/camel/component/keycloak/security/KeycloakPublicKeyResolver.java
+++
b/components/camel-keycloak/src/main/java/org/apache/camel/component/keycloak/security/KeycloakPublicKeyResolver.java
@@ -27,6 +27,7 @@ import java.util.Base64;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.locks.ReentrantLock;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.hc.client5.http.classic.methods.HttpGet;
@@ -46,6 +47,7 @@ public class KeycloakPublicKeyResolver {
private final String serverUrl;
private final String realm;
private final Map<String, PublicKey> keyCache = new ConcurrentHashMap<>();
+ private final ReentrantLock lock = new ReentrantLock();
private volatile long lastRefreshTime = 0;
private static final long CACHE_REFRESH_INTERVAL_MS = 300_000; // 5 minutes
@@ -106,24 +108,29 @@ public class KeycloakPublicKeyResolver {
/**
* Refreshes the public keys from the JWKS endpoint.
*/
- public synchronized void refreshKeys() throws IOException {
- String jwksUrl =
String.format("%s/realms/%s/protocol/openid-connect/certs", serverUrl, realm);
- LOG.debug("Fetching public keys from: {}", jwksUrl);
-
- try (CloseableHttpClient httpClient = HttpClients.createDefault()) {
- HttpGet request = new HttpGet(jwksUrl);
-
- String responseBody = httpClient.execute(request, response -> {
- int statusCode = response.getCode();
- if (statusCode != 200) {
- throw new IOException("Failed to fetch JWKS: HTTP " +
statusCode);
- }
- return EntityUtils.toString(response.getEntity());
- });
+ public void refreshKeys() throws IOException {
+ lock.lock();
+ try {
+ String jwksUrl =
String.format("%s/realms/%s/protocol/openid-connect/certs", serverUrl, realm);
+ LOG.debug("Fetching public keys from: {}", jwksUrl);
+
+ try (CloseableHttpClient httpClient = HttpClients.createDefault())
{
+ HttpGet request = new HttpGet(jwksUrl);
+
+ String responseBody = httpClient.execute(request, response -> {
+ int statusCode = response.getCode();
+ if (statusCode != 200) {
+ throw new IOException("Failed to fetch JWKS: HTTP " +
statusCode);
+ }
+ return EntityUtils.toString(response.getEntity());
+ });
- parseJwks(responseBody);
- lastRefreshTime = System.currentTimeMillis();
- LOG.debug("Successfully loaded {} public keys from JWKS endpoint",
keyCache.size());
+ parseJwks(responseBody);
+ lastRefreshTime = System.currentTimeMillis();
+ LOG.debug("Successfully loaded {} public keys from JWKS
endpoint", keyCache.size());
+ }
+ } finally {
+ lock.unlock();
}
}
diff --git
a/components/camel-oauth/src/main/java/org/apache/camel/oauth/DefaultOAuthTokenValidationFactory.java
b/components/camel-oauth/src/main/java/org/apache/camel/oauth/DefaultOAuthTokenValidationFactory.java
index 07eaecc907ea..318e87c96ace 100644
---
a/components/camel-oauth/src/main/java/org/apache/camel/oauth/DefaultOAuthTokenValidationFactory.java
+++
b/components/camel-oauth/src/main/java/org/apache/camel/oauth/DefaultOAuthTokenValidationFactory.java
@@ -27,6 +27,7 @@ import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.locks.ReentrantLock;
import java.util.function.LongSupplier;
import com.google.gson.JsonArray;
@@ -58,7 +59,7 @@ public class DefaultOAuthTokenValidationFactory implements
OAuthTokenValidationF
private static final long MIN_DISCOVERY_RETRY_INTERVAL_MILLIS = 30_000L;
private static final ConcurrentMap<String, DiscoveryCacheEntry>
DISCOVERY_CACHE = new ConcurrentHashMap<>();
private static final ConcurrentMap<String, FailureRecord>
DISCOVERY_FAILURES = new ConcurrentHashMap<>();
- private static final ConcurrentMap<String, Object> DISCOVERY_LOCKS = new
ConcurrentHashMap<>();
+ private static final ConcurrentMap<String, ReentrantLock> DISCOVERY_LOCKS
= new ConcurrentHashMap<>();
private static volatile LongSupplier currentTimeMillis =
System::currentTimeMillis;
@Override
@@ -279,8 +280,9 @@ public class DefaultOAuthTokenValidationFactory implements
OAuthTokenValidationF
return cached.json;
}
- Object lock = DISCOVERY_LOCKS.computeIfAbsent(discoveryUrl, key -> new
Object());
- synchronized (lock) {
+ ReentrantLock lock = DISCOVERY_LOCKS.computeIfAbsent(discoveryUrl, key
-> new ReentrantLock());
+ lock.lock();
+ try {
now = now();
cached = DISCOVERY_CACHE.get(discoveryUrl);
if (cached != null &&
!cached.isExpired(config.getOidcDiscoveryCacheTtlSeconds(), now)) {
@@ -308,6 +310,8 @@ public class DefaultOAuthTokenValidationFactory implements
OAuthTokenValidationF
DISCOVERY_FAILURES.put(discoveryUrl, new FailureRecord(now,
e));
throw new OAuthException("Failed to discover OIDC endpoints
from " + discoveryUrl, e);
}
+ } finally {
+ lock.unlock();
}
}
diff --git
a/components/camel-oauth/src/main/java/org/apache/camel/oauth/JwksCache.java
b/components/camel-oauth/src/main/java/org/apache/camel/oauth/JwksCache.java
index fef379ce1e74..f781b17f7cc9 100644
--- a/components/camel-oauth/src/main/java/org/apache/camel/oauth/JwksCache.java
+++ b/components/camel-oauth/src/main/java/org/apache/camel/oauth/JwksCache.java
@@ -20,6 +20,7 @@ import java.net.URI;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.locks.ReentrantLock;
import java.util.function.LongSupplier;
import com.nimbusds.jose.jwk.JWKSet;
@@ -45,7 +46,7 @@ final class JwksCache {
private final ConcurrentMap<String, CacheEntry> cache = new
ConcurrentHashMap<>();
private final ConcurrentMap<String, Long> forcedRefreshAttempts = new
ConcurrentHashMap<>();
private final ConcurrentMap<String, FailureRecord> normalRefreshFailures =
new ConcurrentHashMap<>();
- private final ConcurrentMap<String, Object> refreshLocks = new
ConcurrentHashMap<>();
+ private final ConcurrentMap<String, ReentrantLock> refreshLocks = new
ConcurrentHashMap<>();
private volatile LongSupplier currentTimeMillis =
System::currentTimeMillis;
private JwksCache() {
@@ -64,8 +65,9 @@ final class JwksCache {
}
JWKSet refreshJwkSet(String jwksEndpoint, int connectTimeoutMs, int
readTimeoutMs) {
- Object lock = refreshLocks.computeIfAbsent(jwksEndpoint, key -> new
Object());
- synchronized (lock) {
+ ReentrantLock lock = refreshLocks.computeIfAbsent(jwksEndpoint, key ->
new ReentrantLock());
+ lock.lock();
+ try {
CacheEntry existing = cache.get(jwksEndpoint);
long now = now();
if (existing != null && !canAttemptRefresh(jwksEndpoint, now)) {
@@ -78,12 +80,15 @@ final class JwksCache {
normalRefreshFailures.remove(jwksEndpoint);
cache.put(jwksEndpoint, refreshed);
return refreshed.jwkSet;
+ } finally {
+ lock.unlock();
}
}
private JWKSet fetchAndCache(String jwksEndpoint, long ttlSeconds, int
connectTimeoutMs, int readTimeoutMs) {
- Object lock = refreshLocks.computeIfAbsent(jwksEndpoint, key -> new
Object());
- synchronized (lock) {
+ ReentrantLock lock = refreshLocks.computeIfAbsent(jwksEndpoint, key ->
new ReentrantLock());
+ lock.lock();
+ try {
CacheEntry existing = cache.get(jwksEndpoint);
long now = now();
if (existing != null && !existing.isExpired(ttlSeconds, now)) {
@@ -113,6 +118,8 @@ final class JwksCache {
}
throw e;
}
+ } finally {
+ lock.unlock();
}
}
diff --git
a/components/camel-pqc/src/main/java/org/apache/camel/component/pqc/lifecycle/KeyRotationScheduler.java
b/components/camel-pqc/src/main/java/org/apache/camel/component/pqc/lifecycle/KeyRotationScheduler.java
index d936576a43b3..93a319f86347 100644
---
a/components/camel-pqc/src/main/java/org/apache/camel/component/pqc/lifecycle/KeyRotationScheduler.java
+++
b/components/camel-pqc/src/main/java/org/apache/camel/component/pqc/lifecycle/KeyRotationScheduler.java
@@ -25,6 +25,7 @@ import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Function;
import java.util.function.Predicate;
@@ -79,6 +80,7 @@ public class KeyRotationScheduler extends ServiceSupport
implements CamelContext
private Function<KeyMetadata, String> keyIdStrategy =
KeyRotationScheduler::defaultRotatedKeyId;
private KeyRotationListener listener;
+ private final ReentrantLock lock = new ReentrantLock();
private final AtomicLong checksPerformed = new AtomicLong();
private final AtomicLong rotationsPerformed = new AtomicLong();
private final AtomicLong rotationFailures = new AtomicLong();
@@ -110,41 +112,47 @@ public class KeyRotationScheduler extends ServiceSupport
implements CamelContext
*
* @return the number of keys rotated during this pass
*/
- public synchronized int checkAndRotate() {
- checksPerformed.incrementAndGet();
- lastCheckAt = Instant.now();
-
- List<KeyMetadata> keys;
+ public int checkAndRotate() {
+ lock.lock();
try {
- keys = keyManager.listKeys();
- } catch (Exception e) {
- LOG.warn("Failed to list keys during rotation check", e);
- notifyError(null, e);
- return 0;
- }
+ checksPerformed.incrementAndGet();
+ lastCheckAt = Instant.now();
- int rotated = 0;
- for (KeyMetadata metadata : keys) {
- if (metadata == null || !keyFilter.test(metadata)) {
- continue;
- }
- String keyId = metadata.getKeyId();
+ List<KeyMetadata> keys;
try {
- if (keyManager.needsRotation(keyId, maxKeyAge, maxKeyUsage)) {
- String newKeyId = keyIdStrategy.apply(metadata);
- LOG.info("Rotating PQC key '{}' -> '{}' (algorithm={})",
keyId, newKeyId, metadata.getAlgorithm());
- KeyPair newKeyPair = keyManager.rotateKey(keyId, newKeyId,
metadata.getAlgorithm());
- rotationsPerformed.incrementAndGet();
- rotated++;
- notifyRotated(keyId, newKeyId, metadata, newKeyPair);
- }
+ keys = keyManager.listKeys();
} catch (Exception e) {
- rotationFailures.incrementAndGet();
- LOG.warn("Failed to rotate PQC key '{}'", keyId, e);
- notifyError(keyId, e);
+ LOG.warn("Failed to list keys during rotation check", e);
+ notifyError(null, e);
+ return 0;
+ }
+
+ int rotated = 0;
+ for (KeyMetadata metadata : keys) {
+ if (metadata == null || !keyFilter.test(metadata)) {
+ continue;
+ }
+ String keyId = metadata.getKeyId();
+ try {
+ if (keyManager.needsRotation(keyId, maxKeyAge,
maxKeyUsage)) {
+ String newKeyId = keyIdStrategy.apply(metadata);
+ LOG.info("Rotating PQC key '{}' -> '{}'
(algorithm={})", keyId, newKeyId,
+ metadata.getAlgorithm());
+ KeyPair newKeyPair = keyManager.rotateKey(keyId,
newKeyId, metadata.getAlgorithm());
+ rotationsPerformed.incrementAndGet();
+ rotated++;
+ notifyRotated(keyId, newKeyId, metadata, newKeyPair);
+ }
+ } catch (Exception e) {
+ rotationFailures.incrementAndGet();
+ LOG.warn("Failed to rotate PQC key '{}'", keyId, e);
+ notifyError(keyId, e);
+ }
}
+ return rotated;
+ } finally {
+ lock.unlock();
}
- return rotated;
}
@Override
