gnodet-bot commented on code in PR #27649:
URL: https://github.com/apache/camel/pull/27649#discussion_r4236902746
##########
components/camel-ai/camel-semantic/src/main/java/org/apache/camel/language/semantic/SemanticLanguage.java:
##########
@@ -672,10 +731,14 @@ private Object evaluate(Exchange exchange, boolean
asPredicate) {
"Unsupported state type for semantic evaluations:
" + names
+ ". Select strings,
maps or lists explicitly");
}
+ exchange.setProperty(SemanticAudit.REFERENCES,
invocations.values().stream()
+
.map(SemanticAudit.Invocation::getEventId).filter(Objects::nonNull).toList());
// Check every group's input before invoking any provider.
for (Group group : current.groups) {
try {
for (var entry : group.evaluations.entrySet()) {
+ active.clear();
+ active.add(invocations.get(entry.getKey()));
Review Comment:
⚠️ **Misleading `not_executed` label on input-validation failure**
When iterating over a group's entries during input validation, `active` is
narrowed to the single current entry (`active.clear()` + `active.add(current
entry)`). If entry B's `validateInput` throws, `finishFailed` is called with
`active = [B]`. Entry A — which already passed validation — is not in `active`,
so it gets labelled `not_executed` / `evaluation_aborted`. But A's input _was_
validated; the failure happened on B. The label is factually wrong, which makes
audit trails misleading when debugging multi-entry group failures.
Consider widening `active` to include all entries that have already passed
validation, or use a dedicated status (e.g. `validation_aborted`) for entries
that were validated but not yet invoked when a sibling failed.
##########
components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/MemorySemanticAuditStore.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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.semantic;
+
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Deque;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.UUID;
+
+import org.apache.camel.support.service.ServiceSupport;
+
+/** Bounded recent history, lost when the context is destroyed. No
persistent-storage guarantee. */
+public final class MemorySemanticAuditStore extends ServiceSupport implements
SemanticAuditSink, SemanticAuditReader {
+ private final int capacity;
+ private final String epoch = UUID.randomUUID().toString();
+ private final Deque<Entry> records = new ArrayDeque<>();
+ private long sequence;
+ private long evicted;
+
+ public MemorySemanticAuditStore(int capacity) {
+ if (capacity < 1 || capacity > 100000) {
+ throw new IllegalArgumentException("Invalid audit capacity");
+ }
+ this.capacity = capacity;
+ }
+
+ @Override
+ public synchronized void append(SemanticAuditRecord record) {
+ if (records.size() == capacity) {
+ records.removeLast();
+ evicted++;
+ }
+ records.addFirst(new Entry(++sequence, record));
+ }
+
+ @Override
+ public synchronized Optional<SemanticAuditRecord> get(String eventId) {
+ return records.stream().map(Entry::record).filter(r ->
r.getEventId().equals(eventId)).findFirst();
+ }
+
+ @Override
+ public synchronized Map<String, SemanticAuditRecord> getAll(List<String>
eventIds) {
+ if (eventIds.size() > 100) {
+ throw new IllegalArgumentException("Too many audit evidence
references");
+ }
+ var selected = new HashSet<>(eventIds);
+ Map<String, SemanticAuditRecord> found = new LinkedHashMap<>();
+ for (Entry entry : records) {
+ if (selected.remove(entry.record.getEventId())) {
+ found.put(entry.record.getEventId(), entry.record);
+ }
+ }
+ return Map.copyOf(found);
+ }
+
+ @Override
+ public synchronized SemanticAuditPage query(SemanticAuditQuery query) {
+ long before = Long.MAX_VALUE;
+ if (query.getCursor() != null) {
+ String[] parts = query.getCursor().split(":", -1);
+ if (parts.length != 3 || !epoch.equals(parts[0])
+ || !Integer.toString(query.getFilters().hashCode() +
Objects.hashCode(query.getSince())).equals(parts[2])) {
Review Comment:
🐛 **Additive hash combination is collision-prone**
`filters.hashCode() + Objects.hashCode(since)` produces a single `int` via
arithmetic addition. Two different `(filters, since)` pairs can produce the
same sum (e.g. filter hash 3 + since hash 5 = filter hash 5 + since hash 3),
causing a stale cursor to pass the integrity check silently instead of throwing
the expected `IllegalArgumentException`.
Use a combined hash instead:
```suggestion
|| !Integer.toString(Objects.hash(query.getFilters(),
query.getSince())).equals(parts[2])) {
```
##########
components/camel-ai/camel-semantic/src/main/java/org/apache/camel/language/semantic/SemanticLanguage.java:
##########
@@ -686,21 +749,31 @@ private Object evaluate(Exchange exchange, boolean
asPredicate) {
invalid);
}
}
+ active.clear();
+ failureReason = "provider_error";
if (batch) {
Map<String, SemanticResult> results = new
LinkedHashMap<>();
Map<String, Object> decisions = new LinkedHashMap<>();
for (Group group : current.groups) {
if (Thread.currentThread().isInterrupted()) {
throw new InterruptedException("Semantic batch
evaluation interrupted");
}
+ active.clear();
+ group.evaluations.keySet().forEach(name ->
active.add(invocations.get(name)));
+ failureReason = "provider_error";
Map<String, SemanticResult> answers =
group.expert.provider.evaluateBatch(group.evaluations, state);
+ failureReason = "invalid_result";
if (answers == null ||
!answers.keySet().equals(group.evaluations.keySet())) {
throw new IllegalArgumentException("Semantic batch
result names must match evaluation names");
}
for (String name : group.evaluations.keySet()) {
decisions.put(name,
validateResult(name, group.expert,
current.operations.get(name), answers.get(name)));
}
+ for (String name : group.evaluations.keySet()) {
+ invocations.get(name).complete("success",
"evaluation_completed", answers.get(name));
+ }
+ active.clear();
Review Comment:
⚠️ **`validateResult` loop runs before `complete("success")` — entry A
marked `failed` if entry B's result is invalid**
In the batch path, `validateResult` is called for all entries in a first
loop (lines 770–772), then `complete("success", ...)` is called in a second
loop (lines 773–775). If `validateResult` succeeds for entry A but throws for
entry B, the exception propagates before the success loop runs. `finishFailed`
is then called with `active` containing all entries of the group, so entry A —
whose result _was_ valid — is marked `failed` instead of `success`.
Fix by interleaving: call `complete("success", ...)` for each entry
immediately after its `validateResult` succeeds, before moving to the next
entry.
##########
components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/MemorySemanticAuditStore.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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.semantic;
+
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Deque;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.UUID;
+
+import org.apache.camel.support.service.ServiceSupport;
+
+/** Bounded recent history, lost when the context is destroyed. No
persistent-storage guarantee. */
+public final class MemorySemanticAuditStore extends ServiceSupport implements
SemanticAuditSink, SemanticAuditReader {
+ private final int capacity;
+ private final String epoch = UUID.randomUUID().toString();
+ private final Deque<Entry> records = new ArrayDeque<>();
+ private long sequence;
+ private long evicted;
+
+ public MemorySemanticAuditStore(int capacity) {
+ if (capacity < 1 || capacity > 100000) {
+ throw new IllegalArgumentException("Invalid audit capacity");
+ }
+ this.capacity = capacity;
+ }
+
+ @Override
+ public synchronized void append(SemanticAuditRecord record) {
+ if (records.size() == capacity) {
+ records.removeLast();
+ evicted++;
+ }
+ records.addFirst(new Entry(++sequence, record));
+ }
+
+ @Override
+ public synchronized Optional<SemanticAuditRecord> get(String eventId) {
+ return records.stream().map(Entry::record).filter(r ->
r.getEventId().equals(eventId)).findFirst();
+ }
+
+ @Override
+ public synchronized Map<String, SemanticAuditRecord> getAll(List<String>
eventIds) {
+ if (eventIds.size() > 100) {
+ throw new IllegalArgumentException("Too many audit evidence
references");
+ }
+ var selected = new HashSet<>(eventIds);
+ Map<String, SemanticAuditRecord> found = new LinkedHashMap<>();
+ for (Entry entry : records) {
+ if (selected.remove(entry.record.getEventId())) {
+ found.put(entry.record.getEventId(), entry.record);
+ }
+ }
+ return Map.copyOf(found);
+ }
+
+ @Override
+ public synchronized SemanticAuditPage query(SemanticAuditQuery query) {
+ long before = Long.MAX_VALUE;
+ if (query.getCursor() != null) {
+ String[] parts = query.getCursor().split(":", -1);
+ if (parts.length != 3 || !epoch.equals(parts[0])
+ || !Integer.toString(query.getFilters().hashCode() +
Objects.hashCode(query.getSince())).equals(parts[2])) {
+ throw new IllegalArgumentException("Audit cursor does not
match this store or query");
+ }
+ try {
+ before = Long.parseLong(parts[1]);
+ } catch (NumberFormatException e) {
+ throw new IllegalArgumentException("Invalid audit cursor", e);
+ }
+ }
+ boolean expired = before != Long.MAX_VALUE && !records.isEmpty() &&
before <= records.getLast().sequence;
+ List<SemanticAuditRecord> page = new ArrayList<>();
+ long last = 0;
+ boolean more = false;
+ for (Entry entry : records) {
+ if (entry.sequence < before && query.matches(entry.record)) {
+ if (page.size() == query.getLimit()) {
+ more = true;
+ break;
+ }
+ page.add(entry.record);
+ last = entry.sequence;
+ }
+ }
+ String next
+ = more ? epoch + ":" + last + ":" +
(query.getFilters().hashCode() + Objects.hashCode(query.getSince())) : null;
Review Comment:
🐛 **Same additive hash issue in next-cursor construction**
The cursor written here uses the same collision-prone formula as the check
above. If the cursor is later validated with a corrected formula, they'll
diverge. Apply the same fix:
```suggestion
String next
= more ? epoch + ":" + last + ":" +
Objects.hash(query.getFilters(), query.getSince()) : null;
```
##########
components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/SemanticAudit.java:
##########
@@ -0,0 +1,515 @@
+/*
+ * 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.semantic;
+
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.IdentityHashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.builder.ThreadPoolBuilder;
+import org.apache.camel.support.service.ServiceHelper;
+import org.apache.camel.support.service.ServiceSupport;
+import org.apache.camel.util.concurrent.ThreadPoolRejectedPolicy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** Context-owned audit dispatcher and independent semantic observation hooks.
*/
+public final class SemanticAudit extends ServiceSupport {
+ public static final String REFERENCES = "CamelSemanticAuditReferences";
+ public static final String REQUEST = "CamelSemanticAuditRequest";
+ private static final Logger LOG =
LoggerFactory.getLogger(SemanticAudit.class);
+ private static final Object CREATION_LOCK = new Object();
+ private final CamelContext context;
+ private volatile SemanticAuditConfiguration configuration =
SemanticAuditConfiguration.DISABLED;
+ private String configurationSource;
+ private final AtomicLong dropped = new AtomicLong();
+ private final AtomicLong observerErrors = new AtomicLong();
+ private final Map<String, AtomicLong> sinkErrors = new LinkedHashMap<>();
+ private final Map<String, SemanticAuditSink> sinks = new LinkedHashMap<>();
+ private volatile List<SemanticObserver> observers = List.of();
+ private volatile ExecutorService executor;
+ private MemorySemanticAuditStore memory;
+ private volatile SemanticAuditReader reader;
+ private volatile boolean activated;
+ private boolean frozen;
+ private final Invocation noop = new Invocation(null, false);
+
+ private SemanticAudit(CamelContext context) {
+ this.context = context;
+ }
+
+ public static SemanticAudit get(CamelContext context) {
+ var extension = context.getCamelContextExtension();
+ SemanticAudit audit = extension.getContextPlugin(SemanticAudit.class);
+ if (audit == null) {
+ synchronized (CREATION_LOCK) {
+ audit = extension.getContextPlugin(SemanticAudit.class);
+ if (audit == null) {
+ audit = new SemanticAudit(context);
+ extension.addContextPlugin(SemanticAudit.class, audit);
+ }
+ }
+ }
+ return audit;
+ }
+
+ public synchronized void validateConfiguration(String source,
SemanticAuditConfiguration candidate) {
+ if (configurationSource != null && !configurationSource.equals(source)
&& candidate != null) {
+ throw new IllegalArgumentException("Declare semantic audit
configuration in only one resource");
+ }
+ if (candidate == null && !source.equals(configurationSource)) {
+ return;
+ }
+ SemanticAuditConfiguration next = candidate == null ?
SemanticAuditConfiguration.DISABLED : candidate;
+ if (frozen && !configuration.equals(next)) {
+ throw new IllegalArgumentException("Changing semantic audit
configuration requires a context restart");
+ }
+ }
+
+ public synchronized void configure(String source,
SemanticAuditConfiguration candidate) {
+ validateConfiguration(source, candidate);
+ if (candidate != null || source.equals(configurationSource)) {
+ configuration = candidate == null ?
SemanticAuditConfiguration.DISABLED : candidate;
+ configurationSource = candidate == null ? null : source;
+ }
+ }
+
+ synchronized void removeConfigurationSource(String source) {
+ if (source.equals(configurationSource)) {
+ configurationSource = null;
+ if (frozen) {
+ LOG.warn("Semantic audit resource {} was deleted; retaining
the active configuration until context restart",
+ source);
+ } else {
+ configuration = SemanticAuditConfiguration.DISABLED;
+ }
+ }
+ }
+
+ public void activate() {
+ if (!activated) {
+ synchronized (this) {
+ if (!activated) {
+ try {
+ context.addService(this, true, true);
+ activated = true;
+ } catch (Exception e) {
+ throw
RuntimeCamelException.wrapRuntimeCamelException(e);
+ }
+ }
+ }
+ }
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ synchronized (this) {
+ frozen = true;
+ }
+ observers =
List.copyOf(context.getRegistry().findByType(SemanticObserver.class));
+ if (memory == null) {
+ memory = new MemorySemanticAuditStore(configuration.getCapacity());
+ }
+ for (String name : configuration.getSinks()) {
+ SemanticAuditSink sink = switch (name) {
+ case "memory" -> memory;
+ case "log" -> new LoggingSemanticAuditSink();
+ default -> context.getRegistry().lookupByNameAndType(name,
SemanticAuditSink.class);
+ };
+ if (sink == null) {
+ throw new IllegalArgumentException("Unknown semantic audit
sink: " + name);
+ }
+ sinks.put(name, sink);
+ sinkErrors.putIfAbsent(name, new AtomicLong());
+ }
+ reader = "memory".equals(configuration.getReader())
+ ? memory
+ :
context.getRegistry().lookupByNameAndType(configuration.getReader(),
SemanticAuditReader.class);
+ if (reader == null) {
+ throw new IllegalArgumentException("Unknown semantic audit reader:
" + configuration.getReader());
+ }
+ boolean capturing = configuration.isEnabled() ||
configuration.getExperts().containsValue(true);
+ if (!capturing) {
+ return;
+ }
+ try {
+ for (SemanticAuditSink sink : sinks.values()) {
+ ServiceHelper.startService(sink);
+ }
+ ServiceHelper.startService(reader);
+ executor = new
ThreadPoolBuilder(context).poolSize(1).maxPoolSize(1)
+
.maxQueueSize(configuration.getQueueCapacity()).rejectedPolicy(ThreadPoolRejectedPolicy.Abort)
+ .build(this, "SemanticAudit");
+ } catch (Exception e) {
+ ServiceHelper.stopService(sinks.values());
+ ServiceHelper.stopService(reader);
+ throw e;
+ }
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ ExecutorService current = executor;
+ executor = null;
+ if (current != null) {
+ current.shutdown();
+ try {
+ if (!current.awaitTermination(5, TimeUnit.SECONDS)) {
+ dropped.addAndGet(current.shutdownNow().size());
+ }
+ } catch (InterruptedException e) {
+ dropped.addAndGet(current.shutdownNow().size());
+ Thread.currentThread().interrupt();
+ } finally {
+ context.getExecutorServiceManager().shutdownNow(current);
+ }
+ }
+ ServiceHelper.stopService(sinks.values());
+ ServiceHelper.stopService(reader);
+ }
+
+ public synchronized SemanticAuditConfiguration getConfiguration() {
+ return configuration;
+ }
+
+ public SemanticAuditReader getReader() {
+ activate();
+ if (isStarted()) {
+ try {
+ ServiceHelper.startService(reader);
+ } catch (Exception failure) {
+ throw RuntimeCamelException.wrapRuntimeCamelException(failure);
+ }
+ }
+ return reader;
+ }
+
+ public synchronized Map<String, Object> status() {
+ Map<String, Object> status = new LinkedHashMap<>();
+ status.put("enabled", configuration.isEnabled());
+ status.put("experts", configuration.getExperts());
+ status.put("reader", configuration.getReader());
+ status.put("sinks", configuration.getSinks());
+ status.put("capacity", configuration.getCapacity());
+ status.put("retained", memory == null ? 0 : memory.size());
+ status.put("dropped", dropped.get());
+ status.put("observerErrors", observerErrors.get());
+ Map<String, Long> errors = new LinkedHashMap<>();
+ sinkErrors.forEach((name, count) -> errors.put(name, count.get()));
+ status.put("sinkErrors", errors);
+ status.put("state", getStatus().name());
+ status.put("observerCount", observers.size());
+ status.put("openTelemetry",
+ ServiceHelper.isStarted(context.hasService(
+ service -> isOpenTelemetryTracer(service.getClass())
&& ServiceHelper.isStarted(service)))
+ ? "active" : "inactive");
+ return status;
+ }
+
+ private static boolean isOpenTelemetryTracer(Class<?> type) {
+ for (Class<?> current = type; current != null; current =
current.getSuperclass()) {
+ if
(current.getName().equals("org.apache.camel.opentelemetry2.OpenTelemetryTracer")
+ ||
current.getName().equals("org.apache.camel.opentelemetry.OpenTelemetryTracer"))
{
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /** Capture parent observation contexts on the caller thread before
dispatching a console request. */
+ public Request request(String origin) {
+ activate();
+ Map<SemanticObserver, Object> parents = new IdentityHashMap<>();
+ for (SemanticObserver observer : observers) {
+ try {
+ parents.put(observer, observer.captureContext());
+ } catch (Exception | AssertionError | LinkageError failure) {
+ observerErrors.incrementAndGet();
+ }
+ }
+ return new Request(origin, parents);
+ }
+
+ /** A request identity is shared by cancellation observations and eventual
provider completion. */
+ public static final class Request {
+ private final String id = UUID.randomUUID().toString();
+ private final String origin;
+ private final Map<SemanticObserver, Object> parents;
+ private final AtomicBoolean claimed = new AtomicBoolean();
+ private volatile boolean audited;
+
+ private Request(String origin, Map<SemanticObserver, Object> parents) {
+ this.origin = origin;
+ this.parents = parents;
+ }
+ }
+
+ public Invocation begin(
+ Exchange exchange, String name, SemanticEvaluation evaluation,
+ SemanticCapabilities.Operation operation, String expert, String
provider, String batchId, Request request) {
+ activate();
+ if (request == null && exchange != null) {
+ request = exchange.getProperty(REQUEST, Request.class);
+ }
+ boolean first = request != null &&
request.claimed.compareAndSet(false, true);
+ boolean enabled = expert != null && configuration.isEnabled(expert);
+ if (request != null && enabled) {
+ request.audited = true;
+ }
+ if (!enabled && observers.isEmpty()) {
+ return noop;
+ }
+ Map<String, Object> fields = new LinkedHashMap<>();
+ fields.put("schemaVersion", 1);
+ fields.put("eventId", UUID.randomUUID().toString());
+ fields.put("invocationId", first ? request.id :
UUID.randomUUID().toString());
+ fields.put("timestamp", Instant.now().toString());
+ fields.put("startedAt", fields.get("timestamp"));
+ fields.put("category", "evaluation");
+ put(fields, "provider", provider);
+ fields.put("origin", request == null ? exchange == null ? "direct" :
"route" : request.origin);
+ put(fields, "requestId", request == null ? null : request.id);
+ put(fields, "batchId", batchId);
+ put(fields, "definition", name);
+ put(fields, "target", name);
+ put(fields, "expert", expert);
+ put(fields, "operation", evaluation == null ? null :
evaluation.getOperation());
+ put(fields, "contextId", context.getName());
+ if (exchange != null) {
+ put(fields, "exchangeId", exchange.getExchangeId());
+ put(fields, "routeId", exchange.getFromRouteId());
+ }
+ if (operation != null) {
+ Map<String, Object> semantics = new LinkedHashMap<>();
+ put(semantics, "resultType", operation.getResultType().name());
+ put(semantics, "meaning", operation.getResultMeaning());
+ put(semantics, "probabilityMeaning",
operation.getProbabilityMeaning());
+ put(semantics, "confidenceMeaning",
operation.getConfidenceMeaning());
+ // Caller-authored score descriptions and arbitrary parameters are
deliberately omitted.
+ fields.put("semantics", Map.copyOf(semantics));
+ }
+ Invocation invocation = new Invocation(fields, enabled);
+ SemanticAuditRecord started = new SemanticAuditRecord(fields);
+ for (SemanticObserver observer : observers) {
+ try {
+ Object parent = request == null ? observer.captureContext() :
request.parents.get(observer);
+ SemanticObserver.Observation handle =
observer.started(started, parent);
+ if (handle != null) {
+ invocation.handles.add(handle);
+ }
+ } catch (Exception | AssertionError | LinkageError failure) {
+ observerErrors.incrementAndGet();
+ }
+ }
+ return invocation;
+ }
+
+ public final class Invocation {
+ private final Map<String, Object> fields;
+ private final boolean enabled;
+ private final long start = System.nanoTime();
+ private final List<SemanticObserver.Observation> handles = new
ArrayList<>();
+ private boolean completed;
Review Comment:
💡 **`completed` should be `volatile` for clarity**
`Invocation.completed` is a plain `boolean`. The current code appears safe
under a single-thread-completion invariant, but the guard `if (fields == null
|| completed) { return; }` at line 352 is a visibility pattern that relies on
the field being readable across threads. A `volatile` annotation makes the
intent explicit, guards against future refactors that might break the
invariant, and costs nothing at runtime:
```suggestion
private volatile boolean completed;
```
--
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]
