You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@eventmesh.apache.org by "Copilot (via GitHub)" <gi...@apache.org> on 2026/05/08 07:52:59 UTC
Re: [PR] [ISSUE #5249]Refactor/unified runtime pipeline and protocol processors (eventmesh)
Copilot commented on code in PR #5250:
URL: https://github.com/apache/eventmesh/pull/5250#discussion_r3207222985
##########
eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientGroupWrapperTest.java:
##########
@@ -0,0 +1,198 @@
+/*
+ * 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.eventmesh.runtime.core.protocol.tcp.client.group;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import org.apache.eventmesh.api.SendCallback;
+import org.apache.eventmesh.common.protocol.tcp.Header;
+import org.apache.eventmesh.common.protocol.tcp.UserAgent;
+import org.apache.eventmesh.function.api.Router;
+import org.apache.eventmesh.function.filter.pattern.Pattern;
+import org.apache.eventmesh.function.transformer.Transformer;
+import org.apache.eventmesh.runtime.boot.EventMeshServer;
Review Comment:
This test file has unused imports (e.g., doReturn, UserAgent) which will fail compilation and/or formatting checks. Remove the unused imports to keep the test compiling cleanly.
##########
eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientGroupWrapperTest.java:
##########
@@ -0,0 +1,198 @@
+/*
+ * 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.eventmesh.runtime.core.protocol.tcp.client.group;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import org.apache.eventmesh.api.SendCallback;
+import org.apache.eventmesh.common.protocol.tcp.Header;
+import org.apache.eventmesh.common.protocol.tcp.UserAgent;
+import org.apache.eventmesh.function.api.Router;
+import org.apache.eventmesh.function.filter.pattern.Pattern;
+import org.apache.eventmesh.function.transformer.Transformer;
+import org.apache.eventmesh.runtime.boot.EventMeshServer;
+import org.apache.eventmesh.runtime.boot.EventMeshTCPServer;
+import org.apache.eventmesh.runtime.boot.FilterEngine;
+import org.apache.eventmesh.runtime.boot.RouterEngine;
+import org.apache.eventmesh.runtime.boot.TransformerEngine;
+import org.apache.eventmesh.runtime.configuration.EventMeshTCPConfiguration;
+import org.apache.eventmesh.runtime.core.plugin.MQProducerWrapper;
+import org.apache.eventmesh.runtime.core.protocol.tcp.client.group.dispatch.DownstreamDispatchStrategy;
+import org.apache.eventmesh.runtime.core.protocol.tcp.client.session.Session;
+import org.apache.eventmesh.runtime.core.protocol.tcp.client.session.retry.TcpRetryer;
+import org.apache.eventmesh.runtime.core.protocol.tcp.client.session.send.UpStreamMsgContext;
+import org.apache.eventmesh.runtime.metrics.tcp.EventMeshTcpMetricsManager;
+
+import java.nio.charset.StandardCharsets;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+@ExtendWith(MockitoExtension.class)
+public class ClientGroupWrapperTest {
+
+ @Mock
+ private EventMeshTCPServer eventMeshTCPServer;
+
+ @Mock
+ private EventMeshServer eventMeshServer;
+
+ @Mock
+ private EventMeshTCPConfiguration eventMeshTCPConfiguration;
+
+ @Mock
+ private TcpRetryer tcpRetryer;
+
+ @Mock
+ private EventMeshTcpMetricsManager eventMeshTcpMetricsManager;
+
+ @Mock
+ private DownstreamDispatchStrategy downstreamDispatchStrategy;
+
+ @Mock
+ private FilterEngine filterEngine;
+
+ @Mock
+ private TransformerEngine transformerEngine;
+
+ @Mock
+ private RouterEngine routerEngine;
+
+ @Mock
+ private MQProducerWrapper mqProducerWrapper;
+
+ private ClientGroupWrapper clientGroupWrapper;
+
+ @BeforeEach
+ public void setUp() {
+ lenient().when(eventMeshTCPServer.getEventMeshTCPConfiguration()).thenReturn(eventMeshTCPConfiguration);
+ lenient().when(eventMeshTCPServer.getTcpRetryer()).thenReturn(tcpRetryer);
+ lenient().when(eventMeshTCPServer.getEventMeshTcpMetricsManager()).thenReturn(eventMeshTcpMetricsManager);
+ lenient().when(eventMeshTCPConfiguration.getEventMeshStoragePluginType()).thenReturn("standalone");
+ lenient().when(eventMeshTCPServer.getEventMeshServer()).thenReturn(eventMeshServer);
+ lenient().when(eventMeshServer.getFilterEngine()).thenReturn(filterEngine);
+ lenient().when(eventMeshServer.getTransformerEngine()).thenReturn(transformerEngine);
+ lenient().when(eventMeshServer.getRouterEngine()).thenReturn(routerEngine);
+
+ clientGroupWrapper = spy(new ClientGroupWrapper("sysId", "group", eventMeshTCPServer, downstreamDispatchStrategy));
+ // Reflection to set mqProducerWrapper if needed, or we can mock the one created internally if possible?
+ // Since ClientGroupWrapper creates `new MQProducerWrapper`, we can't easily mock it unless we assume it works or use PowerMock.
+ // But ClientGroupWrapper has a getter `getMqProducerWrapper()`, maybe we can set it via reflection or if there is a setter?
+ // No setter.
+ // We will assume `new MQProducerWrapper("standalone")` works fine (it uses SPI, might fail if no plugin).
+ // To verify the pipeline, we mainly care about Engines interaction.
+
+ // However, `send` calls `mqProducerWrapper.send`. If that fails, test fails.
+ // For unit test, we might need to mock the internal mqProducerWrapper.
+ // Since we are mocking `EventMeshTCPServer`, `ClientGroupWrapper` uses it to access engines.
+
+ // Let's try to set the internal mqProducerWrapper field using reflection.
+ try {
+ java.lang.reflect.Field field = ClientGroupWrapper.class.getDeclaredField("mqProducerWrapper");
+ field.setAccessible(true);
+ field.set(clientGroupWrapper, mqProducerWrapper);
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
Review Comment:
The setup swallows reflection failures and prints to stdout via e.printStackTrace(). If reflection fails, the test should fail deterministically (e.g., rethrow AssertionError) and use the logger/test assertions rather than printing stack traces.
##########
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientGroupWrapper.java:
##########
@@ -25,6 +25,8 @@
import org.apache.eventmesh.api.SendCallback;
import org.apache.eventmesh.api.SendResult;
import org.apache.eventmesh.api.exception.OnExceptionContext;
+import org.apache.eventmesh.api.exception.StorageRuntimeException;
+import org.apache.eventmesh.common.exception.EventMeshException;
import org.apache.eventmesh.common.protocol.SubscriptionItem;
Review Comment:
There is an unused import of EventMeshException which will cause compilation to fail. Remove the unused import.
##########
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientGroupWrapper.java:
##########
@@ -136,6 +143,16 @@ public ClientGroupWrapper(String sysId, String group,
this.persistentMsgConsumer = new MQConsumerWrapper(eventMeshTCPServer.getEventMeshTCPConfiguration().getEventMeshStoragePluginType());
this.broadCastMsgConsumer = new MQConsumerWrapper(eventMeshTCPServer.getEventMeshTCPConfiguration().getEventMeshStoragePluginType());
this.mqProducerWrapper = new MQProducerWrapper(eventMeshTCPServer.getEventMeshTCPConfiguration().getEventMeshStoragePluginType());
+
+ this.ingressProcessor = new IngressProcessor(
+ eventMeshTCPServer.getEventMeshServer().getFilterEngine(),
+ eventMeshTCPServer.getEventMeshServer().getTransformerEngine(),
+ eventMeshTCPServer.getEventMeshServer().getRouterEngine()
+ );
+ this.egressProcessor = new EgressProcessor(
+ eventMeshTCPServer.getEventMeshServer().getFilterEngine(),
+ eventMeshTCPServer.getEventMeshServer().getTransformerEngine()
+ );
Review Comment:
ClientGroupWrapper constructs new IngressProcessor/EgressProcessor instances per group wrapper. Since EventMeshServer already initializes shared processors, prefer reusing eventMeshTCPServer.getEventMeshServer().getIngressProcessor()/getEgressProcessor() to avoid duplicated objects and to keep pipeline behavior centralized.
##########
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/EgressProcessor.java:
##########
@@ -0,0 +1,70 @@
+/*
+ * 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.eventmesh.runtime.core.protocol;
+
+import org.apache.eventmesh.runtime.boot.FilterEngine;
+import org.apache.eventmesh.runtime.boot.TransformerEngine;
+
+import java.nio.charset.StandardCharsets;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class EgressProcessor {
+
+ private final FilterEngine filterEngine;
+ private final TransformerEngine transformerEngine;
+
+ public EgressProcessor(FilterEngine filterEngine, TransformerEngine transformerEngine) {
+ this.filterEngine = filterEngine;
+ this.transformerEngine = transformerEngine;
+ }
+
+ public CloudEvent process(CloudEvent event, String pipelineKey) {
+ try {
+ // 1. Filter
+ org.apache.eventmesh.function.filter.pattern.Pattern filterPattern = filterEngine.getFilterPattern(pipelineKey);
+ if (filterPattern != null && event.getData() != null) {
+ String content = new String(event.getData().toBytes(), StandardCharsets.UTF_8);
+ if (!filterPattern.filter(content)) {
+ // Filtered out
+ return null;
+ }
+ }
+
+ // 2. Transformer
+ org.apache.eventmesh.function.transformer.Transformer transformer = transformerEngine.getTransformer(pipelineKey);
+ if (transformer != null && event.getData() != null) {
+ String content = new String(event.getData().toBytes(), StandardCharsets.UTF_8);
+ String transformedContent = transformer.transform(content);
+ event = CloudEventBuilder.from(event)
+ .withData(transformedContent.getBytes(StandardCharsets.UTF_8))
+ .build();
+ }
Review Comment:
EgressProcessor currently runs filter/transformer against event.getData() (payload) only and skips processing when data is null. This is inconsistent with existing filter/transformer usage (full CloudEvent JSON) and can break consumer-side filtering that relies on CloudEvent attributes. Consider using the full CloudEvent JSON string as the processing input and not gating on data presence.
##########
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/BatchProcessResult.java:
##########
@@ -0,0 +1,164 @@
+/*
+ * 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.eventmesh.runtime.core.protocol;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Result tracker for batch message processing.
+ * Tracks success, filtered, and failed message counts for batch operations.
+ */
+public class BatchProcessResult {
+
+ private final int totalCount;
+ private int successCount;
+ private int filteredCount;
+ private int failedCount;
+ private final List<String> failedMessageIds;
+
+ public BatchProcessResult(int totalCount) {
+ this.totalCount = totalCount;
+ this.successCount = 0;
+ this.filteredCount = 0;
+ this.failedCount = 0;
+ this.failedMessageIds = new ArrayList<>();
+ }
+
+ /**
+ * Increment the success count by one.
+ */
+ public void incrementSuccess() {
+ successCount++;
+ }
+
+ /**
+ * Increment the filtered count by one.
+ */
+ public void incrementFiltered() {
+ filteredCount++;
+ }
+
+ /**
+ * Increment the failed count by one and record the failed message ID.
+ *
+ * @param messageId the ID of the failed message
+ */
+ public void incrementFailed(String messageId) {
+ failedCount++;
+ if (messageId != null) {
+ failedMessageIds.add(messageId);
+ }
Review Comment:
BatchProcessResult is mutated from async SendCallback handlers in batch processors, but its counters and failedMessageIds list are not thread-safe. If callbacks can run concurrently, counts and list updates can race. Consider using AtomicInteger/LongAdder and a thread-safe list (or synchronize) if this is updated across threads.
##########
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/RouterEngine.java:
##########
@@ -0,0 +1,92 @@
+/*
+ * 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.eventmesh.runtime.boot;
+
+import org.apache.eventmesh.api.meta.MetaServiceListener;
+import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.function.api.Router;
+import org.apache.eventmesh.function.router.RouterBuilder;
+import org.apache.eventmesh.runtime.meta.MetaStorage;
+
+import org.apache.commons.lang3.StringUtils;
+
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.ConcurrentHashMap;
+
+import com.fasterxml.jackson.databind.JsonNode;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class RouterEngine {
+
+ private final MetaStorage metaStorage;
+
+ private final Map<String, Router> routerMap = new ConcurrentHashMap<>();
+
+ private final String routerPrefix = "router-";
+
+ private MetaServiceListener metaServiceListener;
+
+ public RouterEngine(MetaStorage metaStorage) {
+ this.metaStorage = metaStorage;
+ }
+
+ public void start() {
+ Map<String, String> routerMetaData = metaStorage.getMetaData(routerPrefix, true);
+ for (Entry<String, String> routerDataEntry : routerMetaData.entrySet()) {
+ String key = routerDataEntry.getKey();
+ String value = routerDataEntry.getValue();
+ updateRouterMap(key, value);
+ }
+ metaServiceListener = this::updateRouterMap;
+ }
+
+ private void updateRouterMap(String key, String value) {
+ String group = StringUtils.substringAfter(key, routerPrefix);
+
+ JsonNode routerJsonNodeArray = JsonUtils.getJsonNode(value);
+ if (routerJsonNodeArray != null) {
+ for (JsonNode routerJsonNode : routerJsonNodeArray) {
+ String topic = routerJsonNode.get("topic").asText();
+ String routerConfig = routerJsonNode.get("routerConfig").toString();
+ Router router = RouterBuilder.build(routerConfig);
+ routerMap.put(group + "-" + topic, router);
Review Comment:
RouterEngine extracts routerConfig via JsonNode.toString(), which will include quotes for JSON string values (e.g., "targetTopic"). That causes routed topics to contain extra quotes. If routerConfig can be either a string or an object, consider using asText() when the node is textual and toString() otherwise.
##########
docs/unified-runtime-design.md:
##########
@@ -0,0 +1,201 @@
+# Unified Runtime Design & Usage Guide
+
+## 1. Overview
+The EventMesh Unified Runtime consolidates the capabilities of the core EventMesh Runtime (Protocol handling), Connectors (Source/Sink), and Functions (Filter/Transformer/Router) into a single, cohesive process. This eliminates the need for separate deployments for Connectors ("Runtime V2") and simplifies the architecture.
+
+## 2. Architecture: The Unified Processing Pipeline
+
+The system implements a symmetrical processing chain for both event production (Ingress) and consumption (Egress), but the entry/exit points differ based on the client type (SDK vs. Connector).
+
+### 2.1 Ingress Pipeline (Production)
+
+**Entry Points:**
+* **SDK Client**: Interacts with the Runtime via **Protocol Servers** (TCP/HTTP/gRPC). The Protocol Server receives the request and passes the event to the pipeline.
+* **Source Connector**: Loaded directly into the Runtime as a **Plugin**. The Source Connector pulls data from external systems and internally injects events into the pipeline.
+
+**Flow:**
+`[Entry: Protocol Server (SDK) OR Source Plugin (Connector)] -> [IngressProcessor] -> [Storage]`
+
+**IngressProcessor Pipeline:**
+`[Filter] -> [Transformer] -> [Router]`
+
+1. **Entry**:
+ * **SDK**: Request received by `EventMeshTCPServer`, `EventMeshHTTPServer`, or `EventMeshGrpcServer`.
+ * **Connector**: `SourceWorker` pulls data and converts it to a CloudEvent.
+2. **IngressProcessor**: Encapsulates the unified 3-stage pipeline:
+ * **Filter**: The `FilterEngine` evaluates the event against configured rules. If unmatched, returns null (event dropped).
+ * **Transformer**: The `TransformerEngine` transforms the event payload (e.g., JSON manipulation) if a rule exists.
+ * **Router**: The `RouterEngine` determines the target topic/destination.
+3. **Storage**: The processed event is persisted to the Storage Plugin (RocketMQ, Kafka, etc.).
+
+### 2.2 Egress Pipeline (Consumption)
+
+**Exit Points:**
+* **SDK Client**: The Runtime pushes events to connected SDK clients via the active **Protocol Server** connection.
+* **Sink Connector**: Loaded directly into the Runtime as a **Plugin**. The Runtime passes events to the `SinkWorker`, which writes to external systems.
+
+**Flow:**
+`[Storage] -> [EgressProcessor] -> [Exit: Protocol Server (SDK) OR Sink Plugin (Connector)]`
+
+**EgressProcessor Pipeline:**
+`[Filter] -> [Transformer]`
+
+1. **Storage**: Event retrieved from the storage queue.
+2. **EgressProcessor**: Encapsulates the 2-stage pipeline (no Router on egress):
+ * **Filter**: Evaluated against the consumer group's filter rules. If unmatched, returns null (event not delivered).
+ * **Transformer**: Payload transformed according to the consumer group's needs.
+3. **Exit**:
+ * **SDK**: Event pushed to client via TCP/HTTP/gRPC session.
+ * **Connector**: Event passed to `SinkWorker` for external delivery.
+
+### 2.3 Protocol Processor Migration Status
+
+All protocol processors now use the unified IngressProcessor/EgressProcessor architecture:
+
+**TCP Protocol**: ✅ Complete
+* `ClientGroupWrapper` - Integrated both Ingress (send) and Egress (consume)
+
+**HTTP Protocol**: ✅ Complete
+* `SendAsyncEventProcessor` - Ingress pipeline
+* `SendAsyncMessageProcessor` - Ingress pipeline
+* `SendSyncMessageProcessor` - Ingress pipeline
+* `BatchSendMessageProcessor` - Ingress pipeline with batch statistics
+* `BatchSendMessageV2Processor` - Ingress pipeline with batch statistics
+
+**gRPC Protocol**: ✅ Complete
+* `PublishCloudEventsProcessor` - Ingress pipeline
+* `BatchPublishCloudEventProcessor` - Ingress pipeline with batch statistics
+* `RequestCloudEventProcessor` - Bidirectional (Ingress for request, Egress for response)
+
+**Connectors**: ✅ Complete
+* `SourceWorker` - Ingress pipeline
+* `SinkWorker` - Egress pipeline
+
+## 3. Configuration
+
+### 3.1 Enabling Connectors
+To enable the embedded Connector runtime, update `eventmesh.properties`:
+
+```properties
+# Enable the connector plugin
+eventMesh.connector.plugin.enable=true
+
+# Specify the connector type (source or sink) and name (SPI name)
+eventMesh.connector.plugin.type=source
+eventMesh.connector.plugin.name=my-source-connector
+```
+
+### 3.2 Configuring Functions
+Functions are dynamic and configured via the **MetaStorage** (e.g., Nacos, Etcd).
+
+* **Prefixes**:
+ * Filter: `filter-{group}-{topic}`
+ * Transformer: `transformer-{group}-{topic}`
+ * Router: `router-{group}-{topic}`
+
+**Example Nacos Config (Filter):**
+Key: `filter-myGroup-myTopic`
Review Comment:
The MetaStorage key format documented here (`filter-{group}-{topic}` etc.) doesn't match the current engines: FilterEngine/TransformerEngine/RouterEngine load keys by prefix (`filter-`, `transformer-`, `router-`) and treat the remainder as the group, with topics defined inside the JSON array. The examples/format should be updated to reflect `filter-{group}` / `transformer-{group}` / `router-{group}`.
##########
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageProcessor.java:
##########
@@ -239,53 +240,89 @@ public void processRequest(ChannelHandlerContext ctx, AsyncContext<HttpCommand>
long delta = eventSize;
summaryMetrics.recordSendBatchMsg(delta);
+ // Create BatchProcessResult to track success/filtered/failed counts
+ final BatchProcessResult batchResult = new BatchProcessResult(eventList.size());
+ final String finalBatchId = batchId; // Make batchId effectively final for inner classes
+
if (httpConfiguration.isEventMeshServerBatchMsgBatchEnabled()) {
for (List<CloudEvent> eventlist : topicBatchMessageMappings.values()) {
// TODO: Implementation in API. Consider whether to put it in the plug-in.
CloudEvent event = null;
// TODO: Detect the maximum length of messages for different producers.
- final SendMessageContext sendMessageContext = new SendMessageContext(batchId, event, batchEventMeshProducer, eventMeshHTTPServer);
+ final SendMessageContext sendMessageContext = new SendMessageContext(finalBatchId, event, batchEventMeshProducer, eventMeshHTTPServer);
batchEventMeshProducer.send(sendMessageContext, new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
+ batchResult.incrementSuccess();
}
@Override
public void onException(OnExceptionContext context) {
- BATCH_MSG_LOGGER.warn("", context.getException());
+ batchResult.incrementFailed(event != null ? event.getId() : "unknown");
+ BATCH_MSG_LOGGER.warn("Batch message send failed: {}", event != null ? event.getId() : "unknown",
+ context.getException());
eventMeshHTTPServer.getHttpRetryer().newTimeout(sendMessageContext, 10, TimeUnit.SECONDS);
}
});
}
} else {
+ // Process each event individually with Ingress Pipeline
for (CloudEvent event : eventList) {
- final SendMessageContext sendMessageContext = new SendMessageContext(batchId, event, batchEventMeshProducer, eventMeshHTTPServer);
- batchEventMeshProducer.send(sendMessageContext, new SendCallback() {
+ String messageId = event.getId();
+ String topic = event.getSubject();
+ try {
+ // Apply Ingress Pipeline (Filter -> Transformer -> Router)
+ String pipelineKey = producerGroup + "-" + topic;
+ CloudEvent processedEvent = eventMeshHTTPServer.getEventMeshServer().getIngressProcessor()
+ .process(event, pipelineKey);
+
+ if (processedEvent == null) {
+ // Message filtered by pipeline
+ batchResult.incrementFiltered();
+ BATCH_MSG_LOGGER.info("Batch message filtered by pipeline: batchId={}, messageId={}, topic={}",
+ finalBatchId, messageId, topic);
+ continue;
+ }
- @Override
- public void onSuccess(SendResult sendResult) {
+ // Topic may have been changed by Router
+ final String finalTopic = processedEvent.getSubject();
+ final SendMessageContext sendMessageContext = new SendMessageContext(finalBatchId, processedEvent,
+ batchEventMeshProducer, eventMeshHTTPServer);
- }
+ batchEventMeshProducer.send(sendMessageContext, new SendCallback() {
- @Override
- public void onException(OnExceptionContext context) {
- BATCH_MSG_LOGGER.warn("", context.getException());
- eventMeshHTTPServer.getHttpRetryer().newTimeout(sendMessageContext, 10, TimeUnit.SECONDS);
- }
+ @Override
+ public void onSuccess(SendResult sendResult) {
+ batchResult.incrementSuccess();
+ }
- });
+ @Override
+ public void onException(OnExceptionContext context) {
+ batchResult.incrementFailed(messageId);
+ BATCH_MSG_LOGGER.warn("Batch message send failed: batchId={}, messageId={}, topic={}",
+ finalBatchId, messageId, finalTopic, context.getException());
+ eventMeshHTTPServer.getHttpRetryer().newTimeout(sendMessageContext, 10, TimeUnit.SECONDS);
+ }
+
+ });
+ } catch (Exception e) {
+ batchResult.incrementFailed(messageId);
+ BATCH_MSG_LOGGER.error("Batch message pipeline exception: batchId={}, messageId={}, topic={}",
+ finalBatchId, messageId, topic, e);
+ }
}
}
long elapsed = stopwatch.elapsed(TimeUnit.MILLISECONDS);
summaryMetrics.recordBatchSendMsgCost(elapsed);
- BATCH_MSG_LOGGER.debug("batchMessage|eventMesh2mq|REQ|ASYNC|batchId={}|send2MQCost={}ms|msgNum={}|topics={}",
- batchId, elapsed, eventSize, topicBatchMessageMappings.keySet());
- completeResponse(request, asyncContext, sendMessageBatchResponseHeader, EventMeshRetCode.SUCCESS, null,
- SendMessageBatchResponseBody.class);
+ BATCH_MSG_LOGGER.info("batchMessage|eventMesh2mq|REQ|ASYNC|batchId={}|send2MQCost={}ms|result={}|topics={}",
+ finalBatchId, elapsed, batchResult.toSummary(), topicBatchMessageMappings.keySet());
+
+ completeResponse(request, asyncContext, sendMessageBatchResponseHeader, EventMeshRetCode.SUCCESS,
+ batchResult.toSummary(), SendMessageBatchResponseBody.class);
return;
Review Comment:
The response includes batchResult.toSummary(), but send operations are asynchronous and batchResult is only incremented inside SendCallback handlers. As written, the HTTP response will likely report success/failed counts before callbacks run (often all zeros). Either compute results synchronously before responding, or move the response completion into the callbacks (or clarify that the summary is only for pre-send filtering/pipeline outcomes).
##########
eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/core/protocol/IngressProcessorTest.java:
##########
@@ -0,0 +1,321 @@
+/*
+ * 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.eventmesh.runtime.core.protocol;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
Review Comment:
Unused static import `lenient` (and potentially others) will fail compilation. Remove unused imports from this test file.
##########
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshServer.java:
##########
@@ -93,6 +94,28 @@ public class EventMeshServer {
private EventMeshMetricsManager eventMeshMetricsManager;
+ @Getter
+ private FilterEngine filterEngine;
+
+ @Getter
+ private TransformerEngine transformerEngine;
+
+ @Getter
+ private RouterEngine routerEngine;
+
+ @Getter
+ private org.apache.eventmesh.runtime.core.protocol.IngressProcessor ingressProcessor;
+
+ @Getter
+ private org.apache.eventmesh.runtime.core.protocol.EgressProcessor egressProcessor;
+
+ @Getter
+ private A2APublishSubscribeService a2aPublishSubscribeService;
+
+ public A2APublishSubscribeService getA2APublishSubscribeService() {
+ return a2aPublishSubscribeService;
+ }
Review Comment:
a2aPublishSubscribeService already has a Lombok @Getter, but there is also a manually defined getA2APublishSubscribeService() method. This duplication is unnecessary and can cause confusion if they diverge later; consider removing the manual getter and relying on @Getter.
##########
eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/core/protocol/EgressProcessorTest.java:
##########
@@ -0,0 +1,288 @@
+/*
+ * 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.eventmesh.runtime.core.protocol;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
Review Comment:
Unused static import `lenient` (and potentially others) will fail compilation. Remove unused imports from this test file.
##########
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/IngressProcessor.java:
##########
@@ -0,0 +1,83 @@
+/*
+ * 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.eventmesh.runtime.core.protocol;
+
+import org.apache.eventmesh.runtime.boot.FilterEngine;
+import org.apache.eventmesh.runtime.boot.RouterEngine;
+import org.apache.eventmesh.runtime.boot.TransformerEngine;
+
+import java.nio.charset.StandardCharsets;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class IngressProcessor {
+
+ private final FilterEngine filterEngine;
+ private final TransformerEngine transformerEngine;
+ private final RouterEngine routerEngine;
+
+ public IngressProcessor(FilterEngine filterEngine, TransformerEngine transformerEngine, RouterEngine routerEngine) {
+ this.filterEngine = filterEngine;
+ this.transformerEngine = transformerEngine;
+ this.routerEngine = routerEngine;
+ }
+
+ public CloudEvent process(CloudEvent event, String pipelineKey) {
+ try {
+ // 1. Filter
+ org.apache.eventmesh.function.filter.pattern.Pattern filterPattern = filterEngine.getFilterPattern(pipelineKey);
+ if (filterPattern != null && event.getData() != null) {
+ String content = new String(event.getData().toBytes(), StandardCharsets.UTF_8);
+ if (!filterPattern.filter(content)) {
+ // Filtered out
+ return null;
+ }
+ }
+
+ // 2. Transformer
+ org.apache.eventmesh.function.transformer.Transformer transformer = transformerEngine.getTransformer(pipelineKey);
+ if (transformer != null && event.getData() != null) {
+ String content = new String(event.getData().toBytes(), StandardCharsets.UTF_8);
+ String transformedContent = transformer.transform(content);
+ event = CloudEventBuilder.from(event)
+ .withData(transformedContent.getBytes(StandardCharsets.UTF_8))
+ .build();
+ }
+
+ // 3. Router
+ org.apache.eventmesh.function.api.Router router = routerEngine.getRouter(pipelineKey);
+ if (router != null && event.getData() != null) {
+ String content = new String(event.getData().toBytes(), StandardCharsets.UTF_8);
+ String newTopic = router.route(content);
+ event = CloudEventBuilder.from(event)
+ .withSubject(newTopic)
+ .build();
+ }
Review Comment:
IngressProcessor currently applies filter/transform/router to event.getData() (payload) only and skips processing when data is null. Existing filter/transformer implementations in runtime use JsonUtils.toJSONString(cloudEvent) so patterns can match CloudEvent attributes (e.g., $.source/$.type). This change will break filtering/routing that depends on CloudEvent metadata. Consider passing the full CloudEvent JSON string into Pattern/Transformer/Router (and allowing processing even when data is null).
##########
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncMessageProcessor.java:
##########
@@ -227,6 +227,36 @@ public void processRequest(ChannelHandlerContext ctx, AsyncContext<HttpCommand>
eventMeshHTTPServer);
summaryMetrics.recordSendMsg();
+ // Apply Ingress Pipeline (Filter -> Transformer -> Router)
+ String pipelineKey = producerGroup + "-" + topic;
+ event = eventMeshHTTPServer.getEventMeshServer().getIngressProcessor().process(event, pipelineKey);
+
+ if (event == null) {
+ // Message filtered by pipeline - return success
+ HttpCommand filteredResponse = request.createHttpCommandResponse(
+ sendMessageResponseHeader,
+ SendMessageResponseBody.buildBody(EventMeshRetCode.SUCCESS.getRetCode(),
+ "Message filtered by pipeline"));
+ asyncContext.onComplete(filteredResponse, httpCommand -> {
+ try {
+ HTTP_LOGGER.debug("{}", httpCommand);
+ eventMeshHTTPServer.sendResponse(ctx, httpCommand.httpResponse());
+ summaryMetrics.recordHTTPReqResTimeCost(
+ System.currentTimeMillis() - request.getReqTime());
+ } catch (Exception ex) {
+ // ignore
+ }
+ });
+ MESSAGE_LOGGER.info("message|eventMesh2mq|REQ|ASYNC|filtered|topic={}|bizSeqNo={}|uniqueId={}",
+ topic, bizNo, uniqueId);
+ spanWithException(event, protocolVersion, EventMeshRetCode.SUCCESS);
+ return;
Review Comment:
In the filtered-by-pipeline branch, `event` is null but still passed into spanWithException(), which dereferences the CloudEvent to build extension maps. This will throw a NullPointerException. Use the pre-pipeline CloudEvent for tracing (or skip tracing) when the message is filtered.
##########
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshConnectorBootstrap.java:
##########
@@ -0,0 +1,228 @@
+/*
+ * 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.eventmesh.runtime.boot;
+
+import org.apache.eventmesh.api.AsyncConsumeContext;
+import org.apache.eventmesh.api.EventMeshAction;
+import org.apache.eventmesh.api.EventListener;
+import org.apache.eventmesh.api.SendCallback;
+import org.apache.eventmesh.api.exception.OnExceptionContext;
+import org.apache.eventmesh.common.Constants;
+import org.apache.eventmesh.common.config.CommonConfiguration;
+import org.apache.eventmesh.common.config.connector.Config;
+import org.apache.eventmesh.common.config.connector.SinkConfig;
+import org.apache.eventmesh.common.config.connector.SourceConfig;
+import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.function.api.Router;
+import org.apache.eventmesh.openconnect.Application;
+import org.apache.eventmesh.openconnect.ConnectorWorker;
+import org.apache.eventmesh.openconnect.SinkWorker;
+import org.apache.eventmesh.openconnect.SourceWorker;
+import org.apache.eventmesh.openconnect.api.connector.Connector;
+import org.apache.eventmesh.openconnect.api.sink.Sink;
+import org.apache.eventmesh.openconnect.api.source.Source;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.callback.SendExceptionContext;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.callback.SendResult;
+import org.apache.eventmesh.openconnect.util.ConfigUtil;
+import org.apache.eventmesh.runtime.constants.EventMeshConstants;
+import org.apache.eventmesh.runtime.core.protocol.EgressProcessor;
+import org.apache.eventmesh.runtime.core.protocol.IngressProcessor;
+import org.apache.eventmesh.runtime.core.plugin.MQConsumerWrapper;
+import org.apache.eventmesh.runtime.core.plugin.MQProducerWrapper;
+import org.apache.eventmesh.runtime.util.EventMeshUtil;
+import org.apache.eventmesh.spi.EventMeshExtensionFactory;
+
+import java.util.Properties;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class EventMeshConnectorBootstrap implements EventMeshBootstrap {
+
+ private final EventMeshServer eventMeshServer;
+ private ConnectorWorker worker;
+ private Connector connector;
+ private MQProducerWrapper producer;
+ private MQConsumerWrapper consumer;
+ private IngressProcessor ingressProcessor;
+ private EgressProcessor egressProcessor;
+
+ public EventMeshConnectorBootstrap(EventMeshServer eventMeshServer) {
+ this.eventMeshServer = eventMeshServer;
+ }
+
+ @Override
+ public void init() throws Exception {
+ CommonConfiguration config = eventMeshServer.getConfiguration();
+ if (!config.isEventMeshConnectorPluginEnable()) {
+ return;
+ }
+
+ this.ingressProcessor = new IngressProcessor(
+ eventMeshServer.getFilterEngine(),
+ eventMeshServer.getTransformerEngine(),
+ eventMeshServer.getRouterEngine()
+ );
+ this.egressProcessor = new EgressProcessor(
+ eventMeshServer.getFilterEngine(),
+ eventMeshServer.getTransformerEngine()
+ );
+
+ String type = config.getEventMeshConnectorPluginType();
+ String name = config.getEventMeshConnectorPluginName();
+
+ if ("source".equalsIgnoreCase(type)) {
+ connector = EventMeshExtensionFactory.getExtension(Source.class, name);
+ } else if ("sink".equalsIgnoreCase(type)) {
+ connector = EventMeshExtensionFactory.getExtension(Sink.class, name);
+ }
+
+ if (connector == null) {
+ log.error("Connector not found: type={}, name={}", type, name);
+ return;
+ }
+
+ Config connectorConfig = ConfigUtil.parse(connector.configClass());
+
+ if (Application.isSink(connector.getClass())) {
+ worker = new SinkWorker((Sink) connector, (SinkConfig) connectorConfig);
+ ((SinkWorker) worker).setEmbedded(true);
+
+ SinkConfig sinkConfig = (SinkConfig) connectorConfig;
+ consumer = new MQConsumerWrapper(config.getEventMeshStoragePluginType());
+ Properties props = new Properties();
+ props.put(EventMeshConstants.CONSUMER_GROUP, sinkConfig.getPubSubConfig().getGroup());
+ props.put(EventMeshConstants.INSTANCE_NAME, EventMeshUtil.buildMeshClientID(
+ sinkConfig.getPubSubConfig().getGroup(), config.getEventMeshCluster()));
+ props.put(EventMeshConstants.EVENT_MESH_IDC, config.getEventMeshIDC());
+ consumer.init(props);
+
+ consumer.subscribe(sinkConfig.getPubSubConfig().getSubject());
+
+ consumer.registerEventListener(new EventListener() {
+ @Override
+ public void consume(CloudEvent event, AsyncConsumeContext context) {
+ try {
+ // 1. Egress Pipeline
+ String pipelineKey = sinkConfig.getPubSubConfig().getGroup() + "-" + event.getSubject();
+ event = egressProcessor.process(event, pipelineKey);
+
+ if (event == null) {
+ context.commit(EventMeshAction.CommitMessage);
+ return;
+ }
+
+ ((SinkWorker) worker).handle(event);
+ context.commit(EventMeshAction.CommitMessage);
+ } catch (Exception e) {
+ log.error("Error in Sink processing", e);
+ context.commit(EventMeshAction.ReconsumeLater);
+ }
+ }
+ });
+
+ } else if (Application.isSource(connector.getClass())) {
+ worker = new SourceWorker((Source) connector, (SourceConfig) connectorConfig);
+
+ // Initialize Producer for Source
+ SourceConfig sourceConfig = (SourceConfig) connectorConfig;
+ producer = new MQProducerWrapper(config.getEventMeshStoragePluginType());
+ Properties props = new Properties();
+ props.put(EventMeshConstants.PRODUCER_GROUP, sourceConfig.getPubSubConfig().getGroup());
+ props.put(EventMeshConstants.INSTANCE_NAME, EventMeshUtil.buildMeshClientID(
+ sourceConfig.getPubSubConfig().getGroup(), config.getEventMeshCluster()));
+ props.put(EventMeshConstants.EVENT_MESH_IDC, config.getEventMeshIDC());
+ producer.init(props);
+
+ ((SourceWorker) worker).setPublisher((event, callback) -> {
+ try {
+ // 1. Ingress Pipeline
+ String pipelineKey = sourceConfig.getPubSubConfig().getGroup() + "-" + event.getSubject();
+ event = ingressProcessor.process(event, pipelineKey);
+
+ if (event == null) {
+ SendResult result = new SendResult();
+ result.setTopic(event.getSubject());
+ result.setMessageId(event.getId());
+ callback.onSuccess(result);
+ return;
Review Comment:
When the ingress pipeline filters out an event, this code sets result fields using `event` after it has been replaced with null, causing a NullPointerException. Preserve the original event/topic/id before calling ingressProcessor.process, and use those values when reporting a filtered success.
##########
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/BatchPublishCloudEventProcessor.java:
##########
@@ -58,34 +59,68 @@ public void handleCloudEvent(CloudEventBatch cloudEventBatch, EventEmitter<Cloud
List<io.cloudevents.CloudEvent> cloudEvents = grpcCommandProtocolAdaptor.toBatchCloudEvent(
new BatchEventMeshCloudEventWrapper(cloudEventBatch));
+ // Create BatchProcessResult to track success/filtered/failed counts
+ final BatchProcessResult batchResult = new BatchProcessResult(cloudEvents.size());
+ final String finalTopic = topic;
+ final String finalProducerGroup = producerGroup;
+
for (io.cloudevents.CloudEvent event : cloudEvents) {
String seqNum = event.getId();
String uniqueId = (event.getExtension(ProtocolKey.UNIQUE_ID) == null) ? "" : event.getExtension(ProtocolKey.UNIQUE_ID).toString();
- ProducerManager producerManager = eventMeshGrpcServer.getProducerManager();
- EventMeshProducer eventMeshProducer = producerManager.getEventMeshProducer(producerGroup);
-
- SendMessageContext sendMessageContext = new SendMessageContext(seqNum, event, eventMeshProducer, eventMeshGrpcServer);
+ String eventTopic = event.getSubject();
- eventMeshGrpcServer.getEventMeshGrpcMetricsManager().recordSendMsgToQueue();
- long startTime = System.currentTimeMillis();
- eventMeshProducer.send(sendMessageContext, new SendCallback() {
+ try {
+ // Apply Ingress Pipeline (Filter -> Transformer -> Router)
+ String pipelineKey = finalProducerGroup + "-" + eventTopic;
+ io.cloudevents.CloudEvent processedEvent = eventMeshGrpcServer.getEventMeshServer()
+ .getIngressProcessor().process(event, pipelineKey);
- @Override
- public void onSuccess(SendResult sendResult) {
- long endTime = System.currentTimeMillis();
- log.info("message|eventMesh2mq|REQ|BatchSend|send2MQCost={}ms|topic={}|bizSeqNo={}|uniqueId={}",
- endTime - startTime, topic, seqNum, uniqueId);
+ if (processedEvent == null) {
+ // Message filtered by pipeline
+ batchResult.incrementFiltered();
+ log.info("Batch message filtered by pipeline: topic={}, seqNum={}, uniqueId={}",
+ eventTopic, seqNum, uniqueId);
+ continue;
}
- @Override
- public void onException(OnExceptionContext context) {
- long endTime = System.currentTimeMillis();
- log.error("message|eventMesh2mq|REQ|BatchSend|send2MQCost={}ms|topic={}|bizSeqNo={}|uniqueId={}",
- endTime - startTime, topic, seqNum, uniqueId, context.getException());
- }
- });
+ // Topic may have been changed by Router
+ final String routedTopic = processedEvent.getSubject();
+
+ ProducerManager producerManager = eventMeshGrpcServer.getProducerManager();
+ EventMeshProducer eventMeshProducer = producerManager.getEventMeshProducer(finalProducerGroup);
+
+ SendMessageContext sendMessageContext = new SendMessageContext(seqNum, processedEvent, eventMeshProducer, eventMeshGrpcServer);
+
+ eventMeshGrpcServer.getEventMeshGrpcMetricsManager().recordSendMsgToQueue();
+ long startTime = System.currentTimeMillis();
+ eventMeshProducer.send(sendMessageContext, new SendCallback() {
+
+ @Override
+ public void onSuccess(SendResult sendResult) {
+ batchResult.incrementSuccess();
+ long endTime = System.currentTimeMillis();
+ log.info("message|eventMesh2mq|REQ|BatchSend|send2MQCost={}ms|topic={}|bizSeqNo={}|uniqueId={}",
+ endTime - startTime, routedTopic, seqNum, uniqueId);
+ }
+
+ @Override
+ public void onException(OnExceptionContext context) {
+ batchResult.incrementFailed(seqNum);
+ long endTime = System.currentTimeMillis();
+ log.error("message|eventMesh2mq|REQ|BatchSend|send2MQCost={}ms|topic={}|bizSeqNo={}|uniqueId={}",
+ endTime - startTime, routedTopic, seqNum, uniqueId, context.getException());
+ }
+ });
+ } catch (Exception e) {
+ batchResult.incrementFailed(seqNum);
+ log.error("Batch message pipeline exception: topic={}, seqNum={}, uniqueId={}",
+ eventTopic, seqNum, uniqueId, e);
+ }
}
- ServiceUtils.sendResponseCompleted(StatusCode.SUCCESS, "batch publish success", emitter);
+
+ ServiceUtils.sendResponseCompleted(StatusCode.SUCCESS,
+ "batch publish success: " + batchResult.toSummary(), emitter);
+ log.info("Batch publish completed: topic={}, result={}", finalTopic, batchResult.toSummary());
Review Comment:
The gRPC response message includes batchResult.toSummary(), but batchResult is updated asynchronously in SendCallback handlers. This means the summary sent to the client may not reflect actual send outcomes. Consider waiting for sends to complete (with timeout) before responding, or only reporting pre-send pipeline results.
##########
eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/SourceWorker.java:
##########
@@ -198,16 +210,42 @@ public void startPollAndSend() {
// retry until MAX_RETRY_TIMES is reached
while (retryTimes < MAX_RETRY_TIMES) {
try {
- Package sendResult = eventMeshTCPClient.publish(event, 3000);
- if (sendResult.getHeader().getCode() == OPStatus.SUCCESS.getCode()) {
- // publish success
- // commit record
+ if (this.publisher != null) {
+ CountDownLatch latch = new CountDownLatch(1);
+ final Throwable[] exception = new Throwable[1];
+ publisher.publish(event, new SendMessageCallback() {
+ @Override
+ public void onSuccess(SendResult result) {
+ latch.countDown();
+ }
+
+ @Override
+ public void onException(SendExceptionContext context) {
+ exception[0] = context.getCause();
+ latch.countDown();
+ }
+ });
+ latch.await();
+ if (exception[0] != null) {
+ throw exception[0];
+ }
+
Review Comment:
The embedded-publisher path blocks on latch.await() with no timeout. If the publisher never invokes the callback (or a bug occurs), SourceWorker will hang indefinitely and stop committing offsets. Consider using await(timeout) and handling InterruptedException by restoring the interrupt flag and treating it as a send failure.
##########
docs/plugins/core-engines-configuration.md:
##########
@@ -0,0 +1,138 @@
+# EventMesh Core Engines Configuration Guide
+
+EventMesh provides powerful core engines (`Filter`, `Transformer`, `Router`) to dynamically process messages. These engines are configured via **MetaStorage** (Governance Center, e.g., Nacos, Etcd), supporting on-demand loading and hot-reloading.
+
+## 0. Core Concepts
+
+Before configuration, it is important to understand the specific role of each engine in the message flow:
+
+* **Filter (The Gatekeeper)**: Decides **"Whether to pass"**.
+ * It inspects the message (CloudEvent) attributes. If the message matches the rules, it passes; otherwise, it is dropped.
+ * *Use Case*: Block debug logs from production traffic; Only subscribe to specific event types.
+
+* **Transformer (The Translator)**: Decides **"What it looks like"**.
+ * It modifies the message content (Payload or Metadata) according to templates or scripts.
+ * *Use Case*: Convert XML to JSON; Mask sensitive data (PII); Adapt legacy protocols to new standards.
+
+* **Router (The Dispatcher)**: Decides **"Where to go"**.
+ * It dynamically changes the destination (Topic) of the message.
+ * *Use Case*: Route traffic to a Canary/Gray release topic; Route high-priority orders to a dedicated queue.
+
+---
+
+## 1. Overview
+
+The configuration is not in local property files but distributed via the MetaStorage. EventMesh listens to specific **Keys** based on client Groups.
+
+- **Data Source**: Configured via `eventMesh.metaStorage.plugin.type`.
+- **Loading Mechanism**: Lazy loading & Hot-reloading.
+- **Key Format**: `{EnginePrefix}-{GroupName}-{TopicName}`.
+- **Value Format**: JSON Array.
+- **Pipeline Key**: The engines are invoked using a pipeline key of format `{GroupName}-{TopicName}`, which is used to look up configurations with the prefix.
Review Comment:
This doc states the key format is `{EnginePrefix}-{GroupName}-{TopicName}`, but the sections below describe keys like `router-{producerGroup}` / `filter-{producerGroup}` with topic contained inside the JSON value. Please reconcile the documented key format with the actual engine implementations to avoid misconfiguration.
##########
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageV2Processor.java:
##########
@@ -247,12 +274,12 @@ public void onException(OnExceptionContext context) {
eventMeshHTTPServer.getHttpRetryer().newTimeout(sendMessageContext, 10, TimeUnit.SECONDS);
summaryMetrics.recordBatchSendMsgCost(batchEndTime - batchStartTime);
BATCH_MESSAGE_LOGGER.error(
- "batchMessageV2|eventMesh2mq|REQ|ASYNC|bizSeqNo={}|send2MQCost={}ms|topic={}",
- bizNo, batchEndTime - batchStartTime, topic, e);
+ "batchMessageV2|eventMesh2mq|REQ|ASYNC|bizSeqNo={}|send2MQCost={}ms|topic={}|result={}",
+ finalBizNo, batchEndTime - batchStartTime, finalTopic, batchResult.toSummary(), e);
}
completeResponse(request, asyncContext, sendMessageBatchV2ResponseHeader,
- EventMeshRetCode.SUCCESS, null, SendMessageBatchV2ResponseBody.class);
+ EventMeshRetCode.SUCCESS, batchResult.toSummary(), SendMessageBatchV2ResponseBody.class);
}
Review Comment:
BatchSendMessageV2Processor returns batchResult.toSummary() immediately after initiating an async send, but batchResult is updated in the SendCallback. This makes the returned summary unreliable (typically success=0/failed=0). Consider completing the response after the callback (or avoid returning async-dependent counts).
##########
docs/unified-runtime-design.md:
##########
@@ -0,0 +1,201 @@
+# Unified Runtime Design & Usage Guide
+
+## 1. Overview
+The EventMesh Unified Runtime consolidates the capabilities of the core EventMesh Runtime (Protocol handling), Connectors (Source/Sink), and Functions (Filter/Transformer/Router) into a single, cohesive process. This eliminates the need for separate deployments for Connectors ("Runtime V2") and simplifies the architecture.
+
+## 2. Architecture: The Unified Processing Pipeline
+
+The system implements a symmetrical processing chain for both event production (Ingress) and consumption (Egress), but the entry/exit points differ based on the client type (SDK vs. Connector).
+
+### 2.1 Ingress Pipeline (Production)
+
+**Entry Points:**
+* **SDK Client**: Interacts with the Runtime via **Protocol Servers** (TCP/HTTP/gRPC). The Protocol Server receives the request and passes the event to the pipeline.
+* **Source Connector**: Loaded directly into the Runtime as a **Plugin**. The Source Connector pulls data from external systems and internally injects events into the pipeline.
+
+**Flow:**
+`[Entry: Protocol Server (SDK) OR Source Plugin (Connector)] -> [IngressProcessor] -> [Storage]`
+
+**IngressProcessor Pipeline:**
+`[Filter] -> [Transformer] -> [Router]`
+
+1. **Entry**:
+ * **SDK**: Request received by `EventMeshTCPServer`, `EventMeshHTTPServer`, or `EventMeshGrpcServer`.
+ * **Connector**: `SourceWorker` pulls data and converts it to a CloudEvent.
+2. **IngressProcessor**: Encapsulates the unified 3-stage pipeline:
+ * **Filter**: The `FilterEngine` evaluates the event against configured rules. If unmatched, returns null (event dropped).
+ * **Transformer**: The `TransformerEngine` transforms the event payload (e.g., JSON manipulation) if a rule exists.
+ * **Router**: The `RouterEngine` determines the target topic/destination.
+3. **Storage**: The processed event is persisted to the Storage Plugin (RocketMQ, Kafka, etc.).
+
+### 2.2 Egress Pipeline (Consumption)
+
+**Exit Points:**
+* **SDK Client**: The Runtime pushes events to connected SDK clients via the active **Protocol Server** connection.
+* **Sink Connector**: Loaded directly into the Runtime as a **Plugin**. The Runtime passes events to the `SinkWorker`, which writes to external systems.
+
+**Flow:**
+`[Storage] -> [EgressProcessor] -> [Exit: Protocol Server (SDK) OR Sink Plugin (Connector)]`
+
+**EgressProcessor Pipeline:**
+`[Filter] -> [Transformer]`
+
+1. **Storage**: Event retrieved from the storage queue.
+2. **EgressProcessor**: Encapsulates the 2-stage pipeline (no Router on egress):
+ * **Filter**: Evaluated against the consumer group's filter rules. If unmatched, returns null (event not delivered).
+ * **Transformer**: Payload transformed according to the consumer group's needs.
+3. **Exit**:
+ * **SDK**: Event pushed to client via TCP/HTTP/gRPC session.
+ * **Connector**: Event passed to `SinkWorker` for external delivery.
+
+### 2.3 Protocol Processor Migration Status
+
+All protocol processors now use the unified IngressProcessor/EgressProcessor architecture:
+
+**TCP Protocol**: ✅ Complete
+* `ClientGroupWrapper` - Integrated both Ingress (send) and Egress (consume)
+
+**HTTP Protocol**: ✅ Complete
+* `SendAsyncEventProcessor` - Ingress pipeline
+* `SendAsyncMessageProcessor` - Ingress pipeline
+* `SendSyncMessageProcessor` - Ingress pipeline
+* `BatchSendMessageProcessor` - Ingress pipeline with batch statistics
+* `BatchSendMessageV2Processor` - Ingress pipeline with batch statistics
+
+**gRPC Protocol**: ✅ Complete
+* `PublishCloudEventsProcessor` - Ingress pipeline
+* `BatchPublishCloudEventProcessor` - Ingress pipeline with batch statistics
+* `RequestCloudEventProcessor` - Bidirectional (Ingress for request, Egress for response)
+
+**Connectors**: ✅ Complete
+* `SourceWorker` - Ingress pipeline
+* `SinkWorker` - Egress pipeline
+
+## 3. Configuration
+
+### 3.1 Enabling Connectors
+To enable the embedded Connector runtime, update `eventmesh.properties`:
+
+```properties
+# Enable the connector plugin
+eventMesh.connector.plugin.enable=true
+
+# Specify the connector type (source or sink) and name (SPI name)
+eventMesh.connector.plugin.type=source
+eventMesh.connector.plugin.name=my-source-connector
+```
Review Comment:
The connector enable property name in this example (`eventMesh.connector.plugin.enable`) doesn't match the config field added in CommonConfiguration (`connector.plugin.enabled`). This will confuse users and may prevent the connector bootstrap from being enabled. Align the documented property key with the actual configuration key.
##########
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshServer.java:
##########
@@ -146,6 +172,23 @@ public void init() throws Exception {
if (configuration.isEventMeshServerMetaStorageEnable()) {
metaStorage.init();
}
+
+ // filter and transformer engine init
+ filterEngine = new FilterEngine(metaStorage);
+ filterEngine.start();
+ transformerEngine = new TransformerEngine(metaStorage);
+ transformerEngine.start();
+ routerEngine = new RouterEngine(metaStorage);
+ routerEngine.start();
+
+ // ingress and egress processor init
+ ingressProcessor = new org.apache.eventmesh.runtime.core.protocol.IngressProcessor(filterEngine, transformerEngine, routerEngine);
+ egressProcessor = new org.apache.eventmesh.runtime.core.protocol.EgressProcessor(filterEngine, transformerEngine);
+
Review Comment:
FilterEngine/TransformerEngine/RouterEngine are started unconditionally, even when eventMeshServerMetaStorageEnable is false (metaStorage.init/start are skipped). Since engine.start() immediately calls metaStorage.getMetaData and registers listeners, this can fail when MetaStorage is disabled/uninitialized. Consider initializing/starting these engines only when MetaStorage is enabled, or make engines no-op safely when MetaStorage is disabled.
##########
eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/boot/RouterEngineTest.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * 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.eventmesh.runtime.boot;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
Review Comment:
RouterEngineTest has unused static imports (e.g., eq, mock) which will fail compilation. Remove unused imports; also consider updating the assertion once RouterEngine reads textual routerConfig without adding extra quotes.
##########
eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncEventProcessorTest.java:
##########
@@ -0,0 +1,266 @@
+/*
+ * 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.eventmesh.runtime.core.protocol.http.processor;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import org.apache.eventmesh.api.SendCallback;
+import org.apache.eventmesh.common.protocol.ProtocolTransportObject;
+import org.apache.eventmesh.common.protocol.http.HttpEventWrapper;
+import org.apache.eventmesh.common.protocol.http.common.ProtocolKey;
+import org.apache.eventmesh.protocol.api.ProtocolAdaptor;
+import org.apache.eventmesh.protocol.api.ProtocolPluginFactory;
+import org.apache.eventmesh.runtime.a2a.A2APublishSubscribeService;
+import org.apache.eventmesh.runtime.acl.Acl;
+import org.apache.eventmesh.runtime.boot.EventMeshHTTPServer;
+import org.apache.eventmesh.runtime.boot.EventMeshServer;
+import org.apache.eventmesh.runtime.boot.FilterEngine;
+import org.apache.eventmesh.runtime.boot.HTTPTrace.TraceOperation;
+import org.apache.eventmesh.runtime.boot.RouterEngine;
+import org.apache.eventmesh.runtime.boot.TransformerEngine;
+import org.apache.eventmesh.runtime.configuration.EventMeshHTTPConfiguration;
+import org.apache.eventmesh.runtime.core.protocol.IngressProcessor;
+import org.apache.eventmesh.runtime.core.protocol.http.async.AsyncContext;
+import org.apache.eventmesh.runtime.core.protocol.http.retry.HttpRetryer;
+import org.apache.eventmesh.runtime.core.protocol.producer.EventMeshProducer;
+import org.apache.eventmesh.runtime.core.protocol.producer.ProducerManager;
+import org.apache.eventmesh.runtime.core.protocol.producer.SendMessageContext;
+import org.apache.eventmesh.runtime.metrics.http.EventMeshHttpMetricsManager;
+import org.apache.eventmesh.runtime.metrics.http.HttpMetrics;
+import org.apache.eventmesh.runtime.util.RemotingHelper;
+
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.Map;
Review Comment:
This test imports InetSocketAddress but never uses it, which will fail compilation. Remove the unused import (and any other unused imports that may have been introduced).
--
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: issues-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org