gnodet-bot commented on code in PR #29479: URL: https://github.com/apache/camel/pull/29479#discussion_r4513437125
########## core/camel-support/src/main/java/org/apache/camel/support/NamespacedKeyValueRepository.java: ########## @@ -0,0 +1,268 @@ +/* + * 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.support; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Set; + +import org.apache.camel.spi.Configurer; +import org.apache.camel.spi.KeyValueRepository; +import org.apache.camel.spi.Metadata; +import org.apache.camel.support.service.ServiceHelper; +import org.apache.camel.support.service.ServiceSupport; +import org.apache.camel.util.ObjectHelper; + +/** + * A {@link KeyValueRepository} wrapper that isolates entries by prepending a namespace prefix to all keys. + * <p/> + * This adapter ensures that multiple components, patterns (e.g. Cache EIP, State Store, Aggregator, Idempotent + * Consumer), or named stores sharing a single underlying {@link KeyValueRepository} operate in isolated key partitions + * without key collisions or unintended cross-namespace mutations during operations such as {@link #clear()} or + * {@link #keys()}. + * + * @since 4.24 + */ +@Metadata(label = "bean", + description = "A KeyValueRepository that partitions keys using a namespace prefix.", + annotations = { "interfaceName=org.apache.camel.spi.KeyValueRepository" }) +@Configurer(metadataOnly = true) +public class NamespacedKeyValueRepository extends ServiceSupport implements KeyValueRepository { + + private KeyValueRepository delegate; + private String prefix; + + /** + * Creates an uninitialized namespaced repository adapter. The delegate repository and namespace should be set + * before starting. + */ + public NamespacedKeyValueRepository() { + } + + /** + * Creates a new namespaced wrapper around the given repository. + * + * @param delegate the underlying key-value repository; must not be {@code null} + * @param namespace the namespace or prefix to isolate keys; must not be {@code null} + */ + public NamespacedKeyValueRepository(KeyValueRepository delegate, String namespace) { + this.delegate = ObjectHelper.notNull(delegate, "delegate"); + setNamespace(namespace); + } + + /** + * Creates a new {@link NamespacedKeyValueRepository} wrapping the given repository, or returns the repository if it + * is already wrapped with the identical prefix. + * + * @param repository the key-value repository to wrap + * @param namespace the namespace prefix to apply + * @return the namespaced repository wrapper + */ + public static KeyValueRepository namespaced(KeyValueRepository repository, String namespace) { + if (repository == null) { + return null; + } + ObjectHelper.notNull(namespace, "namespace"); + String p = namespace.endsWith(":") ? namespace : namespace + ":"; + if (repository instanceof NamespacedKeyValueRepository nkvr) { + if (nkvr.getPrefix().equals(p)) { + return repository; + } + } + return new NamespacedKeyValueRepository(repository, p); + } + + /** + * Returns the underlying delegated {@link KeyValueRepository}. + */ + public KeyValueRepository getDelegate() { + return delegate; + } + + /** + * Sets the underlying delegated {@link KeyValueRepository}. + */ + public void setDelegate(KeyValueRepository delegate) { + this.delegate = delegate; + } + + /** + * Returns the namespace prefix applied to all keys. + */ + public String getPrefix() { + return prefix; + } + + /** + * Returns the namespace applied to all keys. + */ + public String getNamespace() { + return prefix; + } + + /** + * Sets the namespace prefix to isolate keys. If the namespace does not end with ':', a trailing colon is added. + */ + public void setNamespace(String namespace) { + ObjectHelper.notNull(namespace, "namespace"); + this.prefix = namespace.endsWith(":") ? namespace : namespace + ":"; + } + + /** + * Sets the namespace prefix to isolate keys. + */ + public void setPrefix(String prefix) { + setNamespace(prefix); + } + + @Override + public Object get(String key) { + ObjectHelper.notNull(delegate, "delegate"); + ObjectHelper.notNull(prefix, "prefix"); + ObjectHelper.notNull(key, "key"); + return delegate.get(prefix + key); + } + + @Override + public Object put(String key, Object value) { + ObjectHelper.notNull(delegate, "delegate"); + ObjectHelper.notNull(prefix, "prefix"); + ObjectHelper.notNull(key, "key"); + return delegate.put(prefix + key, value); + } + + @Override + public Object put(String key, Object value, Duration ttl) { + ObjectHelper.notNull(delegate, "delegate"); + ObjectHelper.notNull(prefix, "prefix"); + ObjectHelper.notNull(key, "key"); + return delegate.put(prefix + key, value, ttl); + } + + @Override + public Object delete(String key) { + ObjectHelper.notNull(delegate, "delegate"); + ObjectHelper.notNull(prefix, "prefix"); + ObjectHelper.notNull(key, "key"); + return delegate.delete(prefix + key); + } + + @Override + public boolean contains(String key) { + ObjectHelper.notNull(delegate, "delegate"); + ObjectHelper.notNull(prefix, "prefix"); + ObjectHelper.notNull(key, "key"); + return delegate.contains(prefix + key); + } + + @Override + public Set<String> keys() { + ObjectHelper.notNull(delegate, "delegate"); + ObjectHelper.notNull(prefix, "prefix"); + Set<String> allKeys = delegate.keys(); + if (allKeys == null || allKeys.isEmpty()) { + return Collections.emptySet(); + } + Set<String> filtered = new LinkedHashSet<>(); + for (String k : allKeys) { + if (k != null && k.startsWith(prefix)) { + filtered.add(k.substring(prefix.length())); + } + } + return Collections.unmodifiableSet(filtered); + } + + @Override + public void clear() { + ObjectHelper.notNull(delegate, "delegate"); + ObjectHelper.notNull(prefix, "prefix"); + Set<String> allKeys = delegate.keys(); + if (allKeys != null && !allKeys.isEmpty()) { + List<String> toDelete = new ArrayList<>(); + for (String k : allKeys) { + if (k != null && k.startsWith(prefix)) { + toDelete.add(k); + } + } + for (String k : toDelete) { + delegate.delete(k); + } + } + } + + @Override + public Object putIfAbsent(String key, Object value) { + ObjectHelper.notNull(delegate, "delegate"); + ObjectHelper.notNull(prefix, "prefix"); + ObjectHelper.notNull(key, "key"); + return delegate.putIfAbsent(prefix + key, value); + } + + @Override + public Object putIfAbsent(String key, Object value, Duration ttl) { + ObjectHelper.notNull(delegate, "delegate"); + ObjectHelper.notNull(prefix, "prefix"); + ObjectHelper.notNull(key, "key"); + return delegate.putIfAbsent(prefix + key, value, ttl); + } + + @Override + public boolean replace(String key, Object expectedOldValue, Object newValue, Duration ttl) { + ObjectHelper.notNull(delegate, "delegate"); + ObjectHelper.notNull(prefix, "prefix"); + ObjectHelper.notNull(key, "key"); + return delegate.replace(prefix + key, expectedOldValue, newValue, ttl); + } + + @Override + public boolean delete(String key, Object expectedValue) { + ObjectHelper.notNull(delegate, "delegate"); + ObjectHelper.notNull(prefix, "prefix"); + ObjectHelper.notNull(key, "key"); + return delegate.delete(prefix + key, expectedValue); + } + + @Override + public int size() { Review Comment: 💡 **`size()` fully materialises all namespace-filtered keys (O(n) allocation)** `keys().size()` fetches all delegate keys, filters them into a `LinkedHashSet`, wraps in `Collections.unmodifiableSet()`, then counts. For backends with large key sets, this allocates a full set just to produce a number. Consider a streaming count to avoid the intermediate allocation: ```suggestion public int size() { Set<String> allKeys = delegate.keys(); if (allKeys == null || allKeys.isEmpty()) return 0; int count = 0; for (String k : allKeys) { if (k != null && k.startsWith(prefix)) count++; } return count; } ``` ########## docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc: ########## @@ -17,6 +17,12 @@ See the xref:camel-upgrade-recipes-tool.adoc[documentation] page for details. The `camel-observability-services` component has been promoted from _Preview_ to _Stable_ support level. +=== KeyValueRepository - namespace partitioning across components and patterns + +In previous versions, `state-store` endpoints and the `Cache` EIP stored keys directly into the underlying `KeyValueRepository` without namespacing. When a single `KeyValueRepository` was shared between state stores, the Cache EIP, or repository adapters (such as `KeyValueAggregationRepository` or `KeyValueIdempotentRepository`), operations on one store (such as `clear()`) or key collisions could affect entries of another component or pattern. Review Comment: ⚠️ **Misleading: `KeyValueIdempotentRepository` is listed as affected but NOT namespaced by this PR** Mentioning `KeyValueIdempotentRepository` and `KeyValueAggregationRepository` in this list implies the collision problem is resolved for them too. It is not — this PR does not namespace either adapter. Suggest clarifying: list these adapters as examples of patterns sharing the same collision risk that are **not** addressed by this PR. If users share a backend with these adapters, they must manually wrap it in `NamespacedKeyValueRepository`. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
