gnodet-bot commented on code in PR #27791:
URL: https://github.com/apache/camel/pull/27791#discussion_r4268605062
##########
components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/internal/SemanticAuditService.java:
##########
@@ -371,27 +395,61 @@ public Invocation begin(
observerErrors.incrementAndGet();
}
}
+ EventHelper.notifyAiEvent(context, () -> event);
return invocation;
}
+ private Set<String> supportedExtensions(Supplier<Set<String>>
capabilities) {
+ try {
+ return Set.copyOf(capabilities.get());
+ } catch (Exception | AssertionError | LinkageError failure) {
+ capabilityErrors.incrementAndGet();
+ return Set.of();
+ }
+ }
+
+ private boolean hasEventConsumers() {
+ var management = context.getManagementStrategy();
+ if (management != null && management.getStartedEventNotifiers() !=
null) {
+ for (var notifier : management.getStartedEventNotifiers()) {
+ try {
+ if (!(notifier instanceof SemanticMonitoringService)
+ && !notifier.isDisabled() &&
!notifier.isIgnoreAiEvents()) {
+ return true;
+ }
+ } catch (Throwable failure) {
+ // Match AI event dispatch: a broken listener must not
change evaluation or block other listeners.
+ LOG.warn("Cannot check semantic event listener
eligibility", failure);
+ }
+ }
+ }
+ return false;
+ }
Review Comment:
⚠️ **`hasEventConsumers()` fast-path is effectively always `true` when
management is active**
The method returns `true` for any notifier where `!notifier.isDisabled() &&
!notifier.isIgnoreAiEvents()`. The `isIgnoreAiEvents()` default is `false`, so
notifiers that exist purely for non-AI purposes (e.g.
`MainDurationEventNotifier`, `EventConsole`, any custom notifier) will trigger
a `true` return even though they never process AI events.
The short-circuit at the call sites (lines 349, 547) is meant to skip event
generation when no one cares — but since generic notifiers default to
`isIgnoreAiEvents=false`, the fast-path is bypassed in virtually every Camel
application with management enabled.
Consider inverting the opt-in: check `notifier.isEnabled(/* an AI event
sentinel */)`, or document that any notifier wishing to opt out of AI event
overhead must call `setIgnoreAiEvents(true)`. The well-known exchange-level
notifiers (`MicrometerExchangeEventNotifier`,
`OpenTelemetryExchangeEventNotifier`) already call `setIgnoreAiEvents(true)`,
but application-specific notifiers will not.
##########
components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/internal/SemanticMonitoringService.java:
##########
@@ -0,0 +1,430 @@
+/*
+ * 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.internal;
+
+import java.time.Instant;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Deque;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+import javax.management.ObjectName;
+import javax.management.StandardMBean;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.CamelContextAware;
+import org.apache.camel.builder.ThreadPoolBuilder;
+import org.apache.camel.semantic.SemanticMetricSource;
+import org.apache.camel.semantic.SemanticMonitoringConfiguration;
+import org.apache.camel.semantic.SemanticMonitoringEvent;
+import org.apache.camel.semantic.SemanticMonitoringSnapshot;
+import org.apache.camel.semantic.SemanticMonitoringStore;
+import org.apache.camel.spi.CamelEvent;
+import org.apache.camel.spi.EventNotifier;
+import org.apache.camel.support.EventNotifierSupport;
+import org.apache.camel.support.service.ServiceHelper;
+import org.apache.camel.util.concurrent.ThreadPoolRejectedPolicy;
+import org.apache.camel.util.json.JsonObject;
+
+/** Bounded read-only projection of supplied measurements, shared by all
surfaces. */
+public final class SemanticMonitoringService extends EventNotifierSupport {
+ private static final Object CREATION_LOCK = new Object();
+ private static final int CAPACITY = 100;
+ private static final int MAX_SOURCES = 32;
+ private final Map<String, SemanticMetricSource> sources = new
LinkedHashMap<>();
+ private final Map<String, SemanticMonitoringStore> stores = new
LinkedHashMap<>();
+ private final Map<String, Cached> latest = new LinkedHashMap<>();
+ private SemanticMonitoringConfiguration configuration = new
SemanticMonitoringConfiguration();
+ private final Map<String, RetainedSample> retainedSamples = new
LinkedHashMap<>();
+ private final Deque<Cached> history = new ArrayDeque<>();
+ private final List<EventNotifier> ownedNotifiers = new ArrayList<>();
+ private final List<ObjectName> mbeans = new ArrayList<>();
+ private final AtomicLong sourceErrors = new AtomicLong();
+ private final AtomicLong storeErrors = new AtomicLong();
+ private final AtomicLong dropped = new AtomicLong();
+ private final AtomicLong evicted = new AtomicLong();
+ private final AtomicLong managementErrors = new AtomicLong();
+ private volatile ExecutorService writer;
+ private ScheduledExecutorService sampler;
+ private volatile String lastSample;
+ private final Map<String, Map<String, Object>> sourceHealth = new
ConcurrentHashMap<>();
+ private final Map<String, Map<String, Object>> storeHealth = new
ConcurrentHashMap<>();
+ private int omittedSources;
+ private int omittedStores;
Review Comment:
⚠️ **JMM visibility gap: `omittedSources` and `omittedStores` should be
`volatile`**
These plain `int` fields are written in `doStart()` (lines 122–123) without
synchronization, then read in `snapshot()` (lines 362–363) outside the
`synchronized (this)` block (which only covers lines 349–356). Under the Java
Memory Model, a thread calling `snapshot()` is not guaranteed to see the values
written during startup.
Since these are written once at startup and used only for display,
`volatile` is the lowest-cost fix:
```suggestion
private volatile int omittedSources;
private volatile int omittedStores;
```
##########
components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/internal/SemanticMonitoringService.java:
##########
@@ -0,0 +1,430 @@
+/*
+ * 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.internal;
+
+import java.time.Instant;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Deque;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+import javax.management.ObjectName;
+import javax.management.StandardMBean;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.CamelContextAware;
+import org.apache.camel.builder.ThreadPoolBuilder;
+import org.apache.camel.semantic.SemanticMetricSource;
+import org.apache.camel.semantic.SemanticMonitoringConfiguration;
+import org.apache.camel.semantic.SemanticMonitoringEvent;
+import org.apache.camel.semantic.SemanticMonitoringSnapshot;
+import org.apache.camel.semantic.SemanticMonitoringStore;
+import org.apache.camel.spi.CamelEvent;
+import org.apache.camel.spi.EventNotifier;
+import org.apache.camel.support.EventNotifierSupport;
+import org.apache.camel.support.service.ServiceHelper;
+import org.apache.camel.util.concurrent.ThreadPoolRejectedPolicy;
+import org.apache.camel.util.json.JsonObject;
+
+/** Bounded read-only projection of supplied measurements, shared by all
surfaces. */
+public final class SemanticMonitoringService extends EventNotifierSupport {
+ private static final Object CREATION_LOCK = new Object();
+ private static final int CAPACITY = 100;
+ private static final int MAX_SOURCES = 32;
+ private final Map<String, SemanticMetricSource> sources = new
LinkedHashMap<>();
+ private final Map<String, SemanticMonitoringStore> stores = new
LinkedHashMap<>();
+ private final Map<String, Cached> latest = new LinkedHashMap<>();
+ private SemanticMonitoringConfiguration configuration = new
SemanticMonitoringConfiguration();
Review Comment:
⚠️ **JMM visibility gap: `configuration` field should be `volatile`**
This field is assigned in `doStart()` (line 114, outside any `synchronized`
block) and read in `snapshot()` at line 374
(`configuration.getHistoryInterval()`) also outside the `synchronized (this)`
block. A thread calling `snapshot()` may observe the stale initial `new
SemanticMonitoringConfiguration()` value rather than the one registered in the
bean registry.
```suggestion
private volatile SemanticMonitoringConfiguration configuration = new
SemanticMonitoringConfiguration();
```
--
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]
