diff --git a/docs/eventmesh-uni-architecture-redesign.md b/docs/eventmesh-uni-architecture-redesign.md index f2cdcea752..25460b09ac 100644 --- a/docs/eventmesh-uni-architecture-redesign.md +++ b/docs/eventmesh-uni-architecture-redesign.md @@ -4882,7 +4882,106 @@ EventMesh 实例本地自采负载指标(`LoadMeter`): 1. **单测**:`EventMeshFrame` 全 msgType 互转(12 例);`OffsetStore` 两 key 空间共存;`LoadMeter` 指标 + 每 client 画像;`ClusterMembership` 心跳负载;dispatch 管线 Frame 化(ReliableDispatcher/SubscriptionManager/ClusterCoordinator)。 2. **E2E**(真 broker):streaming 多轮 + Mode 2 pub/sub + **普通 pub/sub(RocketMQ5BrokerIntegrationTest 2/2)**全绿,内部全程 EventMeshFrame 往返正确;**LegacyTcpClientIntegrationTest(旧 TCP SDK)全绿**(MeshMessage↔Frame 直接转换)。 -3. **构建**:系统 gradle 8.5 + WEOA Nexus(offline)。 +3. **构建**:系统 gradle 8.5 + WEOA Nexus(offline)。 + + +### 19.6 #5299 验收矩阵：单协议路径 + ingress/egress 适配 + + + + + +#### 协议状态标签 + + + + + +| 标签 | 含义 | 当前协议 | + + +|------|------|----------| + + +| **primary** | ingress/egress 全程经 `FrameAdaptor` SPI，内部全程 `EventMeshFrame` | CloudEvents（HTTP/SSE/WS）、MeshMessage（legacy TCP） | + + +| **beta** | 已有 `FrameAdaptor`，但端到端链路未完全收口 | A2A（JSON-RPC 2.0） | + + +| **legacy** | 仅保留兼容桥，不再作为内部表示 | CloudEvent 作为内部中间表示（已废弃，见 §19.1） | + + + + + +#### 路径验收矩阵 + + + + + +| 路径 | ingress | egress | 状态 | + + +|------|---------|--------|------| + + +| HTTP publish / publishBatch / lite publish / lite poll / request / reply | `FrameAdaptors.get("cloudevents").toFrame(...)` | `FrameAdaptor.toCloudEventsJson(...)` | ✅ primary | + + +| ingress 安全链（TokenAuth / Acl / SignatureVerifier） | 直接读 `frame.attributes()` | — | ✅ primary | + + +| legacy TCP ingress | `MeshMessagePackageRouter` → `FrameAdaptors.get("meshmessage").toFrameSilent(pkg)` | — | ✅ primary | + + +| legacy TCP egress | `NettyTcpPushChannel` → `FrameAdaptors.get("meshmessage").fromFrameSilent(frame)` | MeshMessage `Package` | ✅ primary | + + +| A2A gateway | `A2AFrameAdaptor` | A2A JSON-RPC bytes | 🟡 beta | + + + + + +#### 子 PR 落地情况 + + + + + +| 子 PR | 内容 | 状态 | + + +|-------|------|------| + + +| A | `UniHttpServer` 8 个 ingress 端点改经 `FrameAdaptor` SPI；`UniIngressService` 新增 `publishBatchFrames` / `publishLiteFrame` / `pollLiteFrames` / `requestFrame` / `replyFrame` 五个 `EventMeshFrame` typed 方法 | ✅ | + + +| B | 安全链（`IngressFilter` / `FilterChain` / `AclFilter` / `TokenAuthFilter` / `SignatureVerifierFilter`）改吃 `EventMeshFrame`；租户 / 签名 / token 直接读 `frame.attributes()`。CloudEvent 重载保留为 `@Deprecated` 桥 | ✅ | + + +| C | TCP egress：`TcpFrameCodec.encodePush` / `TcpPushChannel.deliver` 改吃 `EventMeshFrame`（去掉 `frame.toCloudEvent()` 往返）；删除死代码 `CloudEventToPackageBody` / `MeshEventToPackageBody`；`UniTcpServer` 去掉未使用的 `bodyMapper` 构造参数 | ✅ | + + +| D | 本节验收矩阵 + 协议状态标签 | ✅ | + + + + + +#### 随 #5299 删除 / 废弃 + + + + + +- `CloudEventToPackageBody`、`MeshEventToPackageBody`：CloudEvent 时代的 TCP egress 编码接口。`NettyTcpPushChannel` 接管 egress 后成为死代码，随 Sub-PR C 删除。 + + +- `IngressFilter.check(CloudEvent, FilterContext)`：`@Deprecated` 桥接，仅供尚未迁移的自定义 filter 编译通过；等 A2A（beta）收口后移除。 --- diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/http/UniHttpServer.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/http/UniHttpServer.java index e67bdd540f..1b813a4aba 100644 --- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/http/UniHttpServer.java +++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/http/UniHttpServer.java @@ -17,6 +17,9 @@ package org.apache.eventmesh.runtime.http; +import org.apache.eventmesh.common.protocol.ByteTransport; +import org.apache.eventmesh.common.wire.EventMeshFrame; +import org.apache.eventmesh.protocol.api.FrameAdaptors; import org.apache.eventmesh.runtime.admin.UniAdminService; import org.apache.eventmesh.runtime.ingress.UniIngressService; import org.apache.eventmesh.runtime.push.BufferedEvent; @@ -35,10 +38,6 @@ import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; -import io.cloudevents.CloudEvent; -import io.cloudevents.core.provider.EventFormatProvider; -import io.cloudevents.jackson.JsonFormat; - import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.node.ObjectNode; @@ -249,10 +248,13 @@ private boolean checkSecurity(HttpExchange exchange, String topic, String client org.apache.eventmesh.runtime.security.FilterContext ctx = new org.apache.eventmesh.runtime.security.FilterContext(topic, clientId, tenant, credential, exchange.getRemoteAddress().getAddress().getHostAddress()); - // For non-publish endpoints there's no CloudEvent body to check; use a minimal stub. - io.cloudevents.CloudEvent stubEvent = io.cloudevents.core.builder.CloudEventBuilder.v1() - .withId("security-check").withSource(java.net.URI.create("eventmesh")).withType("security").build(); - org.apache.eventmesh.runtime.security.FilterVerdict verdict = filterChain.check(stubEvent, ctx); + // For non-publish endpoints there's no EventMeshFrame to check; build a minimal + // security-check frame so the filter chain has *something* to evaluate. The filters + // themselves read tenant / credential from the FilterContext, so a frame with empty + // attributes is sufficient for the auth/acl decision (#5299 sub-PR B). + org.apache.eventmesh.common.wire.EventMeshFrame stubFrame = + org.apache.eventmesh.common.wire.EventMeshFrame.event(java.util.Collections.emptyMap(), new byte[0]); + org.apache.eventmesh.runtime.security.FilterVerdict verdict = filterChain.check(stubFrame, ctx); if (!verdict.isAllowed()) { writeJson(exchange, verdict.getRejectStatus(), error(verdict.getReason())); return false; @@ -285,28 +287,34 @@ private void publish(HttpExchange exchange) throws IOException { return; } byte[] body = readAll(exchange); - CloudEvent event; + EventMeshFrame frame; try { - event = EventFormatProvider.getInstance().resolveFormat(JsonFormat.CONTENT_TYPE).deserialize(body); - } catch (RuntimeException e) { + // Ingress: structured CloudEvents JSON bytes → internal EventMeshFrame. + // (#5299: runtime no longer touches io.cloudevents.CloudEvent directly on the ingress + // path; the protocol adaptor owns the conversion.) + frame = FrameAdaptors.get("cloudevents").toFrame(new ByteTransport(body)); + } catch (RuntimeException | org.apache.eventmesh.protocol.api.exception.ProtocolHandleException e) { writeJson(exchange, 400, error("invalid CloudEvent: " + e.getMessage())); return; } // Security filter chain (§4.5): auth/acl/signature run before the event enters the pipeline. + // #5299 sub-PR B: filters now read directly from EventMeshFrame.attributes() — no more + // CE bridge. Tenant still comes from the CloudEvent extension ("emtenantid") which the + // cloudevents FrameAdaptor round-trips into frame attributes under the same key. if (filterChain != null) { String credential = exchange.getRequestHeaders().getFirst("Authorization"); - String tenant = event.getExtension("emtenantid") != null ? event.getExtension("emtenantid").toString() : null; + String tenant = frame.attributes().get("emtenantid"); org.apache.eventmesh.runtime.security.FilterContext ctx = new org.apache.eventmesh.runtime.security.FilterContext(topic, null, tenant, credential, exchange.getRemoteAddress().getAddress().getHostAddress()); - org.apache.eventmesh.runtime.security.FilterVerdict verdict = filterChain.check(event, ctx); + org.apache.eventmesh.runtime.security.FilterVerdict verdict = filterChain.check(frame, ctx); if (!verdict.isAllowed()) { writeJson(exchange, verdict.getRejectStatus(), error(verdict.getReason())); return; } } try { - ingress.publish(topic, event).get(10, TimeUnit.SECONDS); + ingress.publish(topic, frame).get(10, TimeUnit.SECONDS); writeJson(exchange, 202, ack("accepted")); } catch (Exception e) { // §6.6: a RateLimitedException (per-topic token bucket exhausted) is a 429, not a 500 — @@ -337,12 +345,14 @@ private void publishBatch(HttpExchange exchange) throws IOException { writeJson(exchange, 400, error("expected a CloudEvent JSON array")); return; } - java.util.List events = new java.util.ArrayList<>(node.size()); + // Batch ingress: each element is a CloudEvents-JSON object → internal EventMeshFrame. + // (#5299) + java.util.List frames = new java.util.ArrayList<>(node.size()); for (com.fasterxml.jackson.databind.JsonNode el : node) { - events.add(EventFormatProvider.getInstance().resolveFormat(JsonFormat.CONTENT_TYPE) - .deserialize(mapper.writeValueAsBytes(el))); + frames.add(FrameAdaptors.get("cloudevents") + .toFrame(new ByteTransport(mapper.writeValueAsBytes(el)))); } - ingress.publishBatch(topic, events).get(30, TimeUnit.SECONDS); + ingress.publishBatchFrames(topic, frames).get(30, TimeUnit.SECONDS); writeJson(exchange, 202, ack("accepted")); } catch (Exception e) { if (isRateLimited(e)) { @@ -484,13 +494,16 @@ private void request(HttpExchange exchange) throws IOException { String topic = param(exchange.getRequestURI(), "topic"); long timeout = longParam(exchange.getRequestURI(), "timeoutMs", 30_000L); try { - CloudEvent event = EventFormatProvider.getInstance().resolveFormat(JsonFormat.CONTENT_TYPE) - .deserialize(readAll(exchange)); + // Ingress: structured CloudEvents JSON body → internal EventMeshFrame. (#5299) + EventMeshFrame event = FrameAdaptors.get("cloudevents") + .toFrame(new ByteTransport(readAll(exchange))); if (!checkSecurity(exchange, topic, null)) { return; } - CloudEvent reply = ingress.request(topic, event, timeout); - byte[] replyBytes = EventFormatProvider.getInstance().resolveFormat(JsonFormat.CONTENT_TYPE).serialize(reply); + // request-reply: send Frame in, get Frame back, then serialize to CloudEvents JSON + // for the response body via the egress adapter. + EventMeshFrame reply = ingress.requestFrame(topic, event, timeout); + byte[] replyBytes = FrameAdaptors.toCloudEventsJson(reply); exchange.getResponseHeaders().add("Content-Type", "application/cloudevents+json"); exchange.sendResponseHeaders(200, replyBytes.length); try (OutputStream os = exchange.getResponseBody()) { @@ -509,14 +522,15 @@ private void reply(HttpExchange exchange) throws IOException { try { JsonNode body = readJson(exchange); String corrId = text(body, "correlationId"); - CloudEvent replyEvent = EventFormatProvider.getInstance().resolveFormat(JsonFormat.CONTENT_TYPE) - .deserialize(mapper.writeValueAsBytes(body.get("event"))); + // Ingress: CloudEvents JSON → internal EventMeshFrame. (#5299) + EventMeshFrame replyEvent = FrameAdaptors.get("cloudevents") + .toFrame(new ByteTransport(mapper.writeValueAsBytes(body.get("event")))); // §17.6 reply routing (sticky model - no cross-instance forwarding). // Cross-instance reply forwarding is REMOVED with the forward path: the client posts // the reply to the instance it sent the request to (pinned via instanceUrl); a reply // landing on the wrong instance 404s (unknown correlationId) and the caller retries // on the correct instance. - writeJson(exchange, ingress.reply(corrId, replyEvent) ? 200 : 404, ack("ok")); + writeJson(exchange, ingress.replyFrame(corrId, replyEvent) ? 200 : 404, ack("ok")); } catch (Exception e) { writeJson(exchange, 500, error("reply error: " + e.getMessage())); } @@ -615,9 +629,10 @@ private void litePublish(HttpExchange exchange) throws IOException { return; } try { - CloudEvent event = EventFormatProvider.getInstance().resolveFormat(JsonFormat.CONTENT_TYPE) - .deserialize(readAll(exchange)); - ingress.publishLite(parent, lite, event).get(10, TimeUnit.SECONDS); + // Ingress: structured CloudEvents JSON body → internal EventMeshFrame. (#5299) + EventMeshFrame event = FrameAdaptors.get("cloudevents") + .toFrame(new ByteTransport(readAll(exchange))); + ingress.publishLiteFrame(parent, lite, event).get(10, TimeUnit.SECONDS); writeJson(exchange, 202, ack("accepted")); } catch (Exception e) { writeJson(exchange, 500, error("lite publish failed: " + e.getMessage())); @@ -646,11 +661,12 @@ private void litePoll(HttpExchange exchange) throws IOException { try { int max = intParam(exchange.getRequestURI(), "max", 100); long timeoutMs = longParam(exchange.getRequestURI(), "timeoutMs", 1000L); - List events = ingress.pollLite(parent, lite, max, timeoutMs); + // Egress: drain EventMeshFrames from the LMQ, serialize each as CloudEvents JSON + // via the egress adapter. (#5299) + List events = ingress.pollLiteFrames(parent, lite, max, timeoutMs); com.fasterxml.jackson.databind.node.ArrayNode arr = mapper.createArrayNode(); - for (CloudEvent e : events) { - arr.add(mapper.readTree(EventFormatProvider.getInstance() - .resolveFormat(JsonFormat.CONTENT_TYPE).serialize(e))); + for (EventMeshFrame e : events) { + arr.add(mapper.readTree(FrameAdaptors.toCloudEventsJson(e))); } writeJson(exchange, 200, arr); } catch (NumberFormatException e) { diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/ingress/UniIngressService.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/ingress/UniIngressService.java index 6d47a7d258..f6e617abba 100644 --- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/ingress/UniIngressService.java +++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/ingress/UniIngressService.java @@ -1,903 +1,988 @@ -/* - * 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.ingress; - -import org.apache.eventmesh.api.SendCallback; -import org.apache.eventmesh.api.SendResult; -import org.apache.eventmesh.api.storage.MeshStoragePlugin; -import org.apache.eventmesh.runtime.delivery.DeadLetterSink; -import org.apache.eventmesh.runtime.delivery.PushChannel; -import org.apache.eventmesh.runtime.delivery.ReliableDispatcher; -import org.apache.eventmesh.runtime.metrics.UniMetrics; -import org.apache.eventmesh.runtime.metrics.UniTrace; -import org.apache.eventmesh.runtime.offset.OffsetStore; -import org.apache.eventmesh.runtime.push.BufferedEvent; -import org.apache.eventmesh.runtime.push.LongPollingChannel; -import org.apache.eventmesh.runtime.push.PushService; -import org.apache.eventmesh.runtime.ratelimit.RateLimitedException; -import org.apache.eventmesh.runtime.ratelimit.TokenBucketRateLimiter; -import org.apache.eventmesh.runtime.subscription.CloudEventFilter; -import org.apache.eventmesh.runtime.subscription.DistributionMode; -import org.apache.eventmesh.runtime.subscription.Subscription; -import org.apache.eventmesh.runtime.subscription.SubscriptionManager; - -import java.util.List; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; -import java.util.concurrent.atomic.AtomicLong; - -import io.cloudevents.CloudEvent; -import io.cloudevents.core.builder.CloudEventBuilder; - -import lombok.extern.slf4j.Slf4j; - -/** - * Facade that wires the uni-architecture layers into the end-to-end CloudEvents-over-MQ - * flow (§6): - * 
- *   publish  ─▶ MeshStoragePlugin.send
- *   pullLoop ─▶ MeshStoragePlugin.poll ─▶ SubscriptionManager.targetsFor ─▶ ReliableDispatcher.deliver ─▶ PushService
- *   poll     ◀─ client long-polls PushService
- *   ack      ─▶ PushService.ack ─▶ ReliableDispatcher.ack ─▶ OffsetStore (offset advances only on ACK)
- *
- * - *

This is the orchestration core; the actual HTTP endpoint wiring (§6 UniIngressHandler) - * delegates here. Phase-1 thin storage adapter means {@code partition} is {@code -1} and the offset - * is a per-topic monotonic logical counter until the native storage reimplementation supplies real - * partition/offset (Phase 1 step 2).

- */ -@Slf4j -public class UniIngressService { - - private final MeshStoragePlugin storage; - private final OffsetStore offsetStore; - private final SubscriptionManager subscriptionManager; - private final ReliableDispatcher dispatcher; - private final PushService pushService; - private volatile org.apache.eventmesh.runtime.cluster.ClusterCoordinator cluster; - private volatile org.apache.eventmesh.runtime.cluster.PartitionOwnership partitionOwnership; - /** Self-collected load metrics for session-distribution load balancing (§3). Null until wired. */ - private volatile LoadMeter loadMeter; - - /** Per-topic poll stats for the {@code poll_idle_ratio} gauge (§13.5.1). */ - private final ConcurrentHashMap pollCount = new ConcurrentHashMap<>(); - private final ConcurrentHashMap pollEmptyCount = new ConcurrentHashMap<>(); - - /** Connector offset store (remote side) — String key → String offset value (§8.9). */ - private final java.util.concurrent.ConcurrentHashMap connectorOffsets = new java.util.concurrent.ConcurrentHashMap<>(); - private final UniMetrics metrics; - - private final ConcurrentHashMap channels = new ConcurrentHashMap<>(); - private final ConcurrentHashMap topicOffsetSeq = new ConcurrentHashMap<>(); - private final ConcurrentHashMap> pendingRequests = new ConcurrentHashMap<>(); - private final AtomicLong requestSeq = new AtomicLong(); - private final ConcurrentHashMap topicLimiters = new ConcurrentHashMap<>(); - - /** - * CloudEvents extension carrying the request's correlation id (§17). - * - *

Named without hyphens because the CloudEvents spec restricts extension attribute names to - * lower-case ASCII letters and digits — the redesign doc's {@code x-em-correlation-id} spelling - * is rejected by the SDK's name validation.

- */ - public static final String EXT_CORRELATION_ID = "emcorrelationid"; - - public UniIngressService(MeshStoragePlugin storage, OffsetStore offsetStore) { - this(storage, offsetStore, new SubscriptionManager(), new PushService(), - ReliableDispatcher.DEFAULT_ACK_TIMEOUT_MS, ReliableDispatcher.DEFAULT_MAX_ATTEMPTS, - System::currentTimeMillis); - } - - /** - * Test-friendly constructor with an injectable clock and retry parameters. - */ - public UniIngressService(MeshStoragePlugin storage, OffsetStore offsetStore, - SubscriptionManager subscriptionManager, PushService pushService, - long ackTimeoutMs, int maxAttempts, java.util.function.LongSupplier clock) { - this.storage = storage; - this.offsetStore = offsetStore; - this.subscriptionManager = subscriptionManager; - this.pushService = pushService; - this.metrics = new UniMetrics(); - this.dispatcher = new ReliableDispatcher(ackTimeoutMs, maxAttempts, clock, offsetStore, deadLetterSink(), - metrics, ReliableDispatcher.DEFAULT_JITTER_RATIO); - // Sub-PR B: re-ACK any in-flight deliveries from a previous JVM so they do not become - // orphans after a crash. Safe on first start (store is empty) and idempotent. - this.dispatcher.recover(); - } - - // ---- connector offset (remote side, §8.9) ---- - - public String getConnectorOffset(String connectorId) { - return connectorOffsets.get(connectorId); - } - - public void putConnectorOffset(String connectorId, String offset) { - connectorOffsets.put(connectorId, offset); - } - - /** - * Publish a CloudEvent to {@code topic} (persisted to MQ). Completes when the storage plugin - * acknowledges the write. - */ - public CompletableFuture publish(String topic, CloudEvent event) { - if (loadMeter != null && event.getData() != null) { - loadMeter.recordInflow(event.getData().toBytes().length); - } - CompletableFuture future = new CompletableFuture<>(); - TokenBucketRateLimiter limiter = topicLimiters.get(topic); - if (limiter != null && !limiter.tryAcquire()) { - metrics.incRateLimited(); - future.completeExceptionally(new RateLimitedException(topic)); - return future; - } - try { - // Ingress boundary: CloudEvent (the public/external format) → EventMeshFrame (the internal - // wire unit). From here on the event is a Frame internally; egress converts back to the - // client's protocol (CloudEvents / MeshMessage) at the delivery boundary. - org.apache.eventmesh.common.wire.EventMeshFrame frame = - org.apache.eventmesh.common.wire.EventMeshFrame.fromCloudEvent(event); - storage.send(topic, frame, new SendCallback() { - @Override - public void onSuccess(SendResult sendResult) { - metrics.incPublish(); - UniTrace.end(UniTrace.startPublish(topic, event)); - future.complete(null); - } - - @Override - public void onException(org.apache.eventmesh.api.exception.OnExceptionContext context) { - metrics.incPublishFailed(); - future.completeExceptionally(context.getException()); - } - }); - } catch (Exception e) { - metrics.incPublishFailed(); - future.completeExceptionally(e); - } - return future; - } - - /** - * Publish an already-internal {@link org.apache.eventmesh.common.wire.EventMeshFrame} (no - * CloudEvent→Frame boundary conversion — the caller built the frame directly, e.g. the legacy - * MeshMessage TCP path via {@code MeshMessageFrameCodec}). Completes when storage acks the write. - */ - public CompletableFuture publish(String topic, org.apache.eventmesh.common.wire.EventMeshFrame frame) { - if (loadMeter != null && frame.data() != null) { - loadMeter.recordInflow(frame.data().length); - } - CompletableFuture future = new CompletableFuture<>(); - TokenBucketRateLimiter limiter = topicLimiters.get(topic); - if (limiter != null && !limiter.tryAcquire()) { - metrics.incRateLimited(); - future.completeExceptionally(new RateLimitedException(topic)); - return future; - } - try { - storage.send(topic, frame, new SendCallback() { - @Override - public void onSuccess(SendResult sendResult) { - metrics.incPublish(); - future.complete(null); - } - - @Override - public void onException(org.apache.eventmesh.api.exception.OnExceptionContext context) { - metrics.incPublishFailed(); - future.completeExceptionally(context.getException()); - } - }); - } catch (Exception e) { - metrics.incPublishFailed(); - future.completeExceptionally(e); - } - return future; - } - - /** - * Publish a batch of CloudEvents to {@code topic} (§13.7.3). Completes when all are persisted. - * The storage plugin's own batching (e.g. Kafka producer accumulator) amortises the per-event - * RTT; this layer fans out the per-event futures and joins them. - */ - public CompletableFuture publishBatch(String topic, java.util.List events) { - if (events == null || events.isEmpty()) { - return CompletableFuture.completedFuture(null); - } - CompletableFuture[] futures = events.stream() - .map(e -> publish(topic, e)) - .toArray(CompletableFuture[]::new); - return CompletableFuture.allOf(futures); - } - - /** - * Register a custom push channel for a client (e.g. a WebHook URL for legacy HTTP webhook-push - * subscribers, or a {@code TcpPushChannel} for legacy TCP clients). Overrides the default - * long-polling channel for this {@code clientId} on subsequent dispatches. - */ - public void registerChannel(String clientId, PushChannel channel) { - pushService.register(clientId); - channels.put(clientId, channel); - } - - /** - * Configure a per-topic rate limit (§6.6). Subsequent publishes above {@code permitsPerSecond} - * (burst {@code capacity}) fail the publish future with {@link RateLimitedException}. - */ - public void setTopicRateLimit(String topic, long capacity, double permitsPerSecond) { - topicLimiters.put(topic, new TokenBucketRateLimiter(capacity, permitsPerSecond)); - } - - /** - * Register a subscription. The subscriber retrieves events via {@link #poll}. - * - * @return the subscription id - */ - public String subscribe(String topic, String clientId, DistributionMode mode, CloudEventFilter filter) { - pushService.register(clientId); - String subId = subscriptionManager.subscribe(topic, clientId, mode, filter, event -> { - // No-op: the reliability layer (ReliableDispatcher) owns delivery, not this fire-and- - // forget callback. Kept to satisfy the SubscriptionManager handler contract. - }); - // §13.2: when clustered, also register cluster-wide so other instances can route events - // for this subscriber here (via ClusterCoordinator → HttpForwarder /internal/forward). - // Without this the HTTP /events/subscribe path is local-only and a publish on a peer - // never reaches this subscriber. - if (cluster != null) { - cluster.subscribe(topic, clientId, mode, null); - } - return subId; - } - - /** - * Remove a subscription. - */ - public boolean unsubscribe(String subscriptionId) { - // Resolve topic + clientId before removing so a clustered subscription can be deregistered - // cluster-wide (the coordinator's unsubscribe is keyed by topic+clientId, not subId). - org.apache.eventmesh.runtime.subscription.Subscription sub = subscriptionManager.getSubscription(subscriptionId); - boolean removed = subscriptionManager.unsubscribe(subscriptionId); - if (removed && cluster != null && sub != null) { - cluster.unsubscribe(sub.getTopic(), sub.getClientId()); - } - return removed; - } - - /** Remove one client's subscription to one topic (HTTP /events/unsubscribe with {clientId, topic}). */ - public boolean unsubscribe(String topic, String clientId) { - boolean removed = subscriptionManager.unsubscribe(topic, clientId); - if (removed && cluster != null) { - cluster.unsubscribe(topic, clientId); - } - return removed; - } - - /** Remove ALL subscriptions for a client (HTTP /events/unsubscribe with {clientId} only). - * Propagates cluster-wide + frees the PushService buffer. */ - public int unsubscribeByClient(String clientId) { - java.util.Set topics = subscriptionManager.topicsForClient(clientId); - int removed = subscriptionManager.unsubscribeByClient(clientId); - if (cluster != null) { - for (String topic : topics) { - cluster.unsubscribe(topic, clientId); - } - } - pushService.removeClient(clientId); - return removed; - } - - /** - * Pump: pull a batch from storage, route to each target subscriber via the reliability layer. - * - * @return number of events pulled - */ - public int pullAndDispatch(String topic, int maxEvents, long timeoutMs) { - pollCount.merge(topic, 1L, Long::sum); - // Multi-instance (§13.2.3): poll only the partitions this instance owns. Single-instance - // (no ownership) or unknown partition count → poll the whole topic (partition -1). - java.util.List owned = partitionOwnership == null ? null : partitionOwnership.ownedPartitions(topic); - int total; - if (owned == null) { - total = pullAndDispatchPartition(topic, -1, maxEvents, timeoutMs); - } else if (owned.isEmpty()) { - total = 0; // owns none -> do not poll (avoids duplicate with the real owners) - } else { - total = 0; - for (int p : owned) { - total += pullAndDispatchPartition(topic, p, maxEvents, timeoutMs); - } - } - if (total == 0) { - pollEmptyCount.merge(topic, 1L, Long::sum); - } - return total; - } - - private int pullAndDispatchPartition(String topic, int partition, int maxEvents, long timeoutMs) { - // Pull EventMeshFrames (internal wire); dispatch each as a Frame through the internal - // pipeline (filter/dispatcher/egress all carry Frame now; egress converts to the client's - // protocol at the wire boundary). - List frames = storage.poll(topic, partition, -1, maxEvents, timeoutMs); - if (frames == null || frames.isEmpty()) { - return 0; - } - long start = System.nanoTime(); - for (org.apache.eventmesh.common.wire.EventMeshFrame f : frames) { - if (isExpired(f)) { - // §13.3.4 TTL: drop expired events instead of dispatching. - log.debug("dropping expired event {} on topic {} (emttl elapsed)", f.attributes().get("id"), topic); - continue; - } - io.opentelemetry.api.trace.Span dispatchSpan = UniTrace.startDispatch(topic, f); - if (cluster != null) { - // Multi-instance: route via the cluster coordinator (local vs cross-instance forward). - cluster.dispatch(topic, f); - } else { - // develop's OffsetExtensions (CE-extension-carried MQ offset) is superseded by the - // Frame architecture: the POP check key rides in frame attributes (empopck) and the - // deferred broker ACK fires on client ACK — same at-least-once goal, Frame-native. - long offset = nextOffset(topic); - // P2 fix: if the frame carries a POP check key (RocketMQ 5.x deferred ACK), build a - // callback that ACKs the broker on client ACK (restoring at-least-once). - String popCk = f.attributes().get("empopck"); - Runnable mqAck = (popCk != null) ? () -> storage.ackPulledMessage(topic, popCk) : null; - for (Subscription target : subscriptionManager.targetsFor(topic, f)) { - dispatcher.deliver(topic, partition, offset, f, target.getClientId(), - channelFor(target.getClientId()), mqAck); - } - } - UniTrace.end(dispatchSpan); - } - metrics.addDispatchLatencyNanos(System.nanoTime() - start); - metrics.incDispatched(frames.size()); - return frames.size(); - } - - /** - * TTL expiry check (§13.3.4): an event with an {@code emttl} attribute (ms) and a {@code time} - * is expired when {@code now > time + emttl}. Events without either field never expire. Reads the - * attributes off the internal EventMeshFrame (emttl/time are preserved in its KV section). - */ - private boolean isExpired(org.apache.eventmesh.common.wire.EventMeshFrame event) { - String ttl = event.attributes().get("emttl"); - String time = event.attributes().get("time"); - if (ttl == null || time == null) { - return false; - } - try { - long ttlMs = Long.parseLong(ttl); - long eventTime = java.time.OffsetDateTime.parse(time).toInstant().toEpochMilli(); - return System.currentTimeMillis() > eventTime + ttlMs; - } catch (NumberFormatException | java.time.format.DateTimeParseException e) { - return false; - } - } - - /** - * Local delivery for one subscriber — handed to the reliability layer + the subscriber's push - * channel. Exposed so a {@link org.apache.eventmesh.runtime.cluster.ClusterCoordinator} can route - * same-instance targets here while forwarding remote ones. - */ - public boolean deliverLocal(String topic, String clientId, org.apache.eventmesh.common.wire.EventMeshFrame event) { - long offset = nextOffset(topic); - dispatcher.deliver(topic, -1, offset, event, clientId, channelFor(clientId)); - return true; - } - - /** - * Enable multi-instance coordination: when set, {@link #pullAndDispatch} routes each event - * through the cluster coordinator (local targets via {@link #deliverLocal}, remote via forward). - */ - public void withCluster(org.apache.eventmesh.runtime.cluster.ClusterCoordinator cluster) { - this.cluster = cluster; - } - - /** - * Topics this instance should pull and partition-assign: local active topics UNION cluster-wide - * topics (topics with a remote subscriber discovered via the Meta watch). Without the cluster - * half, an instance with no local subscriber for a topic never pulls it, so messages on its - * partitions can't be forwarded to the remote subscriber (multi-instance message loss). Returns - * local-only when not clustered. Both {@code UniRuntime.pullLoop} and {@code PartitionOwnership}'s - * topic source use this so the pull set and the assignment set stay consistent (otherwise an - * unassigned cluster topic would degrade to poll-all and duplicate). - */ - public java.util.Set activeTopicsClustered() { - java.util.Set topics = new java.util.HashSet<>(getSubscriptionManager().activeTopics()); - if (cluster != null) { - topics.addAll(cluster.subscriptionTopics()); - } - return topics; - } - - // ===================== Lite Topic (RIP-83, 5.x-only) ===================== - // Lite topic ops are exposed only when the storage plugin implements LiteTopicCapable; otherwise - // they fail fast (UnsupportedOperationException). The HTTP layer (/events/lite/*) delegates here. - - /** - * Publish one CloudEvent to a lite topic (parentTopic, liteTopic). The storage plugin routes it - * into the lite topic's LMQ. Requires a {@link org.apache.eventmesh.api.storage.LiteTopicCapable} - * storage (the 5.x plugin); 4.x/kafka/standalone storages throw. - */ - public CompletableFuture publishLite(String parentTopic, String liteTopic, CloudEvent event) { - CompletableFuture future = new CompletableFuture<>(); - if (!(storage instanceof org.apache.eventmesh.api.storage.LiteTopicCapable)) { - future.completeExceptionally(new UnsupportedOperationException("storage does not support lite topic")); - return future; - } - try { - // Boundary: CloudEvent → EventMeshFrame (internal wire unit). - org.apache.eventmesh.common.wire.EventMeshFrame frame = - org.apache.eventmesh.common.wire.EventMeshFrame.fromCloudEvent(event); - ((org.apache.eventmesh.api.storage.LiteTopicCapable) storage).sendLite(parentTopic, liteTopic, frame, - new SendCallback() { - @Override - public void onSuccess(SendResult sendResult) { - future.complete(null); - } - - @Override - public void onException(org.apache.eventmesh.api.exception.OnExceptionContext context) { - future.completeExceptionally(context.getException()); - } - }); - } catch (Exception e) { - future.completeExceptionally(e); - } - return future; - } - - /** - * Pull a batch of CloudEvents from a lite topic (direct pull from the LMQ; no deliveryId / no - * EventMesh reliability layer — the lite consumer self-manages offset in the storage plugin). - * Frames pulled from the LMQ are decoded back to CloudEvents at this boundary. Empty list if lite - * is not supported. - */ - public List pollLite(String parentTopic, String liteTopic, int maxEvents, long timeoutMs) { - if (!(storage instanceof org.apache.eventmesh.api.storage.LiteTopicCapable)) { - return java.util.Collections.emptyList(); - } - java.util.List out = new java.util.ArrayList<>(); - for (org.apache.eventmesh.common.wire.EventMeshFrame f - : ((org.apache.eventmesh.api.storage.LiteTopicCapable) storage).pullLite(parentTopic, liteTopic, maxEvents, timeoutMs)) { - out.add(f.toCloudEvent()); - } - return out; - } - - /** - * Publish a pre-encoded EventMeshFrame byte payload to a lite topic (the internal streaming wire - * path — SessionRouter publishes frame bytes). The payload IS an encoded EventMeshFrame; decode - * to the Frame object the SPI now expects. - */ - public CompletableFuture publishLiteBytes(String parentTopic, String liteTopic, byte[] payload) { - if (loadMeter != null) { - loadMeter.recordInflow(payload.length); - } - CompletableFuture future = new CompletableFuture<>(); - if (!(storage instanceof org.apache.eventmesh.api.storage.LiteTopicCapable)) { - future.completeExceptionally(new UnsupportedOperationException("storage does not support lite topic")); - return future; - } - try { - org.apache.eventmesh.common.wire.EventMeshFrame frame = org.apache.eventmesh.common.wire.EventMeshFrame.decode(payload); - ((org.apache.eventmesh.api.storage.LiteTopicCapable) storage).sendLite(parentTopic, liteTopic, frame, - new SendCallback() { - @Override - public void onSuccess(SendResult sendResult) { - future.complete(null); - } - - @Override - public void onException(org.apache.eventmesh.api.exception.OnExceptionContext context) { - future.completeExceptionally(context.getException()); - } - }); - } catch (Exception e) { - future.completeExceptionally(e); - } - return future; - } - - /** - * Pull a batch of pre-encoded EventMeshFrame byte payloads from a lite topic (the byte counterpart - * of {@link #pollLite}, for the internal streaming wire). Each entry is an encoded frame. - * Empty list if lite is not supported. - */ - public List pollLiteBytes(String parentTopic, String liteTopic, int maxEvents, long timeoutMs) { - if (!(storage instanceof org.apache.eventmesh.api.storage.LiteTopicCapable)) { - return java.util.Collections.emptyList(); - } - java.util.List out = new java.util.ArrayList<>(); - for (org.apache.eventmesh.common.wire.EventMeshFrame f - : ((org.apache.eventmesh.api.storage.LiteTopicCapable) storage).pullLite(parentTopic, liteTopic, maxEvents, timeoutMs)) { - out.add(f.encode()); - } - return out; - } - - /** - * Ensure {@code parentTopic} is lite-capable and declare {@code liteTopic} under it. Throws if the - * storage does not support lite. - */ - public void createLiteTopic(String parentTopic, String liteTopic) throws Exception { - if (!(storage instanceof org.apache.eventmesh.api.storage.LiteTopicCapable)) { - throw new UnsupportedOperationException("storage does not support lite topic"); - } - ((org.apache.eventmesh.api.storage.LiteTopicCapable) storage).createLiteTopic(parentTopic, liteTopic); - } - - /** - * Ensure {@code parentTopic} is lite-capable with the given {@code queueCount} and declare - * {@code liteTopic} under it. Use {@code queueCount=1} for strict in-order delivery (e.g. a - * streaming-call response channel). Throws if the storage does not support lite. - */ - public void createLiteTopic(String parentTopic, String liteTopic, int queueCount) throws Exception { - if (!(storage instanceof org.apache.eventmesh.api.storage.LiteTopicCapable)) { - throw new UnsupportedOperationException("storage does not support lite topic"); - } - ((org.apache.eventmesh.api.storage.LiteTopicCapable) storage).createLiteTopic(parentTopic, liteTopic, queueCount); - } - - /** - * @return true iff the storage plugin implements {@link org.apache.eventmesh.api.storage.LiteTopicCapable}. - */ - public boolean isLiteCapable() { - return storage instanceof org.apache.eventmesh.api.storage.LiteTopicCapable; - } - - /** - * Enable multi-instance partition ownership (§13.2.3): when set, {@link #pullAndDispatch} polls - * only this instance's owned partitions instead of the whole topic. - */ - public void withPartitionOwnership(org.apache.eventmesh.runtime.cluster.PartitionOwnership ownership) { - this.partitionOwnership = ownership; - } - - /** Wire the load meter; ingress/egress points call its record* methods. */ - public void withLoadMeter(LoadMeter loadMeter) { - this.loadMeter = loadMeter; - } - - /** - * @return the wired load meter, or null if not configured (single-instance / tests). - */ - public LoadMeter loadMeter() { - return loadMeter; - } - - /** - * Subscriber long-polls its buffered deliveries. - */ - public List poll(String clientId, int maxEvents, long timeoutMs) { - return pushService.poll(clientId, maxEvents, timeoutMs); - } - - /** - * Subscriber acknowledges a delivery — the offset advances only on ACK (at-least-once). - */ - public boolean ack(String deliveryId) { - return pushService.ack(deliveryId); - } - - /** - * Drive retry / DLQ. Call periodically from a scheduler. - */ - public int dispatcherTick() { - return dispatcher.tick(); - } - - /** - * Operational metrics counters (publish/dispatch/ack/retry/DLQ). The dispatcher and ingress - * share one instance. - */ - public UniMetrics getMetrics() { - return metrics; - } - - // Accessors for the admin facade (Phase 7.5). The service is the single owner of these - // collaborators; exposing them avoids reconstructing them out of band. - public SubscriptionManager getSubscriptionManager() { - return subscriptionManager; - } - - public PushService getPushService() { - return pushService; - } - - public ReliableDispatcher getDispatcher() { - return dispatcher; - } - - /** - * @return the multi-instance partition ownership (null when clustering is disabled). - */ - public org.apache.eventmesh.runtime.cluster.PartitionOwnership getPartitionOwnership() { - return partitionOwnership; - } - - /** - * Stale-poll cleanup (§13.6.5): evict clients that haven't polled within {@code thresholdMs} — - * drops their subscriptions and push buffer so zombie subscriptions don't leak. - * - * @return number of subscriptions removed - */ - public int cleanupStaleClients(long thresholdMs) { - int removed = 0; - for (String cid : pushService.getStaleClientIds(thresholdMs)) { - removed += subscriptionManager.unsubscribeByClient(cid); - pushService.removeClient(cid); - log.info("evicted stale client {} (no poll within {}ms)", cid, thresholdMs); - } - return removed; - } - - /** - * Register OTel observable gauges backed by live runtime state (§13.5.1 gauges with *). - * Call once at boot. Gauges read on each OTel collection cycle. - */ - public void registerRuntimeGauges() { - metrics.registerGauge("eventmesh_pending_queue_size", "total buffered events across all clients", - () -> { - long sum = 0; - for (String cid : pushService.clientIds()) { - sum += pushService.pending(cid); - } - return sum; - }); - metrics.registerGauge("eventmesh_slow_consumer_count", "clients in SLOW or STALLED state", - pushService::slowConsumerCount); - metrics.registerGauge("eventmesh_active_topics", "topics with active subscribers", - () -> subscriptionManager.activeTopics().size()); - metrics.registerGauge("eventmesh_active_subscribers", "active subscriptions across all topics", - () -> { - int sum = 0; - for (String t : subscriptionManager.activeTopics()) { - sum += subscriptionManager.activeSubscriptions(t).size(); - } - return sum; - }); - - // Labelled gauges (§13.5.1) — emit one reading per topic / partition. - metrics.registerLabelledGauge("eventmesh_poll_idle_ratio", - "fraction of poll cycles returning no events (per-mille, per topic)", - () -> { - java.util.List out = new java.util.ArrayList<>(); - for (String t : pollCount.keySet()) { - long total = pollCount.getOrDefault(t, 0L); - long empty = pollEmptyCount.getOrDefault(t, 0L); - long perMille = total == 0 ? 0 : Math.round((double) empty / total * 1000); - out.add(new UniMetrics.LabelledLong( - io.opentelemetry.api.common.Attributes.of(io.opentelemetry.api.common.AttributeKey.stringKey("topic"), t), - perMille)); - } - return out; - }); - - metrics.registerLabelledGauge("eventmesh_partition_owner", - "1 for each partition this instance owns (per topic/partition)", - () -> { - java.util.List out = new java.util.ArrayList<>(); - if (partitionOwnership != null) { - for (String t : subscriptionManager.activeTopics()) { - java.util.List owned = partitionOwnership.ownedPartitions(t); - if (owned == null) { - continue; - } - for (int p : owned) { - out.add(new UniMetrics.LabelledLong( - io.opentelemetry.api.common.Attributes.of( - io.opentelemetry.api.common.AttributeKey.stringKey("topic"), t, - io.opentelemetry.api.common.AttributeKey.longKey("partition"), (long) p, - io.opentelemetry.api.common.AttributeKey.stringKey("instance"), "self"), - 1L)); - } - } - } - return out; - }); - - metrics.registerLabelledGauge("eventmesh_offset_lag", - "MQ end offset - max ACK offset (per topic/partition) — total consumer lag", - () -> { - java.util.List out = new java.util.ArrayList<>(); - if (partitionOwnership == null) { - return out; - } - for (String t : subscriptionManager.activeTopics()) { - java.util.List owned = partitionOwnership.ownedPartitions(t); - if (owned == null) { - continue; - } - // Max ACK offset per partition across all clients (key = clientId#partition). - // Exclude the reserved __mqcursor__ key (MQ physical offset — different - // magnitude from the logical sequence numbers, mixing them corrupts the gauge). - java.util.Map maxAckByPart = new java.util.HashMap<>(); - String reservedPrefix = - org.apache.eventmesh.runtime.delivery.ReliableDispatcher.MQ_CURSOR_CLIENT + "#"; - for (java.util.Map.Entry e : offsetStore.readAllOffsets(t).entrySet()) { - if (e.getKey().startsWith(reservedPrefix)) { - continue; - } - int sep = e.getKey().lastIndexOf('#'); - if (sep > 0) { - try { - int p = Integer.parseInt(e.getKey().substring(sep + 1)); - maxAckByPart.merge(p, e.getValue(), Math::max); - } catch (NumberFormatException expected) { - // offset key suffix is not a numeric partition; skip - } - } - } - for (int p : owned) { - long end = storage.endOffset(t, p); - long ack = maxAckByPart.getOrDefault(p, -1L); - if (end >= 0 && ack >= 0) { - out.add(new UniMetrics.LabelledLong( - io.opentelemetry.api.common.Attributes.of( - io.opentelemetry.api.common.AttributeKey.stringKey("topic"), t, - io.opentelemetry.api.common.AttributeKey.longKey("partition"), (long) p), - Math.max(0, end - ack))); - } - } - } - return out; - }); - - metrics.registerLabelledGauge("eventmesh_push_ack_lag", - "max push offset - max ACK offset (per topic/partition) — in-flight deliveries", - () -> { - java.util.List out = new java.util.ArrayList<>(); - for (String t : subscriptionManager.activeTopics()) { - // Frame architecture: the MQ physical cursor per partition is recorded by the - // dispatcher on client ACK under the reserved key MQ_CURSOR_CLIENT (frame - // attribute emmqoffset, stamped by Kafka / RocketMQ-4.x at poll). Lag = - // storage endOffset (physical watermark) − recorded cursor. Per-subscriber - // entries hold logical sequence numbers and are excluded. - java.util.Map cursorByPart = new java.util.HashMap<>(); - String cursorPrefix = - org.apache.eventmesh.runtime.delivery.ReliableDispatcher.MQ_CURSOR_CLIENT + "#"; - for (java.util.Map.Entry e : offsetStore.readAllOffsets(t).entrySet()) { - if (!e.getKey().startsWith(cursorPrefix)) { - continue; - } - try { - int p = Integer.parseInt(e.getKey().substring(cursorPrefix.length())); - cursorByPart.merge(p, e.getValue(), Math::max); - } catch (NumberFormatException expected) { - } - } - for (java.util.Map.Entry cursorEntry : cursorByPart.entrySet()) { - int p = cursorEntry.getKey(); - long end = storage.endOffset(t, p); - long cursor = cursorEntry.getValue(); - if (end >= 0 && cursor >= 0) { - out.add(new UniMetrics.LabelledLong( - io.opentelemetry.api.common.Attributes.of( - io.opentelemetry.api.common.AttributeKey.stringKey("topic"), t, - io.opentelemetry.api.common.AttributeKey.longKey("partition"), (long) p), - Math.max(0, end - cursor))); - } - } - } - return out; - }); - } - - public OffsetStore getOffsetStore() { - return offsetStore; - } - - public MeshStoragePlugin getStorage() { - return storage; - } - - /** - * Synchronous request-reply (§17). Publishes {@code event}, blocks for the matching reply - * keyed by the {@code x-em-correlation-id} extension, and returns it. On timeout the future is - * failed and a late reply is discarded. Request-reply is independent of the at-least-once - * pub/sub path: it neither retries nor dead-letters. - * - * @throws Exception if the request times out, publishing fails, or the reply errors - */ - public CloudEvent request(String topic, CloudEvent event, long timeoutMs) throws Exception { - String correlationId = readCorrelationId(event); - CloudEvent toPublish = event; - if (correlationId == null) { - correlationId = "req-" + requestSeq.incrementAndGet(); - toPublish = CloudEventBuilder.from(event).withExtension(EXT_CORRELATION_ID, correlationId).build(); - } - CompletableFuture future = new CompletableFuture<>(); - pendingRequests.put(correlationId, future); - try { - publish(topic, toPublish).get(); - } catch (Exception e) { - pendingRequests.remove(correlationId); - throw e; - } - try { - return future.get(timeoutMs, TimeUnit.MILLISECONDS); - } catch (TimeoutException te) { - pendingRequests.remove(correlationId); - throw new TimeoutException("request-reply timed out: " + correlationId); - } catch (ExecutionException ee) { - throw ee; - } finally { - metrics.incRequestReply(); - } - } - - /** - * Deliver a reply to a pending request. Returns false if the request was unknown, already - * replied, or had timed out (late reply discarded). - */ - public boolean reply(String correlationId, CloudEvent replyEvent) { - CompletableFuture future = pendingRequests.remove(correlationId); - if (future == null) { - log.debug("late/unknown reply for correlationId={} discarded", correlationId); - return false; - } - return future.complete(replyEvent); - } - - private static String readCorrelationId(CloudEvent event) { - Object value = event.getExtension(EXT_CORRELATION_ID); - return value == null ? null : value.toString(); - } - - private PushChannel channelFor(String clientId) { - return channels.computeIfAbsent(clientId, id -> new LongPollingChannel(pushService, id)); - } - - /** Per-topic monotonic delivery sequence (EventMesh's logical offset for the dispatcher). */ - private long nextOffset(String topic) { - return topicOffsetSeq.computeIfAbsent(topic, k -> new AtomicLong()).incrementAndGet(); - } - - private DeadLetterSink deadLetterSink() { - return (originalTopic, event, reason, attempts) -> { - String dlqTopic = originalTopic + "_DLQ"; - // Issue #5292: the dispatcher retires the delivery only once this future reports the - // DLQ write as durably recorded by the storage plugin. - java.util.concurrent.CompletableFuture future = new java.util.concurrent.CompletableFuture<>(); - try { - // event is already an EventMeshFrame (internal); store it directly to the DLQ topic. - storage.send(dlqTopic, event, new SendCallback() { - @Override - public void onSuccess(SendResult sendResult) { - log.info("event {} dead-lettered to {} after {} attempts: {}", - event.attributes().get("id"), dlqTopic, attempts, reason); - future.complete(Boolean.TRUE); - } - - @Override - public void onException(org.apache.eventmesh.api.exception.OnExceptionContext context) { - log.error("failed to write DLQ event {} to {}", event.attributes().get("id"), dlqTopic, context.getException()); - future.complete(Boolean.FALSE); - } - }); - } catch (Exception e) { - log.error("failed to send DLQ event {} to {}", event.attributes().get("id"), dlqTopic, e); - future.complete(Boolean.FALSE); - } - return future; - }; - } -} +/* + * 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.ingress; + +import org.apache.eventmesh.api.SendCallback; +import org.apache.eventmesh.api.SendResult; +import org.apache.eventmesh.api.storage.MeshStoragePlugin; +import org.apache.eventmesh.common.wire.EventMeshFrame; +import org.apache.eventmesh.runtime.delivery.DeadLetterSink; +import org.apache.eventmesh.runtime.delivery.PushChannel; +import org.apache.eventmesh.runtime.delivery.ReliableDispatcher; +import org.apache.eventmesh.runtime.metrics.UniMetrics; +import org.apache.eventmesh.runtime.metrics.UniTrace; +import org.apache.eventmesh.runtime.offset.OffsetStore; +import org.apache.eventmesh.runtime.push.BufferedEvent; +import org.apache.eventmesh.runtime.push.LongPollingChannel; +import org.apache.eventmesh.runtime.push.PushService; +import org.apache.eventmesh.runtime.ratelimit.RateLimitedException; +import org.apache.eventmesh.runtime.ratelimit.TokenBucketRateLimiter; +import org.apache.eventmesh.runtime.subscription.CloudEventFilter; +import org.apache.eventmesh.runtime.subscription.DistributionMode; +import org.apache.eventmesh.runtime.subscription.Subscription; +import org.apache.eventmesh.runtime.subscription.SubscriptionManager; + +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicLong; + +import io.cloudevents.CloudEvent; +import io.cloudevents.core.builder.CloudEventBuilder; + +import lombok.extern.slf4j.Slf4j; + +/** + * Facade that wires the uni-architecture layers into the end-to-end CloudEvents-over-MQ + * flow (§6): + * 
+ *   publish  ─▶ MeshStoragePlugin.send
+ *   pullLoop ─▶ MeshStoragePlugin.poll ─▶ SubscriptionManager.targetsFor ─▶ ReliableDispatcher.deliver ─▶ PushService
+ *   poll     ◀─ client long-polls PushService
+ *   ack      ─▶ PushService.ack ─▶ ReliableDispatcher.ack ─▶ OffsetStore (offset advances only on ACK)
+ *
+ * + *

This is the orchestration core; the actual HTTP endpoint wiring (§6 UniIngressHandler) + * delegates here. Phase-1 thin storage adapter means {@code partition} is {@code -1} and the offset + * is a per-topic monotonic logical counter until the native storage reimplementation supplies real + * partition/offset (Phase 1 step 2).

+ */ +@Slf4j +public class UniIngressService { + + private final MeshStoragePlugin storage; + private final OffsetStore offsetStore; + private final SubscriptionManager subscriptionManager; + private final ReliableDispatcher dispatcher; + private final PushService pushService; + private volatile org.apache.eventmesh.runtime.cluster.ClusterCoordinator cluster; + private volatile org.apache.eventmesh.runtime.cluster.PartitionOwnership partitionOwnership; + /** Self-collected load metrics for session-distribution load balancing (§3). Null until wired. */ + private volatile LoadMeter loadMeter; + + /** Per-topic poll stats for the {@code poll_idle_ratio} gauge (§13.5.1). */ + private final ConcurrentHashMap pollCount = new ConcurrentHashMap<>(); + private final ConcurrentHashMap pollEmptyCount = new ConcurrentHashMap<>(); + + /** Connector offset store (remote side) — String key → String offset value (§8.9). */ + private final java.util.concurrent.ConcurrentHashMap connectorOffsets = new java.util.concurrent.ConcurrentHashMap<>(); + private final UniMetrics metrics; + + private final ConcurrentHashMap channels = new ConcurrentHashMap<>(); + private final ConcurrentHashMap topicOffsetSeq = new ConcurrentHashMap<>(); + private final ConcurrentHashMap> pendingRequests = new ConcurrentHashMap<>(); + private final AtomicLong requestSeq = new AtomicLong(); + private final ConcurrentHashMap topicLimiters = new ConcurrentHashMap<>(); + + /** + * CloudEvents extension carrying the request's correlation id (§17). + * + *

Named without hyphens because the CloudEvents spec restricts extension attribute names to + * lower-case ASCII letters and digits — the redesign doc's {@code x-em-correlation-id} spelling + * is rejected by the SDK's name validation.

+ */ + public static final String EXT_CORRELATION_ID = "emcorrelationid"; + + public UniIngressService(MeshStoragePlugin storage, OffsetStore offsetStore) { + this(storage, offsetStore, new SubscriptionManager(), new PushService(), + ReliableDispatcher.DEFAULT_ACK_TIMEOUT_MS, ReliableDispatcher.DEFAULT_MAX_ATTEMPTS, + System::currentTimeMillis); + } + + /** + * Test-friendly constructor with an injectable clock and retry parameters. + */ + public UniIngressService(MeshStoragePlugin storage, OffsetStore offsetStore, + SubscriptionManager subscriptionManager, PushService pushService, + long ackTimeoutMs, int maxAttempts, java.util.function.LongSupplier clock) { + this.storage = storage; + this.offsetStore = offsetStore; + this.subscriptionManager = subscriptionManager; + this.pushService = pushService; + this.metrics = new UniMetrics(); + this.dispatcher = new ReliableDispatcher(ackTimeoutMs, maxAttempts, clock, offsetStore, deadLetterSink(), + metrics, ReliableDispatcher.DEFAULT_JITTER_RATIO); + // Sub-PR B: re-ACK any in-flight deliveries from a previous JVM so they do not become + // orphans after a crash. Safe on first start (store is empty) and idempotent. + this.dispatcher.recover(); + } + + // ---- connector offset (remote side, §8.9) ---- + + public String getConnectorOffset(String connectorId) { + return connectorOffsets.get(connectorId); + } + + public void putConnectorOffset(String connectorId, String offset) { + connectorOffsets.put(connectorId, offset); + } + + /** + * Publish a CloudEvent to {@code topic} (persisted to MQ). Completes when the storage plugin + * acknowledges the write. + */ + public CompletableFuture publish(String topic, CloudEvent event) { + if (loadMeter != null && event.getData() != null) { + loadMeter.recordInflow(event.getData().toBytes().length); + } + CompletableFuture future = new CompletableFuture<>(); + TokenBucketRateLimiter limiter = topicLimiters.get(topic); + if (limiter != null && !limiter.tryAcquire()) { + metrics.incRateLimited(); + future.completeExceptionally(new RateLimitedException(topic)); + return future; + } + try { + // Ingress boundary: CloudEvent (the public/external format) → EventMeshFrame (the internal + // wire unit). From here on the event is a Frame internally; egress converts back to the + // client's protocol (CloudEvents / MeshMessage) at the delivery boundary. + org.apache.eventmesh.common.wire.EventMeshFrame frame = + org.apache.eventmesh.common.wire.EventMeshFrame.fromCloudEvent(event); + storage.send(topic, frame, new SendCallback() { + @Override + public void onSuccess(SendResult sendResult) { + metrics.incPublish(); + UniTrace.end(UniTrace.startPublish(topic, event)); + future.complete(null); + } + + @Override + public void onException(org.apache.eventmesh.api.exception.OnExceptionContext context) { + metrics.incPublishFailed(); + future.completeExceptionally(context.getException()); + } + }); + } catch (Exception e) { + metrics.incPublishFailed(); + future.completeExceptionally(e); + } + return future; + } + + /** + * Publish an already-internal {@link org.apache.eventmesh.common.wire.EventMeshFrame} (no + * CloudEvent→Frame boundary conversion — the caller built the frame directly, e.g. the legacy + * MeshMessage TCP path via {@code MeshMessageFrameCodec}). Completes when storage acks the write. + */ + public CompletableFuture publish(String topic, org.apache.eventmesh.common.wire.EventMeshFrame frame) { + if (loadMeter != null && frame.data() != null) { + loadMeter.recordInflow(frame.data().length); + } + CompletableFuture future = new CompletableFuture<>(); + TokenBucketRateLimiter limiter = topicLimiters.get(topic); + if (limiter != null && !limiter.tryAcquire()) { + metrics.incRateLimited(); + future.completeExceptionally(new RateLimitedException(topic)); + return future; + } + try { + storage.send(topic, frame, new SendCallback() { + @Override + public void onSuccess(SendResult sendResult) { + metrics.incPublish(); + future.complete(null); + } + + @Override + public void onException(org.apache.eventmesh.api.exception.OnExceptionContext context) { + metrics.incPublishFailed(); + future.completeExceptionally(context.getException()); + } + }); + } catch (Exception e) { + metrics.incPublishFailed(); + future.completeExceptionally(e); + } + return future; + } + + /** + * Publish a batch of CloudEvents to {@code topic} (§13.7.3). Completes when all are persisted. + * The storage plugin's own batching (e.g. Kafka producer accumulator) amortises the per-event + * RTT; this layer fans out the per-event futures and joins them. + */ + public CompletableFuture publishBatch(String topic, java.util.List events) { + if (events == null || events.isEmpty()) { + return CompletableFuture.completedFuture(null); + } + CompletableFuture[] futures = events.stream() + .map(e -> publish(topic, e)) + .toArray(CompletableFuture[]::new); + return CompletableFuture.allOf(futures); + } + + /** + * Primary ingress entry for a batch of internal EventMeshFrames (#5299). The HTTP path uses + * {@code FrameAdaptors.get("cloudevents").toFrame(body)} per event then calls this method; the + * CloudEvent-typed {@link #publishBatch(String, java.util.List)} is preserved for binary + * compatibility with the TCP bridge and any out-of-tree caller. + */ + public CompletableFuture publishBatchFrames(String topic, java.util.List frames) { + if (frames == null || frames.isEmpty()) { + return CompletableFuture.completedFuture(null); + } + CompletableFuture[] futures = frames.stream() + .map(f -> publish(topic, f)) + .toArray(CompletableFuture[]::new); + return CompletableFuture.allOf(futures); + } + + /** + * Register a custom push channel for a client (e.g. a WebHook URL for legacy HTTP webhook-push + * subscribers, or a {@code TcpPushChannel} for legacy TCP clients). Overrides the default + * long-polling channel for this {@code clientId} on subsequent dispatches. + */ + public void registerChannel(String clientId, PushChannel channel) { + pushService.register(clientId); + channels.put(clientId, channel); + } + + /** + * Configure a per-topic rate limit (§6.6). Subsequent publishes above {@code permitsPerSecond} + * (burst {@code capacity}) fail the publish future with {@link RateLimitedException}. + */ + public void setTopicRateLimit(String topic, long capacity, double permitsPerSecond) { + topicLimiters.put(topic, new TokenBucketRateLimiter(capacity, permitsPerSecond)); + } + + /** + * Register a subscription. The subscriber retrieves events via {@link #poll}. + * + * @return the subscription id + */ + public String subscribe(String topic, String clientId, DistributionMode mode, CloudEventFilter filter) { + pushService.register(clientId); + String subId = subscriptionManager.subscribe(topic, clientId, mode, filter, event -> { + // No-op: the reliability layer (ReliableDispatcher) owns delivery, not this fire-and- + // forget callback. Kept to satisfy the SubscriptionManager handler contract. + }); + // §13.2: when clustered, also register cluster-wide so other instances can route events + // for this subscriber here (via ClusterCoordinator → HttpForwarder /internal/forward). + // Without this the HTTP /events/subscribe path is local-only and a publish on a peer + // never reaches this subscriber. + if (cluster != null) { + cluster.subscribe(topic, clientId, mode, null); + } + return subId; + } + + /** + * Remove a subscription. + */ + public boolean unsubscribe(String subscriptionId) { + // Resolve topic + clientId before removing so a clustered subscription can be deregistered + // cluster-wide (the coordinator's unsubscribe is keyed by topic+clientId, not subId). + org.apache.eventmesh.runtime.subscription.Subscription sub = subscriptionManager.getSubscription(subscriptionId); + boolean removed = subscriptionManager.unsubscribe(subscriptionId); + if (removed && cluster != null && sub != null) { + cluster.unsubscribe(sub.getTopic(), sub.getClientId()); + } + return removed; + } + + /** Remove one client's subscription to one topic (HTTP /events/unsubscribe with {clientId, topic}). */ + public boolean unsubscribe(String topic, String clientId) { + boolean removed = subscriptionManager.unsubscribe(topic, clientId); + if (removed && cluster != null) { + cluster.unsubscribe(topic, clientId); + } + return removed; + } + + /** Remove ALL subscriptions for a client (HTTP /events/unsubscribe with {clientId} only). + * Propagates cluster-wide + frees the PushService buffer. */ + public int unsubscribeByClient(String clientId) { + java.util.Set topics = subscriptionManager.topicsForClient(clientId); + int removed = subscriptionManager.unsubscribeByClient(clientId); + if (cluster != null) { + for (String topic : topics) { + cluster.unsubscribe(topic, clientId); + } + } + pushService.removeClient(clientId); + return removed; + } + + /** + * Pump: pull a batch from storage, route to each target subscriber via the reliability layer. + * + * @return number of events pulled + */ + public int pullAndDispatch(String topic, int maxEvents, long timeoutMs) { + pollCount.merge(topic, 1L, Long::sum); + // Multi-instance (§13.2.3): poll only the partitions this instance owns. Single-instance + // (no ownership) or unknown partition count → poll the whole topic (partition -1). + java.util.List owned = partitionOwnership == null ? null : partitionOwnership.ownedPartitions(topic); + int total; + if (owned == null) { + total = pullAndDispatchPartition(topic, -1, maxEvents, timeoutMs); + } else if (owned.isEmpty()) { + total = 0; // owns none -> do not poll (avoids duplicate with the real owners) + } else { + total = 0; + for (int p : owned) { + total += pullAndDispatchPartition(topic, p, maxEvents, timeoutMs); + } + } + if (total == 0) { + pollEmptyCount.merge(topic, 1L, Long::sum); + } + return total; + } + + private int pullAndDispatchPartition(String topic, int partition, int maxEvents, long timeoutMs) { + // Pull EventMeshFrames (internal wire); dispatch each as a Frame through the internal + // pipeline (filter/dispatcher/egress all carry Frame now; egress converts to the client's + // protocol at the wire boundary). + List frames = storage.poll(topic, partition, -1, maxEvents, timeoutMs); + if (frames == null || frames.isEmpty()) { + return 0; + } + long start = System.nanoTime(); + for (org.apache.eventmesh.common.wire.EventMeshFrame f : frames) { + if (isExpired(f)) { + // §13.3.4 TTL: drop expired events instead of dispatching. + log.debug("dropping expired event {} on topic {} (emttl elapsed)", f.attributes().get("id"), topic); + continue; + } + io.opentelemetry.api.trace.Span dispatchSpan = UniTrace.startDispatch(topic, f); + if (cluster != null) { + // Multi-instance: route via the cluster coordinator (local vs cross-instance forward). + cluster.dispatch(topic, f); + } else { + // develop's OffsetExtensions (CE-extension-carried MQ offset) is superseded by the + // Frame architecture: the POP check key rides in frame attributes (empopck) and the + // deferred broker ACK fires on client ACK — same at-least-once goal, Frame-native. + long offset = nextOffset(topic); + // P2 fix: if the frame carries a POP check key (RocketMQ 5.x deferred ACK), build a + // callback that ACKs the broker on client ACK (restoring at-least-once). + String popCk = f.attributes().get("empopck"); + Runnable mqAck = (popCk != null) ? () -> storage.ackPulledMessage(topic, popCk) : null; + for (Subscription target : subscriptionManager.targetsFor(topic, f)) { + dispatcher.deliver(topic, partition, offset, f, target.getClientId(), + channelFor(target.getClientId()), mqAck); + } + } + UniTrace.end(dispatchSpan); + } + metrics.addDispatchLatencyNanos(System.nanoTime() - start); + metrics.incDispatched(frames.size()); + return frames.size(); + } + + /** + * TTL expiry check (§13.3.4): an event with an {@code emttl} attribute (ms) and a {@code time} + * is expired when {@code now > time + emttl}. Events without either field never expire. Reads the + * attributes off the internal EventMeshFrame (emttl/time are preserved in its KV section). + */ + private boolean isExpired(org.apache.eventmesh.common.wire.EventMeshFrame event) { + String ttl = event.attributes().get("emttl"); + String time = event.attributes().get("time"); + if (ttl == null || time == null) { + return false; + } + try { + long ttlMs = Long.parseLong(ttl); + long eventTime = java.time.OffsetDateTime.parse(time).toInstant().toEpochMilli(); + return System.currentTimeMillis() > eventTime + ttlMs; + } catch (NumberFormatException | java.time.format.DateTimeParseException e) { + return false; + } + } + + /** + * Local delivery for one subscriber — handed to the reliability layer + the subscriber's push + * channel. Exposed so a {@link org.apache.eventmesh.runtime.cluster.ClusterCoordinator} can route + * same-instance targets here while forwarding remote ones. + */ + public boolean deliverLocal(String topic, String clientId, org.apache.eventmesh.common.wire.EventMeshFrame event) { + long offset = nextOffset(topic); + dispatcher.deliver(topic, -1, offset, event, clientId, channelFor(clientId)); + return true; + } + + /** + * Enable multi-instance coordination: when set, {@link #pullAndDispatch} routes each event + * through the cluster coordinator (local targets via {@link #deliverLocal}, remote via forward). + */ + public void withCluster(org.apache.eventmesh.runtime.cluster.ClusterCoordinator cluster) { + this.cluster = cluster; + } + + /** + * Topics this instance should pull and partition-assign: local active topics UNION cluster-wide + * topics (topics with a remote subscriber discovered via the Meta watch). Without the cluster + * half, an instance with no local subscriber for a topic never pulls it, so messages on its + * partitions can't be forwarded to the remote subscriber (multi-instance message loss). Returns + * local-only when not clustered. Both {@code UniRuntime.pullLoop} and {@code PartitionOwnership}'s + * topic source use this so the pull set and the assignment set stay consistent (otherwise an + * unassigned cluster topic would degrade to poll-all and duplicate). + */ + public java.util.Set activeTopicsClustered() { + java.util.Set topics = new java.util.HashSet<>(getSubscriptionManager().activeTopics()); + if (cluster != null) { + topics.addAll(cluster.subscriptionTopics()); + } + return topics; + } + + // ===================== Lite Topic (RIP-83, 5.x-only) ===================== + // Lite topic ops are exposed only when the storage plugin implements LiteTopicCapable; otherwise + // they fail fast (UnsupportedOperationException). The HTTP layer (/events/lite/*) delegates here. + + /** + * Publish one CloudEvent to a lite topic (parentTopic, liteTopic). The storage plugin routes it + * into the lite topic's LMQ. Requires a {@link org.apache.eventmesh.api.storage.LiteTopicCapable} + * storage (the 5.x plugin); 4.x/kafka/standalone storages throw. + */ + public CompletableFuture publishLite(String parentTopic, String liteTopic, CloudEvent event) { + CompletableFuture future = new CompletableFuture<>(); + if (!(storage instanceof org.apache.eventmesh.api.storage.LiteTopicCapable)) { + future.completeExceptionally(new UnsupportedOperationException("storage does not support lite topic")); + return future; + } + try { + // Boundary: CloudEvent → EventMeshFrame (internal wire unit). + org.apache.eventmesh.common.wire.EventMeshFrame frame = + org.apache.eventmesh.common.wire.EventMeshFrame.fromCloudEvent(event); + ((org.apache.eventmesh.api.storage.LiteTopicCapable) storage).sendLite(parentTopic, liteTopic, frame, + new SendCallback() { + @Override + public void onSuccess(SendResult sendResult) { + future.complete(null); + } + + @Override + public void onException(org.apache.eventmesh.api.exception.OnExceptionContext context) { + future.completeExceptionally(context.getException()); + } + }); + } catch (Exception e) { + future.completeExceptionally(e); + } + return future; + } + + /** + * Primary lite ingress: publish a single internal EventMeshFrame to a lite topic (LMQ) (#5299). + * The HTTP path uses {@code FrameAdaptors.get("cloudevents").toFrame(body)} then calls this + * method; no further CloudEvent touches runtime state. + */ + public CompletableFuture publishLiteFrame(String parentTopic, String liteTopic, EventMeshFrame frame) { + CompletableFuture future = new CompletableFuture<>(); + if (!(storage instanceof org.apache.eventmesh.api.storage.LiteTopicCapable)) { + future.completeExceptionally(new UnsupportedOperationException("storage does not support lite topic")); + return future; + } + if (loadMeter != null && frame.data() != null) { + loadMeter.recordInflow(frame.data().length); + } + try { + ((org.apache.eventmesh.api.storage.LiteTopicCapable) storage).sendLite(parentTopic, liteTopic, frame, + new SendCallback() { + @Override + public void onSuccess(SendResult sendResult) { + future.complete(null); + } + + @Override + public void onException(org.apache.eventmesh.api.exception.OnExceptionContext context) { + future.completeExceptionally(context.getException()); + } + }); + } catch (Exception e) { + future.completeExceptionally(e); + } + return future; + } + + /** + * Pull a batch of CloudEvents from a lite topic (direct pull from the LMQ; no deliveryId / no + * EventMesh reliability layer — the lite consumer self-manages offset in the storage plugin). + * Frames pulled from the LMQ are decoded back to CloudEvents at this boundary. Empty list if lite + * is not supported. + */ + public List pollLite(String parentTopic, String liteTopic, int maxEvents, long timeoutMs) { + if (!(storage instanceof org.apache.eventmesh.api.storage.LiteTopicCapable)) { + return java.util.Collections.emptyList(); + } + java.util.List out = new java.util.ArrayList<>(); + for (org.apache.eventmesh.common.wire.EventMeshFrame f + : ((org.apache.eventmesh.api.storage.LiteTopicCapable) storage).pullLite(parentTopic, liteTopic, maxEvents, timeoutMs)) { + out.add(f.toCloudEvent()); + } + return out; + } + + /** + * Primary lite poll: drain a batch of internal EventMeshFrames from the LMQ (#5299). The + * egress boundary (CloudEventsFrameAdaptor) converts back to CloudEvents JSON for the HTTP + * response body. Returns an empty list if the storage does not support lite topics. + */ + public List pollLiteFrames(String parentTopic, String liteTopic, int maxEvents, long timeoutMs) { + if (!(storage instanceof org.apache.eventmesh.api.storage.LiteTopicCapable)) { + return java.util.Collections.emptyList(); + } + return ((org.apache.eventmesh.api.storage.LiteTopicCapable) storage) + .pullLite(parentTopic, liteTopic, maxEvents, timeoutMs); + } + + /** + * Publish a pre-encoded EventMeshFrame byte payload to a lite topic (the internal streaming wire + * path — SessionRouter publishes frame bytes). The payload IS an encoded EventMeshFrame; decode + * to the Frame object the SPI now expects. + */ + public CompletableFuture publishLiteBytes(String parentTopic, String liteTopic, byte[] payload) { + if (loadMeter != null) { + loadMeter.recordInflow(payload.length); + } + CompletableFuture future = new CompletableFuture<>(); + if (!(storage instanceof org.apache.eventmesh.api.storage.LiteTopicCapable)) { + future.completeExceptionally(new UnsupportedOperationException("storage does not support lite topic")); + return future; + } + try { + org.apache.eventmesh.common.wire.EventMeshFrame frame = org.apache.eventmesh.common.wire.EventMeshFrame.decode(payload); + ((org.apache.eventmesh.api.storage.LiteTopicCapable) storage).sendLite(parentTopic, liteTopic, frame, + new SendCallback() { + @Override + public void onSuccess(SendResult sendResult) { + future.complete(null); + } + + @Override + public void onException(org.apache.eventmesh.api.exception.OnExceptionContext context) { + future.completeExceptionally(context.getException()); + } + }); + } catch (Exception e) { + future.completeExceptionally(e); + } + return future; + } + + /** + * Pull a batch of pre-encoded EventMeshFrame byte payloads from a lite topic (the byte counterpart + * of {@link #pollLite}, for the internal streaming wire). Each entry is an encoded frame. + * Empty list if lite is not supported. + */ + public List pollLiteBytes(String parentTopic, String liteTopic, int maxEvents, long timeoutMs) { + if (!(storage instanceof org.apache.eventmesh.api.storage.LiteTopicCapable)) { + return java.util.Collections.emptyList(); + } + java.util.List out = new java.util.ArrayList<>(); + for (org.apache.eventmesh.common.wire.EventMeshFrame f + : ((org.apache.eventmesh.api.storage.LiteTopicCapable) storage).pullLite(parentTopic, liteTopic, maxEvents, timeoutMs)) { + out.add(f.encode()); + } + return out; + } + + /** + * Ensure {@code parentTopic} is lite-capable and declare {@code liteTopic} under it. Throws if the + * storage does not support lite. + */ + public void createLiteTopic(String parentTopic, String liteTopic) throws Exception { + if (!(storage instanceof org.apache.eventmesh.api.storage.LiteTopicCapable)) { + throw new UnsupportedOperationException("storage does not support lite topic"); + } + ((org.apache.eventmesh.api.storage.LiteTopicCapable) storage).createLiteTopic(parentTopic, liteTopic); + } + + /** + * Ensure {@code parentTopic} is lite-capable with the given {@code queueCount} and declare + * {@code liteTopic} under it. Use {@code queueCount=1} for strict in-order delivery (e.g. a + * streaming-call response channel). Throws if the storage does not support lite. + */ + public void createLiteTopic(String parentTopic, String liteTopic, int queueCount) throws Exception { + if (!(storage instanceof org.apache.eventmesh.api.storage.LiteTopicCapable)) { + throw new UnsupportedOperationException("storage does not support lite topic"); + } + ((org.apache.eventmesh.api.storage.LiteTopicCapable) storage).createLiteTopic(parentTopic, liteTopic, queueCount); + } + + /** + * @return true iff the storage plugin implements {@link org.apache.eventmesh.api.storage.LiteTopicCapable}. + */ + public boolean isLiteCapable() { + return storage instanceof org.apache.eventmesh.api.storage.LiteTopicCapable; + } + + /** + * Enable multi-instance partition ownership (§13.2.3): when set, {@link #pullAndDispatch} polls + * only this instance's owned partitions instead of the whole topic. + */ + public void withPartitionOwnership(org.apache.eventmesh.runtime.cluster.PartitionOwnership ownership) { + this.partitionOwnership = ownership; + } + + /** Wire the load meter; ingress/egress points call its record* methods. */ + public void withLoadMeter(LoadMeter loadMeter) { + this.loadMeter = loadMeter; + } + + /** + * @return the wired load meter, or null if not configured (single-instance / tests). + */ + public LoadMeter loadMeter() { + return loadMeter; + } + + /** + * Subscriber long-polls its buffered deliveries. + */ + public List poll(String clientId, int maxEvents, long timeoutMs) { + return pushService.poll(clientId, maxEvents, timeoutMs); + } + + /** + * Subscriber acknowledges a delivery — the offset advances only on ACK (at-least-once). + */ + public boolean ack(String deliveryId) { + return pushService.ack(deliveryId); + } + + /** + * Drive retry / DLQ. Call periodically from a scheduler. + */ + public int dispatcherTick() { + return dispatcher.tick(); + } + + /** + * Operational metrics counters (publish/dispatch/ack/retry/DLQ). The dispatcher and ingress + * share one instance. + */ + public UniMetrics getMetrics() { + return metrics; + } + + // Accessors for the admin facade (Phase 7.5). The service is the single owner of these + // collaborators; exposing them avoids reconstructing them out of band. + public SubscriptionManager getSubscriptionManager() { + return subscriptionManager; + } + + public PushService getPushService() { + return pushService; + } + + public ReliableDispatcher getDispatcher() { + return dispatcher; + } + + /** + * @return the multi-instance partition ownership (null when clustering is disabled). + */ + public org.apache.eventmesh.runtime.cluster.PartitionOwnership getPartitionOwnership() { + return partitionOwnership; + } + + /** + * Stale-poll cleanup (§13.6.5): evict clients that haven't polled within {@code thresholdMs} — + * drops their subscriptions and push buffer so zombie subscriptions don't leak. + * + * @return number of subscriptions removed + */ + public int cleanupStaleClients(long thresholdMs) { + int removed = 0; + for (String cid : pushService.getStaleClientIds(thresholdMs)) { + removed += subscriptionManager.unsubscribeByClient(cid); + pushService.removeClient(cid); + log.info("evicted stale client {} (no poll within {}ms)", cid, thresholdMs); + } + return removed; + } + + /** + * Register OTel observable gauges backed by live runtime state (§13.5.1 gauges with *). + * Call once at boot. Gauges read on each OTel collection cycle. + */ + public void registerRuntimeGauges() { + metrics.registerGauge("eventmesh_pending_queue_size", "total buffered events across all clients", + () -> { + long sum = 0; + for (String cid : pushService.clientIds()) { + sum += pushService.pending(cid); + } + return sum; + }); + metrics.registerGauge("eventmesh_slow_consumer_count", "clients in SLOW or STALLED state", + pushService::slowConsumerCount); + metrics.registerGauge("eventmesh_active_topics", "topics with active subscribers", + () -> subscriptionManager.activeTopics().size()); + metrics.registerGauge("eventmesh_active_subscribers", "active subscriptions across all topics", + () -> { + int sum = 0; + for (String t : subscriptionManager.activeTopics()) { + sum += subscriptionManager.activeSubscriptions(t).size(); + } + return sum; + }); + + // Labelled gauges (§13.5.1) — emit one reading per topic / partition. + metrics.registerLabelledGauge("eventmesh_poll_idle_ratio", + "fraction of poll cycles returning no events (per-mille, per topic)", + () -> { + java.util.List out = new java.util.ArrayList<>(); + for (String t : pollCount.keySet()) { + long total = pollCount.getOrDefault(t, 0L); + long empty = pollEmptyCount.getOrDefault(t, 0L); + long perMille = total == 0 ? 0 : Math.round((double) empty / total * 1000); + out.add(new UniMetrics.LabelledLong( + io.opentelemetry.api.common.Attributes.of(io.opentelemetry.api.common.AttributeKey.stringKey("topic"), t), + perMille)); + } + return out; + }); + + metrics.registerLabelledGauge("eventmesh_partition_owner", + "1 for each partition this instance owns (per topic/partition)", + () -> { + java.util.List out = new java.util.ArrayList<>(); + if (partitionOwnership != null) { + for (String t : subscriptionManager.activeTopics()) { + java.util.List owned = partitionOwnership.ownedPartitions(t); + if (owned == null) { + continue; + } + for (int p : owned) { + out.add(new UniMetrics.LabelledLong( + io.opentelemetry.api.common.Attributes.of( + io.opentelemetry.api.common.AttributeKey.stringKey("topic"), t, + io.opentelemetry.api.common.AttributeKey.longKey("partition"), (long) p, + io.opentelemetry.api.common.AttributeKey.stringKey("instance"), "self"), + 1L)); + } + } + } + return out; + }); + + metrics.registerLabelledGauge("eventmesh_offset_lag", + "MQ end offset - max ACK offset (per topic/partition) — total consumer lag", + () -> { + java.util.List out = new java.util.ArrayList<>(); + if (partitionOwnership == null) { + return out; + } + for (String t : subscriptionManager.activeTopics()) { + java.util.List owned = partitionOwnership.ownedPartitions(t); + if (owned == null) { + continue; + } + // Max ACK offset per partition across all clients (key = clientId#partition). + // Exclude the reserved __mqcursor__ key (MQ physical offset — different + // magnitude from the logical sequence numbers, mixing them corrupts the gauge). + java.util.Map maxAckByPart = new java.util.HashMap<>(); + String reservedPrefix = + org.apache.eventmesh.runtime.delivery.ReliableDispatcher.MQ_CURSOR_CLIENT + "#"; + for (java.util.Map.Entry e : offsetStore.readAllOffsets(t).entrySet()) { + if (e.getKey().startsWith(reservedPrefix)) { + continue; + } + int sep = e.getKey().lastIndexOf('#'); + if (sep > 0) { + try { + int p = Integer.parseInt(e.getKey().substring(sep + 1)); + maxAckByPart.merge(p, e.getValue(), Math::max); + } catch (NumberFormatException expected) { + // offset key suffix is not a numeric partition; skip + } + } + } + for (int p : owned) { + long end = storage.endOffset(t, p); + long ack = maxAckByPart.getOrDefault(p, -1L); + if (end >= 0 && ack >= 0) { + out.add(new UniMetrics.LabelledLong( + io.opentelemetry.api.common.Attributes.of( + io.opentelemetry.api.common.AttributeKey.stringKey("topic"), t, + io.opentelemetry.api.common.AttributeKey.longKey("partition"), (long) p), + Math.max(0, end - ack))); + } + } + } + return out; + }); + + metrics.registerLabelledGauge("eventmesh_push_ack_lag", + "max push offset - max ACK offset (per topic/partition) — in-flight deliveries", + () -> { + java.util.List out = new java.util.ArrayList<>(); + for (String t : subscriptionManager.activeTopics()) { + // Frame architecture: the MQ physical cursor per partition is recorded by the + // dispatcher on client ACK under the reserved key MQ_CURSOR_CLIENT (frame + // attribute emmqoffset, stamped by Kafka / RocketMQ-4.x at poll). Lag = + // storage endOffset (physical watermark) − recorded cursor. Per-subscriber + // entries hold logical sequence numbers and are excluded. + java.util.Map cursorByPart = new java.util.HashMap<>(); + String cursorPrefix = + org.apache.eventmesh.runtime.delivery.ReliableDispatcher.MQ_CURSOR_CLIENT + "#"; + for (java.util.Map.Entry e : offsetStore.readAllOffsets(t).entrySet()) { + if (!e.getKey().startsWith(cursorPrefix)) { + continue; + } + try { + int p = Integer.parseInt(e.getKey().substring(cursorPrefix.length())); + cursorByPart.merge(p, e.getValue(), Math::max); + } catch (NumberFormatException expected) { + } + } + for (java.util.Map.Entry cursorEntry : cursorByPart.entrySet()) { + int p = cursorEntry.getKey(); + long end = storage.endOffset(t, p); + long cursor = cursorEntry.getValue(); + if (end >= 0 && cursor >= 0) { + out.add(new UniMetrics.LabelledLong( + io.opentelemetry.api.common.Attributes.of( + io.opentelemetry.api.common.AttributeKey.stringKey("topic"), t, + io.opentelemetry.api.common.AttributeKey.longKey("partition"), (long) p), + Math.max(0, end - cursor))); + } + } + } + return out; + }); + } + + public OffsetStore getOffsetStore() { + return offsetStore; + } + + public MeshStoragePlugin getStorage() { + return storage; + } + + /** + * Synchronous request-reply (§17). Publishes {@code event}, blocks for the matching reply + * keyed by the {@code x-em-correlation-id} extension, and returns it. On timeout the future is + * failed and a late reply is discarded. Request-reply is independent of the at-least-once + * pub/sub path: it neither retries nor dead-letters. + * + * @throws Exception if the request times out, publishing fails, or the reply errors + */ + public CloudEvent request(String topic, CloudEvent event, long timeoutMs) throws Exception { + String correlationId = readCorrelationId(event); + CloudEvent toPublish = event; + if (correlationId == null) { + correlationId = "req-" + requestSeq.incrementAndGet(); + toPublish = CloudEventBuilder.from(event).withExtension(EXT_CORRELATION_ID, correlationId).build(); + } + CompletableFuture future = new CompletableFuture<>(); + pendingRequests.put(correlationId, future); + try { + publish(topic, toPublish).get(); + } catch (Exception e) { + pendingRequests.remove(correlationId); + throw e; + } + try { + return future.get(timeoutMs, TimeUnit.MILLISECONDS); + } catch (TimeoutException te) { + pendingRequests.remove(correlationId); + throw new TimeoutException("request-reply timed out: " + correlationId); + } catch (ExecutionException ee) { + throw ee; + } finally { + metrics.incRequestReply(); + } + } + + /** + * Deliver a reply to a pending request. Returns false if the request was unknown, already + * replied, or had timed out (late reply discarded). + */ + public boolean reply(String correlationId, CloudEvent replyEvent) { + CompletableFuture future = pendingRequests.remove(correlationId); + if (future == null) { + log.debug("late/unknown reply for correlationId={} discarded", correlationId); + return false; + } + return future.complete(replyEvent); + } + + /** + * Primary request-reply path (#5299, §17): publish a Frame, wait for the matching reply Frame. + * Implementation note: this method round-trips through CloudEvent (via {@code + * EventMeshFrame.fromCloudEvent / toCloudEvent}) so the existing {@code pendingRequests} map + * (typed as {@code CompletableFuture}) stays untouched. A future sub-PR will + * re-type {@code pendingRequests} to {@code CompletableFuture} and remove + * the round-trip. The HTTP path now uses this method (the previous {@code request(CloudEvent)} + * is preserved for binary compat with the TCP bridge). + */ + public EventMeshFrame requestFrame(String topic, EventMeshFrame frame, long timeoutMs) throws Exception { + CloudEvent reply = request(topic, frame.toCloudEvent(), timeoutMs); + return EventMeshFrame.fromCloudEvent(reply); + } + + /** + * Primary reply path: complete the pending request future with a Frame (#5299). Round-trips + * through CloudEvent for the same reason as {@link #requestFrame(String, EventMeshFrame, long)}. + */ + public boolean replyFrame(String correlationId, EventMeshFrame replyFrame) { + return reply(correlationId, replyFrame.toCloudEvent()); + } + + private static String readCorrelationId(CloudEvent event) { + Object value = event.getExtension(EXT_CORRELATION_ID); + return value == null ? null : value.toString(); + } + + private PushChannel channelFor(String clientId) { + return channels.computeIfAbsent(clientId, id -> new LongPollingChannel(pushService, id)); + } + + /** Per-topic monotonic delivery sequence (EventMesh's logical offset for the dispatcher). */ + private long nextOffset(String topic) { + return topicOffsetSeq.computeIfAbsent(topic, k -> new AtomicLong()).incrementAndGet(); + } + + private DeadLetterSink deadLetterSink() { + return (originalTopic, event, reason, attempts) -> { + String dlqTopic = originalTopic + "_DLQ"; + // Issue #5292: the dispatcher retires the delivery only once this future reports the + // DLQ write as durably recorded by the storage plugin. + java.util.concurrent.CompletableFuture future = new java.util.concurrent.CompletableFuture<>(); + try { + // event is already an EventMeshFrame (internal); store it directly to the DLQ topic. + storage.send(dlqTopic, event, new SendCallback() { + @Override + public void onSuccess(SendResult sendResult) { + log.info("event {} dead-lettered to {} after {} attempts: {}", + event.attributes().get("id"), dlqTopic, attempts, reason); + future.complete(Boolean.TRUE); + } + + @Override + public void onException(org.apache.eventmesh.api.exception.OnExceptionContext context) { + log.error("failed to write DLQ event {} to {}", event.attributes().get("id"), dlqTopic, context.getException()); + future.complete(Boolean.FALSE); + } + }); + } catch (Exception e) { + log.error("failed to send DLQ event {} to {}", event.attributes().get("id"), dlqTopic, e); + future.complete(Boolean.FALSE); + } + return future; + }; + } +} diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/AclFilter.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/AclFilter.java index aa7d8ba9e0..30a68887ff 100644 --- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/AclFilter.java +++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/AclFilter.java @@ -17,12 +17,12 @@ package org.apache.eventmesh.runtime.security; +import org.apache.eventmesh.common.wire.EventMeshFrame; + import java.util.ArrayList; import java.util.Collections; import java.util.List; -import io.cloudevents.CloudEvent; - /** * Topic-level authorization (§13.4.2). Holds an {@link AclRule} set sorted by priority (descending, * DENY wins ties) and matches each request against it: the first matching rule's effect applies, @@ -58,13 +58,23 @@ public synchronized void setRules(List newRules) { this.rules = sorted; } + /** + * Tenant / clientId come from {@code ctx} (set by the HTTP handler before the filter chain). + * We no longer read them from the event because that path was CloudEvent-specific; sub-PR + * B keeps the contract simple by reading the principal from the context and only using the + * frame to confirm an event-shaped payload arrived. + */ @Override - public FilterVerdict check(CloudEvent event, FilterContext ctx) { + public FilterVerdict check(EventMeshFrame frame, FilterContext ctx) { String principal = ctx.getTenant() != null ? ctx.getTenant() : ctx.getClientId(); String resource = ctx.getTopic(); if (principal == null || resource == null) { return FilterVerdict.deny(FilterVerdict.STATUS_FORBIDDEN, "no principal/resource for ACL"); } + if (frame != null && !frame.isEvent()) { + return FilterVerdict.deny(FilterVerdict.STATUS_FORBIDDEN, + "ACL applies to EVENT frames only (got msgType=" + frame.msgType() + ")"); + } // action not yet carried in FilterContext — pass null so rule action doesn't restrict (any matches). for (AclRule rule : rules) { if (rule.matches(principal, resource, null)) { diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/FilterChain.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/FilterChain.java index 270db39057..be9c56f1cd 100644 --- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/FilterChain.java +++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/FilterChain.java @@ -17,13 +17,13 @@ package org.apache.eventmesh.runtime.security; +import org.apache.eventmesh.common.wire.EventMeshFrame; + import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.List; -import io.cloudevents.CloudEvent; - /** * Ordered ingress security pipeline (§4.5): AuthFilter → AclFilter → SignatureVerifier, etc. * @@ -46,10 +46,22 @@ public FilterChain(IngressFilter... filters) { /** * Run every filter; return the first denying verdict, or {@link FilterVerdict#allow()} if all * pass. + * + * @deprecated since #5299 — call {@link #check(EventMeshFrame, FilterContext)} instead. Kept + * as a bridge for the TCP path until sub-PR C migrates. + */ + @Deprecated + public FilterVerdict check(io.cloudevents.CloudEvent event, FilterContext ctx) { + return check(EventMeshFrame.fromCloudEvent(event), ctx); + } + + /** + * Run every filter on the runtime's internal frame; return the first denying verdict, or + * {@link FilterVerdict#allow()} if all pass. This is the primary ingress path since #5299. */ - public FilterVerdict check(CloudEvent event, FilterContext ctx) { + public FilterVerdict check(EventMeshFrame frame, FilterContext ctx) { for (IngressFilter filter : filters) { - FilterVerdict verdict = filter.check(event, ctx); + FilterVerdict verdict = filter.check(frame, ctx); if (!verdict.isAllowed()) { return verdict; } diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/IngressFilter.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/IngressFilter.java index 573b6828e1..e4193f121b 100644 --- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/IngressFilter.java +++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/IngressFilter.java @@ -17,18 +17,37 @@ package org.apache.eventmesh.runtime.security; +import org.apache.eventmesh.common.wire.EventMeshFrame; + import io.cloudevents.CloudEvent; /** * One stage of the ingress security pipeline (§4.5). Implementations: authentication (who are * you — {@code TokenAuthFilter}), authorization (what may you do — {@code AclFilter}), signature * verification ({@code SignatureVerifierFilter}). TLS / mTLS is enforced at the transport, not here. + * + *

Filters operate on the runtime's internal wire format ({@link EventMeshFrame}) since + * #5299; the legacy {@code CloudEvent} overload is retained as a bridge for code paths that have + * not yet migrated (notably the TCP ingress in sub-PR C). Implementations should override the + * {@code EventMeshFrame} variant; the {@code CloudEvent} variant is implemented as a default + * that delegates via {@code frame.toCloudEvent()} so existing custom filters keep working.

*/ -@FunctionalInterface public interface IngressFilter { /** * Decide whether {@code event} from {@code ctx} may proceed. + * + * @deprecated since #5299 — override {@link #check(EventMeshFrame, FilterContext)} instead. + * Will be removed once all ingress paths (HTTP, TCP, A2A) emit {@link EventMeshFrame}. + */ + @Deprecated + default FilterVerdict check(CloudEvent event, FilterContext ctx) { + return check(EventMeshFrame.fromCloudEvent(event), ctx); + } + + /** + * Decide whether {@code frame} from {@code ctx} may proceed. Default implementation reads + * tenant / signature / token directly from {@code frame.attributes()}. */ - FilterVerdict check(CloudEvent event, FilterContext ctx); + FilterVerdict check(EventMeshFrame frame, FilterContext ctx); } diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/SignatureVerifierFilter.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/SignatureVerifierFilter.java index df73ee864f..34dca2d404 100644 --- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/SignatureVerifierFilter.java +++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/SignatureVerifierFilter.java @@ -17,14 +17,15 @@ package org.apache.eventmesh.runtime.security; +import org.apache.eventmesh.common.wire.EventMeshFrame; + import java.nio.charset.StandardCharsets; import java.util.Locale; +import java.util.Map; import javax.crypto.Mac; import javax.crypto.spec.SecretKeySpec; -import io.cloudevents.CloudEvent; - /** * Verifies an HMAC-SHA256 signature over a canonical projection of the CloudEvent (§13.4.4), so the * receiver can detect tampering and assert provenance. The signature travels in the @@ -43,13 +44,16 @@ public SignatureVerifierFilter(String secret) { } @Override - public FilterVerdict check(CloudEvent event, FilterContext ctx) { - Object provided = event.getExtension(EXT_SIGNATURE); - if (!(provided instanceof String)) { + public FilterVerdict check(EventMeshFrame frame, FilterContext ctx) { + // The signature travels in the frame attributes under the same key the legacy CloudEvent + // extension used, so a signed CE-JSON payload becomes a signed frame automatically after + // the cloudevents FrameAdaptor round-trip. + String provided = frame.attributes().get(EXT_SIGNATURE); + if (provided == null) { return FilterVerdict.deny(FilterVerdict.STATUS_UNAUTHENTICATED, "missing signature"); } - String expected = sign(canonical(event)); - if (constantTimeEquals(expected, (String) provided)) { + String expected = sign(canonical(frame)); + if (constantTimeEquals(expected, provided)) { return FilterVerdict.allow(); } return FilterVerdict.deny(FilterVerdict.STATUS_UNAUTHENTICATED, "signature mismatch"); @@ -57,7 +61,7 @@ public FilterVerdict check(CloudEvent event, FilterContext ctx) { /** * Compute the signature over {@code message} — also used by clients/tests to produce the value - * placed in the {@code emsignature} extension. + * placed in the {@code emsignature} extension/attribute. */ public String sign(String message) { try { @@ -70,11 +74,12 @@ public String sign(String message) { } } - static String canonical(CloudEvent event) { - Object id = event.getId(); - Object source = event.getSource(); - Object type = event.getType(); - return String.valueOf(id) + "|" + String.valueOf(source) + "|" + String.valueOf(type); + static String canonical(EventMeshFrame frame) { + // Same projection as before (#5299 sub-PR B): id|source|type, all read from frame + // attributes. For non-EVENT frames the canonical string still computes (we may want to + // tighten to isEvent() in a follow-up if streaming chunks ever need sigs). + Map a = frame.attributes(); + return a.getOrDefault("id", "") + "|" + a.getOrDefault("source", "") + "|" + a.getOrDefault("type", ""); } private static String toHex(byte[] bytes) { diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/TokenAuthFilter.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/TokenAuthFilter.java index 2f44f44bbb..62ae9b3426 100644 --- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/TokenAuthFilter.java +++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/TokenAuthFilter.java @@ -17,11 +17,11 @@ package org.apache.eventmesh.runtime.security; +import org.apache.eventmesh.common.wire.EventMeshFrame; + import java.util.Collections; import java.util.Set; -import io.cloudevents.CloudEvent; - /** * Authenticates the caller by a bearer token (§13.4.1). In production this delegates to the * existing security-plugin ({@code auth-token} / {@code auth-http-basic}); the uni skeleton @@ -36,7 +36,7 @@ public TokenAuthFilter(Set validTokens) { } @Override - public FilterVerdict check(CloudEvent event, FilterContext ctx) { + public FilterVerdict check(EventMeshFrame frame, FilterContext ctx) { String credential = ctx.getCredential(); if (credential != null && validTokens.contains(credential)) { return FilterVerdict.allow(); diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/transport/tcp/CloudEventToPackageBody.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/transport/tcp/CloudEventToPackageBody.java deleted file mode 100644 index f92a3682fd..0000000000 --- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/transport/tcp/CloudEventToPackageBody.java +++ /dev/null @@ -1,31 +0,0 @@ -/* - * 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.transport.tcp; - -import io.cloudevents.CloudEvent; - -/** - * Adapts a delivered CloudEvent into the legacy TCP {@code Package} body the old client expects - * (egress direction). Production uses {@code MeshMessageProtocolAdaptor.fromCloudEvent(...)}; - * tests inject a stub (e.g. event → a Map the test can read back). - */ -@FunctionalInterface -public interface CloudEventToPackageBody { - - Object toBody(CloudEvent event); -} diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/transport/tcp/MeshEventToPackageBody.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/transport/tcp/MeshEventToPackageBody.java deleted file mode 100644 index 3c13c620d0..0000000000 --- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/transport/tcp/MeshEventToPackageBody.java +++ /dev/null @@ -1,50 +0,0 @@ -/* - * 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.transport.tcp; - -import org.apache.eventmesh.common.protocol.tcp.EventMeshMessage; - -import java.nio.charset.StandardCharsets; - -import io.cloudevents.CloudEvent; - -/** - * Egress encoder for the TCP push frame: builds the legacy {@code EventMeshMessage} body the client - * expects (topic ← CloudEvent subject, content ← CloudEvent data). The {@link NettyTcpPushChannel} - * wraps this body in an {@code ASYNC_MESSAGE_TO_CLIENT} Package; the netty {@code Codec} serializes - * it onto the wire. - * - *

This builds the body directly rather than routing through {@code MeshMessageProtocolAdaptor}. - * fromCloudEvent, because that adaptor's protocol-desc switching requires CloudEvents to carry - * legacy protocol metadata extensions that push frames don't naturally have; the wire payload a - * legacy TCP subscriber receives is the {@code EventMeshMessage} anyway.

- */ -public class MeshEventToPackageBody implements CloudEventToPackageBody { - - @Override - public Object toBody(CloudEvent event) { - EventMeshMessage message = new EventMeshMessage(); - if (event.getSubject() != null) { - message.setTopic(event.getSubject()); - } - if (event.getData() != null) { - message.setBody(new String(event.getData().toBytes(), StandardCharsets.UTF_8)); - } - return message; - } -} diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/transport/tcp/TcpFrameCodec.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/transport/tcp/TcpFrameCodec.java index b925b8dfa6..dabb615d49 100644 --- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/transport/tcp/TcpFrameCodec.java +++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/transport/tcp/TcpFrameCodec.java @@ -17,22 +17,23 @@ package org.apache.eventmesh.runtime.transport.tcp; -import io.cloudevents.CloudEvent; +import org.apache.eventmesh.common.wire.EventMeshFrame; /** - * Encodes a delivered CloudEvent (plus its delivery id) into the TCP {@code Package} wire bytes the - * legacy client expects, and decodes a client ACK frame back into the delivery id it acknowledges. + * Encodes a delivered {@link EventMeshFrame} (plus its delivery id) into the TCP {@code Package} + * wire bytes the legacy client expects, and decodes a client ACK frame back into the delivery id it + * acknowledges. * - *

Production implementation reuses the existing {@code Codec} + {@code MeshMessageProtocolAdaptor} - * (reverse direction) and carries the delivery id in a Package header/extension so the client's ACK - * frame echoes it. Tests inject a deterministic stub.

+ *

Production implementation reuses the existing {@code Codec} + the {@code meshmessage} + * {@code FrameAdaptor} (reverse direction) and carries the delivery id in a Package header/extension + * so the client's ACK frame echoes it. Tests inject a deterministic stub.

*/ public interface TcpFrameCodec { /** - * Encode a push frame for {@code event}, tagged with {@code deliveryId} so the client can ACK it. + * Encode a push frame for {@code frame}, tagged with {@code deliveryId} so the client can ACK it. */ - byte[] encodePush(String deliveryId, CloudEvent event); + byte[] encodePush(String deliveryId, EventMeshFrame frame); /** * Extract the delivery id from a client ACK frame, or {@code null} if the frame isn't an ACK. diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/transport/tcp/TcpPushChannel.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/transport/tcp/TcpPushChannel.java index 5c8836fbad..68e4d489c1 100644 --- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/transport/tcp/TcpPushChannel.java +++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/transport/tcp/TcpPushChannel.java @@ -21,19 +21,18 @@ import org.apache.eventmesh.runtime.delivery.AckCallback; import org.apache.eventmesh.runtime.delivery.PushChannel; -import io.cloudevents.CloudEvent; - import lombok.extern.slf4j.Slf4j; /** * A legacy TCP client session as a {@link PushChannel} (egress side of the compatibility bridge). * *

When the new {@code ReliableDispatcher} picks a TCP subscriber as a target, it hands the - * CloudEvent here; this channel encodes it into the legacy {@code Package} wire format (via - * {@link TcpFrameCodec}), writes it to the socket ({@code TcpSessionSink}), and parks the ACK - * callback in {@link TcpAckRegistry} until the client's ACK frame arrives. Reliability (redelivery, - * DLQ) is therefore shared with every other transport — the TCP client just looks like another - * push target to the core.

+ * {@link EventMeshFrame} here (the internal wire unit — no CloudEvent intermediary since #5299); + * this channel encodes it into the legacy {@code Package} wire format (via {@link TcpFrameCodec}), + * writes it to the socket ({@code TcpSessionSink}), and parks the ACK callback in + * {@link TcpAckRegistry} until the client's ACK frame arrives. Reliability (redelivery, DLQ) is + * therefore shared with every other transport — the TCP client just looks like another push target + * to the core.

*/ @Slf4j public class TcpPushChannel implements PushChannel { @@ -57,20 +56,12 @@ public TcpPushChannel(TcpFrameCodec codec, TcpSessionSink sink, TcpAckRegistry a @Override public void deliver(String deliveryId, EventMeshFrame event, AckCallback callback) { - // Egress boundary: convert the internal Frame to a CloudEvent for the legacy TCP wire format. - CloudEvent ce; - try { - ce = event.toCloudEvent(); - } catch (RuntimeException e) { - log.warn("tcp push frame->CloudEvent conversion failed for delivery={}", deliveryId, e); - callback.nack(e); - return; - } + // Egress boundary: encode the internal Frame straight onto the legacy TCP wire format. byte[] frame; try { - frame = codec.encodePush(deliveryId, ce); + frame = codec.encodePush(deliveryId, event); } catch (RuntimeException e) { - log.warn("tcp push encode failed for delivery={}", deliveryId, e); + log.warn("tcp push frame->wire encode failed for delivery={}", deliveryId, e); callback.nack(e); return; } diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/transport/tcp/UniTcpServer.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/transport/tcp/UniTcpServer.java index a9988e5101..8a5d471bc6 100644 --- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/transport/tcp/UniTcpServer.java +++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/transport/tcp/UniTcpServer.java @@ -60,19 +60,16 @@ public class UniTcpServer { private final UniIngressService ingress; private final TcpAckRegistry ackRegistry; private final PackageRouter router; - private final CloudEventToPackageBody bodyMapper; private EventLoopGroup bossGroup; private EventLoopGroup workerGroup; private Channel serverChannel; private final ConcurrentHashMap clientChannels = new ConcurrentHashMap<>(); - public UniTcpServer(UniIngressService ingress, TcpAckRegistry ackRegistry, PackageRouter router, - CloudEventToPackageBody bodyMapper) { + public UniTcpServer(UniIngressService ingress, TcpAckRegistry ackRegistry, PackageRouter router) { this.ingress = ingress; this.ackRegistry = ackRegistry; this.router = router; - this.bodyMapper = bodyMapper; } /** @@ -93,7 +90,7 @@ protected void initChannel(SocketChannel ch) { ch.pipeline() .addLast(new Codec.Encoder()) .addLast(new Codec.Decoder()) - .addLast(new FrameHandler(ingress, ackRegistry, router, clientChannels, bodyMapper)); + .addLast(new FrameHandler(ingress, ackRegistry, router, clientChannels)); } }); serverChannel = bootstrap.bind(port).sync().channel(); @@ -119,7 +116,9 @@ public void stop() { * UNSUBSCRIBE/GOODBYE) are handled here directly because they need channel context (the * subscriber's clientId comes from the HELLO {@link UserAgent#getGroup()}, not the SUBSCRIBE * body). Message commands (ASYNC_MESSAGE_TO_SERVER / ASYNC_MESSAGE_TO_CLIENT_ACK) go through the - * {@link PackageRouter} for CloudEvents translation. Static + package-private so it can be + * {@link PackageRouter}, which decodes the MeshMessage straight into the internal + * {@link org.apache.eventmesh.common.wire.EventMeshFrame} wire unit — no CloudEvent hop since + * #5299. Static + package-private so it can be * exercised directly via netty {@code EmbeddedChannel} in tests. */ static final class FrameHandler extends SimpleChannelInboundHandler { @@ -131,15 +130,13 @@ static final class FrameHandler extends SimpleChannelInboundHandler { private final TcpAckRegistry ackRegistry; private final PackageRouter router; private final ConcurrentHashMap clientChannels; - private final CloudEventToPackageBody bodyMapper; FrameHandler(UniIngressService ingress, TcpAckRegistry ackRegistry, PackageRouter router, - ConcurrentHashMap clientChannels, CloudEventToPackageBody bodyMapper) { + ConcurrentHashMap clientChannels) { this.ingress = ingress; this.ackRegistry = ackRegistry; this.router = router; this.clientChannels = clientChannels; - this.bodyMapper = bodyMapper; } @Override diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/LegacyTcpClientIntegrationTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/LegacyTcpClientIntegrationTest.java index a484bd445c..c2babba48e 100644 --- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/LegacyTcpClientIntegrationTest.java +++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/LegacyTcpClientIntegrationTest.java @@ -33,7 +33,6 @@ import org.apache.eventmesh.common.wire.EventMeshFrame; import org.apache.eventmesh.runtime.ingress.UniIngressService; import org.apache.eventmesh.runtime.offset.InMemoryOffsetStore; -import org.apache.eventmesh.runtime.transport.tcp.MeshEventToPackageBody; import org.apache.eventmesh.runtime.transport.tcp.MeshMessagePackageRouter; import org.apache.eventmesh.runtime.transport.tcp.TcpAckRegistry; import org.apache.eventmesh.runtime.transport.tcp.UniTcpServer; @@ -159,8 +158,7 @@ private void boot() throws Exception { } }, 0, 100, java.util.concurrent.TimeUnit.MILLISECONDS); - server = new UniTcpServer(ingress, new TcpAckRegistry(), new MeshMessagePackageRouter(), - new MeshEventToPackageBody()); + server = new UniTcpServer(ingress, new TcpAckRegistry(), new MeshMessagePackageRouter()); port = server.start(0); } diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/LegacyTcpClusterBrokerIntegrationTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/LegacyTcpClusterBrokerIntegrationTest.java index a466b233bb..b292b1ed4d 100644 --- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/LegacyTcpClusterBrokerIntegrationTest.java +++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/LegacyTcpClusterBrokerIntegrationTest.java @@ -31,7 +31,6 @@ import org.apache.eventmesh.runtime.boot.EventMeshApplication; import org.apache.eventmesh.runtime.cluster.NacosMetaStore; import org.apache.eventmesh.runtime.offset.InMemoryOffsetStore; -import org.apache.eventmesh.runtime.transport.tcp.MeshEventToPackageBody; import org.apache.eventmesh.runtime.transport.tcp.MeshMessagePackageRouter; import org.apache.eventmesh.runtime.transport.tcp.TcpAckRegistry; import org.apache.eventmesh.runtime.transport.tcp.UniTcpServer; @@ -119,7 +118,7 @@ void oldSdkOverRealBrokerAndNacosMeta() throws Exception { // 3. TCP server (not auto-booted by EventMeshApplication) on the cluster-enabled ingress. tcpServer = new UniTcpServer(app.runtime().ingress(), new TcpAckRegistry(), - new MeshMessagePackageRouter(), new MeshEventToPackageBody()); + new MeshMessagePackageRouter()); tcpPort = tcpServer.start(0); // 4. Subscriber: real old SDK. clientId = HELLO UserAgent.group. diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/security/EventMeshFrameFilterTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/security/EventMeshFrameFilterTest.java new file mode 100644 index 0000000000..4668e46774 --- /dev/null +++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/security/EventMeshFrameFilterTest.java @@ -0,0 +1,145 @@ +/* + * 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.security; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import org.apache.eventmesh.common.wire.EventMeshFrame; + +import java.util.LinkedHashMap; +import java.util.Map; + +import org.junit.jupiter.api.Test; + +/** + * #5299 sub-PR B: the filter chain now operates on {@link EventMeshFrame} instead of + * {@code io.cloudevents.CloudEvent}. These tests exercise the new primary path; the + * CloudEvent-based tests in {@link SecurityFilterTest} remain as the legacy bridge contract. + */ +class EventMeshFrameFilterTest { + + @Test + void tokenAuthReadsCredentialFromContextNotFrame() { + TokenAuthFilter auth = new TokenAuthFilter(java.util.Collections.singleton("good-token")); + + assertTrue(auth.check(frame(), ctx("good-token", "tenantA", "orders")).isAllowed()); + FilterVerdict missing = auth.check(frame(), ctx(null, "tenantA", "orders")); + assertFalse(missing.isAllowed()); + assertEquals(401, missing.getRejectStatus()); + } + + @Test + void aclDeniesFrameWithNonEventMsgType() { + // A STREAM_REQ frame should be rejected by the ACL filter even if the principal/resource + // are otherwise fine — the filter applies to event ingress only. We craft the frame via + // decode() because the 5-arg ctor is package-private; the wire format is documented in + // EventMeshFrame (header = magic(1) | ver(1) | msgType(1) | flags(1) | seq(4) | keyCount(2) | dataLen(4)). + byte[] streamReqBytes = new byte[] { + (byte) 0xEF, 1, EventMeshFrame.TYPE_STREAM_REQ, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0 + }; + EventMeshFrame streamFrame = EventMeshFrame.decode(streamReqBytes); + AclFilter acl = new AclFilter(java.util.List.of( + new AclRule("tenantA", "orders", AclRule.Action.ANY, AclRule.Effect.ALLOW, 10))); + FilterVerdict denied = acl.check(streamFrame, ctx("tok", "tenantA", "orders")); + assertFalse(denied.isAllowed()); + assertEquals(403, denied.getRejectStatus()); + } + + @Test + void aclAllowsGrantedTopicRejectsForbidden() { + AclFilter acl = new AclFilter(java.util.List.of( + new AclRule("tenantA", "orders", AclRule.Action.ANY, AclRule.Effect.ALLOW, 10))); + + assertTrue(acl.check(frame(), ctx("tok", "tenantA", "orders")).isAllowed()); + FilterVerdict forbidden = acl.check(frame(), ctx("tok", "tenantA", "payments")); + assertFalse(forbidden.isAllowed()); + assertEquals(403, forbidden.getRejectStatus()); + } + + @Test + void chainFailsClosedAtFirstDenyingFilter() { + TokenAuthFilter auth = new TokenAuthFilter(java.util.Collections.singleton("good-token")); + AclFilter acl = new AclFilter(java.util.List.of( + new AclRule("tenantA", "orders", AclRule.Action.ANY, AclRule.Effect.ALLOW, 10))); + FilterChain chain = new FilterChain(auth, acl); + + // Bad credential → denied by auth (401), acl never consulted. + FilterVerdict badCred = chain.check(frame(), ctx("wrong", "tenantA", "orders")); + assertFalse(badCred.isAllowed()); + assertEquals(401, badCred.getRejectStatus()); + + // Good credential but forbidden topic → denied by acl (403). + FilterVerdict badTopic = chain.check(frame(), ctx("good-token", "tenantA", "payments")); + assertFalse(badTopic.isAllowed()); + assertEquals(403, badTopic.getRejectStatus()); + + // Both pass → allow. + assertTrue(chain.check(frame(), ctx("good-token", "tenantA", "orders")).isAllowed()); + } + + @Test + void signatureVerifierReadsSignatureFromFrameAttributes() { + SignatureVerifierFilter verifier = new SignatureVerifierFilter("shared-secret"); + EventMeshFrame unsigned = frame(); + String goodSig = verifier.sign(SignatureVerifierFilter.canonical(unsigned)); + EventMeshFrame signed = withAttr(unsigned, SignatureVerifierFilter.EXT_SIGNATURE, goodSig); + + assertTrue(verifier.check(signed, ctx("tok", "tenantA", "orders")).isAllowed()); + + // Tampered signature (flip last hex char) → reject. + String tampered = goodSig.substring(0, goodSig.length() - 1) + + (goodSig.charAt(goodSig.length() - 1) == '0' ? '1' : '0'); + EventMeshFrame bad = withAttr(signed, SignatureVerifierFilter.EXT_SIGNATURE, tampered); + assertFalse(verifier.check(bad, ctx("tok", "tenantA", "orders")).isAllowed()); + + // Missing signature → reject. + assertFalse(verifier.check(unsigned, ctx("tok", "tenantA", "orders")).isAllowed(), + "missing signature rejected"); + } + + @Test + void tenantFromFrameAttributesReachesContext() { + // The HTTP handler now reads tenant from frame.attributes() and threads it into + // FilterContext before the chain runs; this test pins that contract. + EventMeshFrame f = withAttr(frame(), "emtenantid", "tenantZ"); + String tenant = f.attributes().get("emtenantid"); + assertEquals("tenantZ", tenant); + // The downstream AclFilter then uses ctx.tenant — the test in aclAllowsGrantedTopicRejectsForbidden + // already exercises the "tenant from ctx" path; here we just document the contract. + } + + private static EventMeshFrame frame() { + Map attrs = new LinkedHashMap<>(); + attrs.put("id", "e-1"); + attrs.put("source", "svc"); + attrs.put("type", "order.created"); + return EventMeshFrame.event(attrs, new byte[0]); + } + + private static EventMeshFrame withAttr(EventMeshFrame base, String name, String value) { + Map attrs = new LinkedHashMap<>(base.attributes()); + attrs.put(name, value); + return EventMeshFrame.event(attrs, base.data()); + } + + private static FilterContext ctx(String credential, String tenant, String topic) { + return new FilterContext(topic, "client-1", tenant, credential, "127.0.0.1"); + } +} diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/security/SecurityFilterTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/security/SecurityFilterTest.java index 3993c0c7e1..116854653a 100644 --- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/security/SecurityFilterTest.java +++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/security/SecurityFilterTest.java @@ -21,6 +21,8 @@ import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertTrue; +import org.apache.eventmesh.common.wire.EventMeshFrame; + import java.net.URI; import java.util.Collections; @@ -94,7 +96,7 @@ void signatureVerifierAcceptsValidRejectsTampered() { SignatureVerifierFilter verifier = new SignatureVerifierFilter("shared-secret"); CloudEvent event = event(); - String goodSig = verifier.sign(SignatureVerifierFilter.canonical(event)); + String goodSig = verifier.sign(SignatureVerifierFilter.canonical(EventMeshFrame.fromCloudEvent(event))); CloudEvent signed = CloudEventBuilder.from(event).withExtension(SignatureVerifierFilter.EXT_SIGNATURE, goodSig).build(); assertTrue(verifier.check(signed, ctx("tok", "tenantA", "orders")).isAllowed()); diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/transport/tcp/MeshMessagePackageRouterTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/transport/tcp/MeshMessagePackageRouterTest.java index fee3281050..bfa79a95f2 100644 --- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/transport/tcp/MeshMessagePackageRouterTest.java +++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/transport/tcp/MeshMessagePackageRouterTest.java @@ -25,6 +25,8 @@ import org.apache.eventmesh.common.protocol.tcp.EventMeshMessage; import org.apache.eventmesh.common.protocol.tcp.Header; import org.apache.eventmesh.common.protocol.tcp.Package; +import org.apache.eventmesh.common.wire.EventMeshFrame; +import org.apache.eventmesh.protocol.meshmessage.MeshMessageFrameAdaptor; import java.net.URI; import java.nio.charset.StandardCharsets; @@ -42,14 +44,16 @@ class MeshMessagePackageRouterTest { @Test - void egressEncodesCloudEventIntoEventMeshMessageBody() { + void egressEncodesFrameIntoEventMeshMessageBody() throws Exception { CloudEvent event = CloudEventBuilder.v1() .withId("e-1").withSource(URI.create("svc")).withType("order.created") .withSubject("orders") .withData("hello".getBytes(StandardCharsets.UTF_8)) .build(); + EventMeshFrame frame = EventMeshFrame.fromCloudEvent(event); - Object body = new MeshEventToPackageBody().toBody(event); + // Egress is frame-native since #5299: Frame -> Package(EventMeshMessage), no CloudEvent hop. + Object body = ((Package) new MeshMessageFrameAdaptor().fromFrame(frame)).getBody(); assertNotNull(body, "egress body must be produced"); assertTrue(body instanceof EventMeshMessage, "body is the legacy EventMeshMessage"); @@ -70,9 +74,9 @@ void ackFrameRoutesToAckRequest() { assertEquals("d-99", req.getDeliveryId()); } - // NOTE: publish ingress (ASYNC_MESSAGE_TO_SERVER → CloudEvent) routes through - // MeshMessageProtocolAdaptor.toCloudEvent, which is itself covered by eventmesh-protocol- - // meshmessage's own test suite with real wire packages (body=JSON string + protocol header - // properties produced by the Codec). The in-JVM object/string asymmetry makes a direct - // round-trip unit test unrepresentative, so it is intentionally not asserted here. + // NOTE: publish ingress (ASYNC_MESSAGE_TO_SERVER → EventMeshFrame) routes through + // MeshMessageFrameAdaptor, which is itself covered by eventmesh-protocol-meshmessage's own + // test suite with real wire packages (body=JSON string + protocol header properties produced + // by the Codec). The in-JVM object/string asymmetry makes a direct round-trip unit test + // unrepresentative, so it is intentionally not asserted here. } diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/transport/tcp/TcpCompatibilityBridgeTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/transport/tcp/TcpCompatibilityBridgeTest.java index 2e99641bf7..e0d8ae5db3 100644 --- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/transport/tcp/TcpCompatibilityBridgeTest.java +++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/transport/tcp/TcpCompatibilityBridgeTest.java @@ -111,8 +111,8 @@ private static CloudEvent event(String id) { private static final class StubCodec implements TcpFrameCodec { @Override - public byte[] encodePush(String deliveryId, CloudEvent event) { - return ("PUSH:" + deliveryId + ":" + event.getId()).getBytes(StandardCharsets.UTF_8); + public byte[] encodePush(String deliveryId, EventMeshFrame frame) { + return ("PUSH:" + deliveryId + ":" + frame.attributes().get("id")).getBytes(StandardCharsets.UTF_8); } @Override diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/transport/tcp/UniTcpServerTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/transport/tcp/UniTcpServerTest.java index 425bbf0592..034f9cbe55 100644 --- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/transport/tcp/UniTcpServerTest.java +++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/transport/tcp/UniTcpServerTest.java @@ -98,15 +98,8 @@ void fullPushToClientAndAckLoopAdvancesOffset() throws Exception { } return null; }; - // Body mapper: event → a Map the "client" (test) can read back. - CloudEventToPackageBody bodyMapper = event -> { - Map b = new HashMap<>(); - b.put("id", event.getId()); - return b; - }; - EmbeddedChannel client = new EmbeddedChannel(new UniTcpServer.FrameHandler( - ingress, new TcpAckRegistry(), router, new ConcurrentHashMap<>(), bodyMapper)); + ingress, new TcpAckRegistry(), router, new ConcurrentHashMap<>())); // 1a. client HELLO (carries clientId in UserAgent.group) → server stashes it on the channel. UserAgent ua = UserAgent.builder().group("c1").host("test").port(1).build(); @@ -149,7 +142,7 @@ void fullPushToClientAndAckLoopAdvancesOffset() throws Exception { private static UniTcpServer.FrameHandler newHandler(UniIngressService ingress, PackageRouter router) { return new UniTcpServer.FrameHandler(ingress, new TcpAckRegistry(), router, - new ConcurrentHashMap<>(), event -> null); + new ConcurrentHashMap<>()); } private static final class InMemoryStorage implements MeshStoragePlugin {