From 69e808fd2f476fb70109174cf7720ad158f1e79c Mon Sep 17 00:00:00 2001 From: qqeasonchen Date: Mon, 31 Aug 2026 16:33:53 +0800 Subject: [PATCH 1/3] fix(protocol): route UniHttpServer ingress through FrameAdaptor SPI (#5299 Sub-PR A) The HTTP ingress path now converts structured CloudEvents JSON bytes to EventMeshFrame at the boundary via FrameAdaptor SPI, so UniIngressService no longer imports io.cloudevents.CloudEvent on the HTTP code path. New ingress methods (Frame-typed): - publishBatchFrames(topic, List) - publishLiteFrame(parent, lite, EventMeshFrame) - pollLiteFrames(parent, lite, max, timeoutMs) -> List - requestFrame(topic, EventMeshFrame, timeout) -> EventMeshFrame - replyFrame(correlationId, EventMeshFrame) The CloudEvent-typed overloads are preserved for binary compatibility with the TCP bridge. Sub-PR B will migrate FilterChain; Sub-PR C will migrate the TCP path. Closes part of #5299. --- .../eventmesh/runtime/http/UniHttpServer.java | 71 +- .../runtime/ingress/UniIngressService.java | 1891 +++++++++-------- 2 files changed, 1031 insertions(+), 931 deletions(-) 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..3bf62e914a 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; @@ -285,28 +284,36 @@ 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. + // TODO(#5299 sub-PR B): convert filterChain.check() to take an EventMeshFrame and look up + // emtenantid from frame attributes; the CloudEvent here is a temporary bridge until the + // filter chain migrates. + io.cloudevents.CloudEvent eventForAcl = frame.toCloudEvent(); if (filterChain != null) { String credential = exchange.getRequestHeaders().getFirst("Authorization"); - String tenant = event.getExtension("emtenantid") != null ? event.getExtension("emtenantid").toString() : null; + String tenant = eventForAcl.getExtension("emtenantid") != null + ? eventForAcl.getExtension("emtenantid").toString() : null; 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(eventForAcl, 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 +344,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 +493,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 +521,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 +628,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 +660,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; + }; + } +} From f17c3c7c49062e2ddc6f5c668696331494b642cd Mon Sep 17 00:00:00 2001 From: qqeasonchen Date: Mon, 31 Aug 2026 16:54:11 +0800 Subject: [PATCH 2/3] fix(protocol): migrate ingress filter chain to EventMeshFrame (#5299 Sub-PR B) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The ingress security pipeline (§4.5) — TokenAuthFilter / AclFilter / SignatureVerifierFilter / FilterChain — now operates on the runtime's internal wire format (org.apache.eventmesh.common.wire.EventMeshFrame) instead of io.cloudevents.CloudEvent. This continues the #5299 single- protocol-path migration: sub-PR A routed the HTTP ingress through the FrameAdaptor SPI; sub-PR B extends the same boundary inward to the filter chain so that auth/acl/signature now read directly from frame.attributes(). Changes: - IngressFilter: new check(EventMeshFrame, FilterContext) primary method; the CloudEvent overload is now a default that bridges via EventMeshFrame.fromCloudEvent(...) for backward compat. - FilterChain: new check(EventMeshFrame, FilterContext) overload; CloudEvent variant is deprecated and bridges via fromCloudEvent. - AclFilter: implement check(EventMeshFrame, ...) — also rejects non-EVENT frames (e.g. STREAM_REQ / STREAM_CHUNK) since the ingress security pipeline is event-shaped only. - TokenAuthFilter: implement check(EventMeshFrame, ...) — credential still comes from FilterContext (HTTP Authorization header), so the frame body is unused for this stage. - SignatureVerifierFilter: implement check(EventMeshFrame, ...) — signature travels in frame.attributes() under the same key (emsignature) the legacy CloudEvent extension used, so signed CloudEvents round-trip transparently through the cloudevents FrameAdaptor. - UniHttpServer: drop the temporary frame.toCloudEvent() bridge in the publish() filter call; thread tenant directly from frame.attributes(). Also drop the synthetic CloudEvent stub used for pre-publish security check — replaced with a minimal EventMeshFrame.event(emptyMap, []). - New test class: EventMeshFrameFilterTest (6 tests) exercising the EventMeshFrame-typed path parallel to the existing SecurityFilterTest which still covers the CloudEvent bridge. Backward compat: existing custom filters implementing IngressFilter.check(CloudEvent, ...) keep working because the interface now provides a default implementation. They will compile unchanged and continue to be invoked via the deprecated chain overload until sub-PR C migrates the TCP path and we can drop the bridge entirely. Refs #5299. --- .../eventmesh/runtime/http/UniHttpServer.java | 23 +-- .../eventmesh/runtime/security/AclFilter.java | 14 +- .../runtime/security/FilterChain.java | 20 ++- .../runtime/security/IngressFilter.java | 23 ++- .../security/SignatureVerifierFilter.java | 29 ++-- .../runtime/security/TokenAuthFilter.java | 4 +- .../security/EventMeshFrameFilterTest.java | 144 ++++++++++++++++++ 7 files changed, 223 insertions(+), 34 deletions(-) create mode 100644 eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/security/EventMeshFrameFilterTest.java 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 3bf62e914a..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 @@ -248,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; @@ -295,18 +298,16 @@ private void publish(HttpExchange exchange) throws IOException { return; } // Security filter chain (§4.5): auth/acl/signature run before the event enters the pipeline. - // TODO(#5299 sub-PR B): convert filterChain.check() to take an EventMeshFrame and look up - // emtenantid from frame attributes; the CloudEvent here is a temporary bridge until the - // filter chain migrates. - io.cloudevents.CloudEvent eventForAcl = frame.toCloudEvent(); + // #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 = eventForAcl.getExtension("emtenantid") != null - ? eventForAcl.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(eventForAcl, ctx); + org.apache.eventmesh.runtime.security.FilterVerdict verdict = filterChain.check(frame, ctx); if (!verdict.isAllowed()) { writeJson(exchange, verdict.getRejectStatus(), error(verdict.getReason())); return; 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..816e9e3f05 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 @@ -21,7 +21,7 @@ import java.util.Collections; import java.util.List; -import io.cloudevents.CloudEvent; +import org.apache.eventmesh.common.wire.EventMeshFrame; /** * Topic-level authorization (§13.4.2). Holds an {@link AclRule} set sorted by priority (descending, @@ -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..2843cc4f15 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,35 @@ package org.apache.eventmesh.runtime.security; -import io.cloudevents.CloudEvent; +import org.apache.eventmesh.common.wire.EventMeshFrame; /** * 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..ad7a718354 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 @@ -19,11 +19,12 @@ 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; +import org.apache.eventmesh.common.wire.EventMeshFrame; /** * Verifies an HMAC-SHA256 signature over a canonical projection of the CloudEvent (§13.4.4), so 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..d565b19fe1 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 @@ -20,7 +20,7 @@ import java.util.Collections; import java.util.Set; -import io.cloudevents.CloudEvent; +import org.apache.eventmesh.common.wire.EventMeshFrame; /** * Authenticates the caller by a bearer token (§13.4.1). In production this delegates to the @@ -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/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..2fc18b35c3 --- /dev/null +++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/security/EventMeshFrameFilterTest.java @@ -0,0 +1,144 @@ +/* + * 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 org.junit.jupiter.api.Test; + +import java.util.LinkedHashMap; +import java.util.Map; + +/** + * #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"); + } +} From 4cf9c8ce694bfb5d91aecce3be68817b55f2f8d2 Mon Sep 17 00:00:00 2001 From: qqeasonchen Date: Mon, 31 Aug 2026 17:32:35 +0800 Subject: [PATCH 3/3] fix(protocol): complete #5299 single protocol path (sub-PR B fix + C + D) Sub-PR B fix: - restore the missing io.cloudevents.CloudEvent import in IngressFilter (broke compileJava) - move EventMeshFrame into the correct checkstyle ImportOrder group (style/checkStyle.xml puts org.apache.eventmesh first; maxWarnings=0 so it fails CI) - update SecurityFilterTest for the frame-based SignatureVerifierFilter.canonical(...) Sub-PR C: migrate the TCP egress path to EventMeshFrame - TcpFrameCodec.encodePush / TcpPushChannel.deliver now take an EventMeshFrame directly, dropping the frame -> CloudEvent -> wire round trip - delete the dead CloudEvent-era egress SPI: CloudEventToPackageBody, MeshEventToPackageBody - UniTcpServer drops the now-unused bodyMapper constructor parameter - TCP ingress was already frame-native (MeshMessagePackageRouter / TcpRequest / NettyTcpPushChannel) Sub-PR D: document the #5299 acceptance matrix and protocol status labels (docs/eventmesh-uni-architecture-redesign.md section 19.6) Refs #5299. --- docs/eventmesh-uni-architecture-redesign.md | 101 +++++++++++++++++- .../eventmesh/runtime/security/AclFilter.java | 4 +- .../runtime/security/IngressFilter.java | 2 + .../security/SignatureVerifierFilter.java | 4 +- .../runtime/security/TokenAuthFilter.java | 4 +- .../tcp/CloudEventToPackageBody.java | 31 ------ .../transport/tcp/MeshEventToPackageBody.java | 50 --------- .../runtime/transport/tcp/TcpFrameCodec.java | 17 +-- .../runtime/transport/tcp/TcpPushChannel.java | 27 ++--- .../runtime/transport/tcp/UniTcpServer.java | 15 ++- .../it/LegacyTcpClientIntegrationTest.java | 4 +- ...LegacyTcpClusterBrokerIntegrationTest.java | 3 +- .../security/EventMeshFrameFilterTest.java | 3 +- .../runtime/security/SecurityFilterTest.java | 4 +- .../tcp/MeshMessagePackageRouterTest.java | 18 ++-- .../tcp/TcpCompatibilityBridgeTest.java | 4 +- .../transport/tcp/UniTcpServerTest.java | 11 +- 17 files changed, 154 insertions(+), 148 deletions(-) delete mode 100644 eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/transport/tcp/CloudEventToPackageBody.java delete mode 100644 eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/transport/tcp/MeshEventToPackageBody.java 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/security/AclFilter.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/AclFilter.java index 816e9e3f05..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 org.apache.eventmesh.common.wire.EventMeshFrame; - /** * 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, 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 2843cc4f15..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 @@ -19,6 +19,8 @@ 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 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 ad7a718354..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,6 +17,8 @@ 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; @@ -24,8 +26,6 @@ import javax.crypto.Mac; import javax.crypto.spec.SecretKeySpec; -import org.apache.eventmesh.common.wire.EventMeshFrame; - /** * 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 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 d565b19fe1..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 org.apache.eventmesh.common.wire.EventMeshFrame; - /** * 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 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 index 2fc18b35c3..4668e46774 100644 --- 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 @@ -22,11 +22,12 @@ import static org.junit.jupiter.api.Assertions.assertTrue; import org.apache.eventmesh.common.wire.EventMeshFrame; -import org.junit.jupiter.api.Test; 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 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 {