gnodet-bot commented on code in PR #29479:
URL: https://github.com/apache/camel/pull/29479#discussion_r4514235467
##########
components/camel-state-store/camel-state-store/src/main/java/org/apache/camel/component/statestore/StateStoreComponent.java:
##########
@@ -49,31 +50,49 @@ public class StateStoreComponent extends DefaultComponent {
+ " registry auto-discovery for endpoints without
an explicit backend.")
private StateStoreBackendFactory backendFactory;
+ @Metadata(description = "Whether to use a store-specific namespace prefix
to isolate keys on a shared backend repository.",
+ defaultValue = "true")
+ private boolean namespaced = true;
+
@Override
protected Endpoint createEndpoint(String uri, String remaining,
Map<String, Object> parameters) throws Exception {
StateStoreEndpoint endpoint = new StateStoreEndpoint(uri, this);
endpoint.setStoreName(remaining);
+ endpoint.setNamespaced(isNamespaced());
setProperties(endpoint, parameters);
return endpoint;
}
+ private static String toNamespace(String storeName) {
+ return "statestore:" + storeName.replace("%", "%25").replace(":",
"%3A");
+ }
+
/**
* Returns the backend for the given store name, creating one if needed.
Uses first-one-wins semantics: if a backend
* already exists for this store name, subsequent calls return the
existing one regardless of the explicit backend
* passed.
*/
KeyValueRepository getOrCreateBackend(String storeName, KeyValueRepository
explicitBackend) {
+ return getOrCreateBackend(storeName, explicitBackend, isNamespaced());
+ }
+
+ KeyValueRepository getOrCreateBackend(String storeName, KeyValueRepository
explicitBackend, boolean namespaced) {
if (explicitBackend != null) {
- KeyValueRepository existing = backends.putIfAbsent(storeName,
explicitBackend);
+ KeyValueRepository target = namespaced
+ ? NamespacedKeyValueRepository.namespaced(explicitBackend,
toNamespace(storeName))
+ : explicitBackend;
+ KeyValueRepository existing = backends.putIfAbsent(storeName,
target);
if (existing != null) {
- if (existing != explicitBackend) {
+ KeyValueRepository existingDelegate
+ = existing instanceof NamespacedKeyValueRepository n ?
n.getDelegate() : existing;
+ if (existingDelegate != explicitBackend) {
LOG.warn("Store '{}' already has a backend ({}). Ignoring
explicit backend=# reference ({}).",
- storeName, existing.getClass().getSimpleName(),
explicitBackend.getClass().getSimpleName());
+ storeName,
existingDelegate.getClass().getSimpleName(),
explicitBackend.getClass().getSimpleName());
}
return existing;
}
- ServiceHelper.startService(explicitBackend);
- return explicitBackend;
+ ServiceHelper.startService(target);
+ return target;
}
return backends.computeIfAbsent(storeName, k -> {
Review Comment:
⚠️ **`backends` map race: mixed `namespaced` settings for the same store
name**
`backends` is keyed by `storeName` only. If
`state-store:orders?namespaced=false` starts first, it stores the raw backend
in `backends["orders"]`. When `state-store:orders?namespaced=true` calls
`computeIfAbsent("orders", ...)`, the existing key is returned immediately —
the lambda is never evaluated — and the second endpoint silently uses an
un-namespaced backend.
Consider keying by `(storeName, namespaced)` pair, or detect the mismatch
(similar to the existing explicit-backend warning at line 89) and throw or warn.
##########
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() {
+ return keys().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 many keys across namespaces, this allocates a full copy just to
count. Consider a streaming count instead:
```suggestion
public int size() {
Set<String> allKeys = delegate.keys();
if (allKeys == null) {
return 0;
}
int count = 0;
for (String k : allKeys) {
if (k != null && k.startsWith(prefix)) {
count++;
}
}
return count;
}
```
##########
components/camel-state-store/camel-state-store/src/main/docs/state-store-component.adoc:
##########
@@ -206,3 +206,18 @@ for the State Store component, the Idempotent Consumer,
the Aggregator, and the
See the KeyValueRepository section in the User Manual for the full list of
available
backends, configuration examples, and instructions for writing a custom
backend.
+
+== Namespace Isolation
+
+When using a shared `KeyValueRepository` (such as a single bean
auto-discovered from the registry or referenced via `backend=#shared`),
`state-store` automatically isolates entries per store name under the prefix
`statestore:<storeName>:`.
+
+This ensures that:
+* Multiple state stores sharing the same backend do not collide on keys.
Review Comment:
💡 **Same AsciiDoc issue: blank line needed before `*` bullets**
```suggestion
This ensures that:
+
* Multiple state stores sharing the same backend do not collide on keys.
```
##########
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) {
Review Comment:
💡 **`getPrefix()` / `getNamespace()` and `setPrefix()` / `setNamespace()`
duplicate each other**
Both getter pairs return the same `prefix` field; `setPrefix()` simply
delegates to `setNamespace()`. This doubles the API surface without adding
capability. Consider keeping only the `namespace` pair as the primary and
removing (or deprecating) the `prefix` aliases.
##########
docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc:
##########
@@ -17,6 +17,17 @@ 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 and State Store - namespace partitioning on shared
backends
+
+In Camel 4.23, `state-store` endpoints stored keys directly into the
underlying `KeyValueRepository` without namespacing. When a single
`KeyValueRepository` was shared between multiple 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.
+
+Starting in 4.24:
+* `StateStoreComponent` automatically namespaces keys on shared backend
repositories (auto-discovered from the registry or referenced via
`backend=#ref`) under `statestore:<storeName>:<key>`. Delimiter characters in
store names are escaped to guarantee that store names sharing a prefix remain
strictly isolated. Operations such as `clear()`, `keys()`, and `size()` are
scoped to the store's namespace.
Review Comment:
💡 **AsciiDoc: blank line required before the first `*` bullet**
AsciiDoc requires a blank line (or list continuation `+`) between a
paragraph and the first `*` list item. Without it, the bullets are folded into
the preceding paragraph as inline text.
```suggestion
Starting in 4.24:
+
* `StateStoreComponent` automatically namespaces keys on shared backend
repositories (auto-discovered from the registry or referenced via
`backend=#ref`) under `statestore:<storeName>:<key>`. Delimiter characters in
store names are escaped to guarantee that store names sharing a prefix remain
strictly isolated. Operations such as `clear()`, `keys()`, and `size()` are
scoped to the store's namespace.
```
--
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]
