Skip to content

CAMEL-20199: Replace synchronized with ReentrantLock in OAuth and security components #25246

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
gnodet merged 1 commit into from
Jul 31, 2026
Merged

CAMEL-20199: Replace synchronized with ReentrantLock in OAuth and security components #25246

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
41 changes: 24 additions & 17 deletions ...src/main/java/org/apache/camel/component/keycloak/security/KeycloakPublicKeyResolver.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
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;
Expand All @@ -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

Expand Down Expand Up @@ -106,24 +108,29 @@ PublicKey selectKey(String kid) throws IOException {
/**
* 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();
}
}

Expand Down
10 changes: 7 additions & 3 deletions .../camel-oauth/src/main/java/org/apache/camel/oauth/DefaultOAuthTokenValidationFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
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;
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -279,8 +280,9 @@ private static JsonObject discoverEndpoints(OAuthTokenValidationConfig config, S
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)) {
Expand Down Expand Up @@ -308,6 +310,8 @@ private static JsonObject discoverEndpoints(OAuthTokenValidationConfig config, S
DISCOVERY_FAILURES.put(discoveryUrl, new FailureRecord(now, e));
throw new OAuthException("Failed to discover OIDC endpoints from " + discoveryUrl, e);
}
} finally {
lock.unlock();
}
}

Expand Down
17 changes: 12 additions & 5 deletions components/camel-oauth/src/main/java/org/apache/camel/oauth/JwksCache.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
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;
Expand All @@ -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() {
Expand All @@ -64,8 +65,9 @@ JWKSet getJwkSet(String jwksEndpoint, long ttlSeconds, int connectTimeoutMs, int
}

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)) {
Expand All @@ -78,12 +80,15 @@ JWKSet refreshJwkSet(String jwksEndpoint, int connectTimeoutMs, int readTimeoutM
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)) {
Expand Down Expand Up @@ -113,6 +118,8 @@ private JWKSet fetchAndCache(String jwksEndpoint, long ttlSeconds, int connectTi
}
throw e;
}
} finally {
lock.unlock();
}
}

Expand Down
66 changes: 37 additions & 29 deletions ...amel-pqc/src/main/java/org/apache/camel/component/pqc/lifecycle/KeyRotationScheduler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
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;

Expand Down Expand Up @@ -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();
Expand Down Expand Up @@ -110,41 +112,47 @@ public void setCamelContext(CamelContext 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
Expand Down