You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@eventmesh.apache.org by ch...@apache.org on 2026/09/15 07:57:41 UTC
(eventmesh) branch develop updated: [ISSUE #5399] Add eventmesh-storage-memory in-memory storage plugin (#5400)
This is an automated email from the ASF dual-hosted git repository.
qqeasonchen pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/eventmesh.git
The following commit(s) were added to refs/heads/develop by this push:
new 49a86751c [ISSUE #5399] Add eventmesh-storage-memory in-memory storage plugin (#5400)
49a86751c is described below
commit 49a86751c2cc3180ea99ab34b26cfbd52cda21b5
Author: Eason Chen <qq...@gmail.com>
AuthorDate: Tue Sep 15 15:57:34 2026 +0800
[ISSUE #5399] Add eventmesh-storage-memory in-memory storage plugin (#5400)
* [ISSUE #5399] Add eventmesh-storage-memory in-memory storage plugin
- New module eventmesh-storage-plugin:eventmesh-storage-memory:/n MemoryMeshStoragePlugin implements MeshStoragePlugin + all 7 StorageCapabilities
sub-interfaces (production-scope promotion of the InMemoryStoragePlugin TCK
self-test fixture from eventmesh-storage-api src/test).
- SPI registration memory=org.apache.eventmesh.storage.memory.MemoryMeshStoragePlugin
(select with EVENTMESH_STORAGE_TYPE=memory).
- settings.gradle module include.
- docs/feature/storage-spi.md capability matrix gains a Memory column.
- TCK subclass MemoryMeshStoragePluginTCKTest + round-trip unit tests.
* [ISSUE #5399] Make memory the zero-dependency default + add memory e2e test
- MemoryStorageE2EIntegrationTest: always-on publish->subscribe->poll->ack->
offset-advance round trip over the full UniRuntime path with the memory plugin
loaded via SPI. No broker, no -D gate - runs on every CI build.
- eventmesh-storage-memory/gradle.properties: pluginType=storage/pluginName=memory
so installPlugin ships the plugin jar in the runtime image.
- Default storage switched to memory for zero-dependency quick start:
docker/Dockerfile (ENV EVENTMESH_STORAGE_TYPE=memory), k8s runtime-statefulset,
EventMeshApplication system-property default ("standalone"->"memory"),
conf/eventmesh.properties backend list.
- Docs: getting-started (docker run with no env / source path / backend table),
configuration.md (memory section + default), k8s README + configmap comments.
- eventmesh-runtime test classpath gains the memory plugin.
* [ISSUE #5399] Honor the self-resume pull contract in MemoryMeshStoragePlugin
The runtime pull-loop calls storage.poll(topic, partition, -1, ...) where
startOffset<0 means "resume from the plugin-managed position" (Kafka/RocketMQ
use the consumer position). The TCK-fixture- inherited implementation returned
the whole log from offset 0 on every tick, which under the real pull-loop would
re-dispatch all history each tick.
- poll(): keep a per-(topic, partition) cursor; startOffset<0 resumes from it,
advance it after each successful batch.
- alignPullOffset(): rewind that same per-key cursor (was a single global
counter in the fixture), wiring the ACK-timeout redelivery path.
* [ISSUE #5399] checkstyle: drop unused import in MemoryMeshStoragePluginTCKTest
---
deploy/kubernetes/README.md | 9 +-
deploy/kubernetes/runtime-configmap.yaml | 2 +-
deploy/kubernetes/runtime-statefulset.yaml | 2 +-
docker/Dockerfile | 8 +-
docs/feature/storage-spi.md | 18 +-
docs/quickstart/configuration.md | 11 +-
docs/quickstart/getting-started.md | 17 +-
eventmesh-runtime/build.gradle | 2 +
eventmesh-runtime/conf/eventmesh.properties | 3 +-
.../runtime/boot/EventMeshApplication.java | 4 +-
.../it/MemoryStorageE2EIntegrationTest.java | 126 ++++++++++
.../eventmesh-storage-memory/build.gradle | 33 +++
.../eventmesh-storage-memory/gradle.properties | 19 ++
.../storage/memory/MemoryMeshStoragePlugin.java | 259 +++++++++++++++++++++
....apache.eventmesh.api.storage.MeshStoragePlugin | 16 ++
.../memory/MemoryMeshStoragePluginTCKTest.java | 94 ++++++++
settings.gradle | 1 +
17 files changed, 595 insertions(+), 29 deletions(-)
diff --git a/deploy/kubernetes/README.md b/deploy/kubernetes/README.md
index 4e6b41d3d..3a912800d 100644
--- a/deploy/kubernetes/README.md
+++ b/deploy/kubernetes/README.md
@@ -56,7 +56,8 @@ kubectl -n eventmesh-system run curl --rm -it --image=curlimages/curl -- curl
## Storage backends
-The image ships the kafka / rocketmq / rocketmq5 storage plugins. Switch with
-`EVENTMESH_STORAGE_TYPE` on the runtime container and point
-`eventMesh.server.<backend>.namesrvAddr` in the ConfigMap at the in-cluster
-service (e.g. Strimzi Kafka bootstrap or a RocketMQ name server Service).
+The image ships the memory (default, zero-dependency) / kafka / rocketmq / rocketmq5
+storage plugins. Switch with `EVENTMESH_STORAGE_TYPE` on the runtime container
+and point `eventMesh.server.<backend>.namesrvAddr` in the ConfigMap at the
+in-cluster service (e.g. Strimzi Kafka bootstrap or a RocketMQ name server
+Service).
diff --git a/deploy/kubernetes/runtime-configmap.yaml b/deploy/kubernetes/runtime-configmap.yaml
index a730c098c..540a76c99 100644
--- a/deploy/kubernetes/runtime-configmap.yaml
+++ b/deploy/kubernetes/runtime-configmap.yaml
@@ -31,7 +31,7 @@ data:
# here configure the storage plugin; JAVA_OPTS in the StatefulSet adds the
# -D flags (admin token, meta store, ws port, ...) from env/Secrets.
eventmesh.properties: |
- # Storage plugin settings (kafka | rocketmq | rocketmq5; the storage type
+ # Storage plugin settings (memory | kafka | rocketmq | rocketmq5; the storage type
# itself is selected by EVENTMESH_STORAGE_TYPE on the pod).
eventMesh.server.kafka.namesrvAddr=my-cluster-kafka-bootstrap.kafka.svc.cluster.local:9092
eventMesh.server.rocketmq.namesrvAddr=my-rocketmq-name-server.rocketmq.svc.cluster.local:9876
diff --git a/deploy/kubernetes/runtime-statefulset.yaml b/deploy/kubernetes/runtime-statefulset.yaml
index 4a1329de4..feb5e9236 100644
--- a/deploy/kubernetes/runtime-statefulset.yaml
+++ b/deploy/kubernetes/runtime-statefulset.yaml
@@ -56,7 +56,7 @@ spec:
containerPort: 8081
env:
- name: EVENTMESH_STORAGE_TYPE
- value: kafka
+ value: memory
- name: EVENTMESH_HTTP_PORT
value: "8080"
- name: EVENTMESH_ADMIN_PORT
diff --git a/docker/Dockerfile b/docker/Dockerfile
index 8fe530e23..81d61b905 100644
--- a/docker/Dockerfile
+++ b/docker/Dockerfile
@@ -74,10 +74,10 @@ ENV DOCKER=true
ENV EVENTMESH_HOME=/data/app/eventmesh
ENV EVENTMESH_LOG_HOME=/data/app/eventmesh/logs
ENV CONFPATH=/data/app/eventmesh/conf
-# Only kafka/rocketmq storage plugins ship (standalone was removed). Override
-# with -e EVENTMESH_STORAGE_TYPE=rocketmq. Storage bootstrap (namesrv) is read
-# from conf/eventmesh.properties — mount a volume over conf/ to override.
-ENV EVENTMESH_STORAGE_TYPE=kafka
+# memory ships as the zero-dependency default for dev/CI/quick start - no broker
+# needed. Override with -e EVENTMESH_STORAGE_TYPE=kafka|rocketmq|rocketmq5 and
+# supply the backend address via conf/eventmesh.properties (mount a volume over
+ENV EVENTMESH_STORAGE_TYPE=memory
ENV EVENTMESH_HTTP_PORT=8080
ENV EVENTMESH_ADMIN_PORT=8081
ENV EVENTMESH_OFFSET_PATH=/data/app/eventmesh/data/offset
diff --git a/docs/feature/storage-spi.md b/docs/feature/storage-spi.md
index 4f3fe28b3..ee95b357c 100644
--- a/docs/feature/storage-spi.md
+++ b/docs/feature/storage-spi.md
@@ -35,15 +35,15 @@ actually implement them.
## Capability matrix
-| Capability | Universal? | Kafka | RocketMQ 4.x | RocketMQ 5.x | Why |
-|-------------------------|-------------|-------|--------------|--------------|-----------------------------------------------------------------------|
-| TopicManagement | yes (U) | Y | Y | Y | All 3 MQs expose `createTopic` (Kafka Admin / RocketMQ Admin). |
-| PartitionAssignment | yes (U) | Y | Y | Y | All 3 MQs treat the topic as a partitionable stream. |
-| ExplicitOffsetCommit | yes (U) | Y | Y | Y | All 3 MQs persist an explicit commitOffset on demand. |
-| EndOffsetQuery | no (Kafka) | Y | N | N | Only Kafka exposes a high-watermark via `endOffsets`. |
-| AlignPullOffset | mixed | Y | Y | N | Kafka + RocketMQ 4.x manage a client-side pull cursor; R5 POP is broker-managed. |
-| DeferredPopAck | R5 only | N | N | Y | R5 POP holds messages invisible until client ACK. |
-| LiteTopic | R5 only | N | N | Y | R5 RIP-83 lite sub-topics; not in Kafka or R4. |
+| Capability | Universal? | Kafka | RocketMQ 4.x | RocketMQ 5.x | Memory | Why |
+|-------------------------|-------------|-------|--------------|--------------|-------|-----------------------------------------------------------------------|
+| TopicManagement | yes (U) | Y | Y | Y | Y | All 3 MQs expose `createTopic` (Kafka Admin / RocketMQ Admin). |
+| PartitionAssignment | yes (U) | Y | Y | Y | Y | All 3 MQs treat the topic as a partitionable stream. |
+| ExplicitOffsetCommit | yes (U) | Y | Y | Y | Y | All 3 MQs persist an explicit commitOffset on demand. |
+| EndOffsetQuery | no (Kafka) | Y | N | N | Y | Only Kafka exposes a high-watermark via `endOffsets`. |
+| AlignPullOffset | mixed | Y | Y | N | Y | Kafka + RocketMQ 4.x manage a client-side pull cursor; R5 POP is broker-managed. |
+| DeferredPopAck | R5 only | N | N | Y | Y | R5 POP holds messages invisible until client ACK. |
+| LiteTopic | R5 only | N | N | Y | Y | R5 RIP-83 lite sub-topics; not in Kafka or R4. |
U = Universal (every backend MUST implement).
diff --git a/docs/quickstart/configuration.md b/docs/quickstart/configuration.md
index 81d1ed550..a0705ff1e 100644
--- a/docs/quickstart/configuration.md
+++ b/docs/quickstart/configuration.md
@@ -20,12 +20,21 @@ precedence is **system property (`-D`) > environment variable > properties file*
| Key | Env override | Values | Default |
|---|---|---|---|
-| `eventmesh.storage.type` | `EVENTMESH_STORAGE_TYPE` | `rocketmq` / `rocketmq5` / `kafka` | `kafka` |
+| `eventmesh.storage.type` | `EVENTMESH_STORAGE_TYPE` | `memory` / `rocketmq` / `rocketmq5` / `kafka` | `memory` |
<a name="storage-backends"></a>
## 2. Storage backends
+### Memory (`memory`, default)
+
+In-process WAL with zero external dependency - the default for dev / CI / quick start.
+State is process-local and lost on restart; not for production.
+
+| Key | Default | Description |
+|---|---|---|
+| (none) | - | No keys required - no broker address |
+
### RocketMQ 4.x (`rocketmq`)
| Key | Default | Description |
diff --git a/docs/quickstart/getting-started.md b/docs/quickstart/getting-started.md
index d1eaab4c4..47524af9c 100644
--- a/docs/quickstart/getting-started.md
+++ b/docs/quickstart/getting-started.md
@@ -18,7 +18,8 @@ subscriber, using the recommended **HTTP + CloudEvents** path. Configuration ref
## 1. Prerequisites
- JDK 21+ (Temurin recommended)
-- Docker (for the container path), or a local install of one storage backend:
+- Docker (for the container path - the default `memory` backend needs no broker), or a
+ local install of one storage backend for production-like runs:
- [Apache RocketMQ](https://rocketmq.apache.org) 4.x or 5.x, **or**
- [Apache Kafka](https://kafka.apache.org) 2.8+ (3.x recommended)
- (SDK only) Java 11+ application with `eventmesh-sdk-java` on the classpath
@@ -31,6 +32,7 @@ changes.
| Backend | Type value | Notes |
|---|---|---|
+| Memory (default) | `memory` | in-process WAL, zero dependency - dev/CI/quick start only |
| RocketMQ 4.x | `rocketmq` | classic PULL over remoting |
| RocketMQ 5.x | `rocketmq5` | 5.x POP + Lite Topic support |
| Kafka | `kafka` | assign+seek+poll (no consumer groups), SASL/SSL supported |
@@ -42,12 +44,15 @@ changes.
```shell
sudo docker pull apache/eventmesh:latest
sudo docker run -d --name eventmesh \
- -e EVENTMESH_STORAGE_TYPE=kafka \
- -e EVENTMESH_KAFKA_NAMESRV=YOUR_KAFKA:9092 \
-p 8080:8080 -p 8081:8081 \
apache/eventmesh:latest
```
+The image defaults to the `memory` storage backend - zero external
+dependency, ready for a smoke test as-is. For a real broker add
+`-e EVENTMESH_STORAGE_TYPE=kafka` (or `rocketmq` / `rocketmq5`) and the
+backend address keys shown below.
+
Ports: `8080` = traffic HTTP (`/events/*`), `8081` = admin HTTP (`/admin/*`). The WebSocket
push port (`8082`) is opt-in.
@@ -57,9 +62,9 @@ push port (`8082`) is opt-in.
git clone https://github.com/apache/eventmesh.git
cd eventmesh
-# pick your backend via EVENTMESH_STORAGE_TYPE (rocketmq | rocketmq5 | kafka)
-export EVENTMESH_STORAGE_TYPE=kafka
-export EVENTMESH_KAFKA_NAMESRV=localhost:9092
+# the memory backend is the default - no env needed. For a real broker:
+# export EVENTMESH_STORAGE_TYPE=kafka (or rocketmq | rocketmq5)
+# export EVENTMESH_KAFKA_NAMESRV=localhost:9092
./gradlew :eventmesh-runtime:clean :eventmesh-runtime:dist
cd eventmesh-runtime/dist && bash bin/start.sh
diff --git a/eventmesh-runtime/build.gradle b/eventmesh-runtime/build.gradle
index 4cda3b662..f683c41f9 100644
--- a/eventmesh-runtime/build.gradle
+++ b/eventmesh-runtime/build.gradle
@@ -61,6 +61,8 @@ dependencies {
testImplementation project(':eventmesh-storage-plugin:eventmesh-storage-rocketmq5')
// KafkaClientE2EIntegrationTest boots the full app with the kafka plugin (incl. SASL/JAAS).
testImplementation project(':eventmesh-storage-plugin:eventmesh-storage-kafka')
+ // MemoryStorageE2EIntegrationTest boots the runtime over the zero-dependency memory plugin.
+ testImplementation project(':eventmesh-storage-plugin:eventmesh-storage-memory')
// Topic creation in RealBrokerIntegrationTest (broker may have autoCreateTopicEnable=false).
testImplementation 'org.apache.rocketmq:rocketmq-tools:4.9.5'
// ClientBrokerIntegrationTest drives the runtime via the real CloudEventsClient SDK over HTTP.
diff --git a/eventmesh-runtime/conf/eventmesh.properties b/eventmesh-runtime/conf/eventmesh.properties
index 762374945..233fed8c6 100644
--- a/eventmesh-runtime/conf/eventmesh.properties
+++ b/eventmesh-runtime/conf/eventmesh.properties
@@ -20,11 +20,12 @@
# init(). -D system properties (set by bin/start.sh or JVM flags) override these values.
#
# Storage backend — set EVENTMESH_STORAGE_TYPE (or -Deventmesh.storage.type) to one of:
+# memory - in-process WAL, zero external dependency (dev/CI default; NOT for production)
# rocketmq — RocketMQ 4.x (remoting direct connect, classic PULL)
# rocketmq5 — RocketMQ 5.x (remoting direct connect, 5.x POP + Lite Topic)
# kafka — Kafka (kafka-clients, assign+seek+poll, no consumer group, SASL support)
#
-# Uncomment the section matching your backend and set the broker address.
+# memory needs no broker address. Uncomment the section matching any other backend.
#
# ===== RocketMQ 4.x (EVENTMESH_STORAGE_TYPE=rocketmq) =====
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshApplication.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshApplication.java
index a4693695c..eb80ad3b0 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshApplication.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshApplication.java
@@ -34,7 +34,7 @@ import lombok.extern.slf4j.Slf4j;
* the traffic {@link UniHttpServer}, and the independent {@link UniAdminServer} as one process with
* a single graceful-shutdown hook. This is the only {@code main} for the new architecture.
*
- * <p>Config via system properties: {@code eventmesh.storage.type} (kafka/rocketmq/…),
+ * <p>Config via system properties: {@code eventmesh.storage.type} (memory by default; kafka/rocketmq/rocketmq5 for a real broker),
* {@code eventmesh.http.port}, {@code eventmesh.admin.port}, {@code eventmesh.offset.path}.
*/
@Slf4j
@@ -375,7 +375,7 @@ public class EventMeshApplication {
}
public static void main(String[] args) throws Exception {
- String storageType = System.getProperty("eventmesh.storage.type", "standalone");
+ String storageType = System.getProperty("eventmesh.storage.type", "memory");
int httpPort = Integer.getInteger("eventmesh.http.port", 8080);
final int adminPort = Integer.getInteger("eventmesh.admin.port", 8081);
final String offsetPath = System.getProperty("eventmesh.offset.path", "./data/offset");
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/MemoryStorageE2EIntegrationTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/MemoryStorageE2EIntegrationTest.java
new file mode 100644
index 000000000..b9b11d1b9
--- /dev/null
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/MemoryStorageE2EIntegrationTest.java
@@ -0,0 +1,126 @@
+/*
+ * 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.it;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.apache.eventmesh.api.storage.MeshStoragePlugin;
+import org.apache.eventmesh.runtime.boot.UniRuntime;
+import org.apache.eventmesh.runtime.offset.InMemoryOffsetStore;
+import org.apache.eventmesh.runtime.push.BufferedEvent;
+import org.apache.eventmesh.runtime.subscription.DistributionMode;
+import org.apache.eventmesh.spi.EventMeshExtensionFactory;
+import org.apache.eventmesh.storage.memory.MemoryMeshStoragePlugin;
+
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+/**
+ * End-to-end integration test of the uni runtime over the in-memory storage
+ * plugin (zero-dependency tier of the E2E suite).
+ *
+ * <p><b>Always on</b> - unlike {@code RealBrokerIntegrationTest} this needs no
+ * broker, no {@code -Dit.storage} gate and no TCP probe: the memory backend
+ * boots in-process. It exercises the same publish - subscribe - poll - ack -
+ * offset-advance contract over the full {@link UniRuntime} ingress/pull-loop
+ * path, so a regression in the runtime-storage boundary is caught on every
+ * CI build.</p>
+ *
+ * <p>The memory plugin is loaded through the real SPI path
+ * ({@code EventMeshExtensionFactory.getExtension(.., "memory")}) to also prove
+ * the {@code META-INF/eventmesh} registration is well-formed.</p>
+ */
+class MemoryStorageE2EIntegrationTest {
+
+ @Test
+ void publishSubscribePollAckOverMemoryStorage() throws Exception {
+ String topic = "em-it-memory";
+ String clientId = "it-memory-client-1";
+
+ // 1. Load the memory plugin via SPI (the same path the real runtime uses).
+ MeshStoragePlugin storage = EventMeshExtensionFactory.getExtension(MeshStoragePlugin.class, "memory");
+ assertNotNull(storage, "no MeshStoragePlugin registered for 'memory' - check the SPI file"
+ + " META-INF/eventmesh/org.apache.eventmesh.api.storage.MeshStoragePlugin");
+ assertTrue(storage instanceof MemoryMeshStoragePlugin);
+
+ // 2. Boot the runtime with the same wiring RealBrokerIntegrationTest uses.
+ UniRuntime runtime = new UniRuntime(storage, new InMemoryOffsetStore(), 200L, 500L, 100, 500L);
+ runtime.withStorageConfig(new Properties());
+ try {
+ runtime.start();
+
+ // 3. Subscribe BEFORE publishing (memory poll is offset-index based; subscribing
+ // first means offset 0 covers the event regardless of pull-loop timing).
+ runtime.ingress().subscribe(topic, clientId, DistributionMode.BROADCAST, null);
+ CloudEvent event = CloudEventBuilder.v1()
+ .withId("it-mem-1").withSource(URI.create("it")).withType("it.memory.event").build();
+ runtime.ingress().publish(topic, event).get(10, TimeUnit.SECONDS);
+
+ // 4. The background pull-loop dispatches into the client buffer.
+ List<BufferedEvent> received = new ArrayList<>();
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(15);
+ while (received.isEmpty() && System.nanoTime() < deadline) {
+ received.addAll(runtime.ingress().poll(clientId, 100, 500L));
+ }
+ assertFalse(received.isEmpty(), "event not delivered within 15s - check pull-loop over memory storage");
+
+ // 5. ACK advances the offset - the at-least-once contract, zero-dependency tier.
+ assertTrue(runtime.ingress().ack(received.get(0).getDeliveryId()),
+ "ack failed for delivery " + received.get(0).getDeliveryId());
+ long offset = maxAckedOffset(runtime, topic, clientId);
+ assertTrue(offset >= 1, "offset did not advance after ACK: " + offset);
+
+ // 6. Re-poll after full ACK is empty (no un-ACKed backlog left).
+ List<BufferedEvent> drained = new ArrayList<>();
+ long drainDeadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(3);
+ while (drained.isEmpty() && System.nanoTime() < drainDeadline) {
+ drained.addAll(runtime.ingress().poll(clientId, 100, 200L));
+ }
+ assertEquals(0, drained.size(), "no residual events expected after ACK");
+ } finally {
+ runtime.shutdown();
+ }
+ }
+
+ /**
+ * Highest offset recorded for {@code clientId} across all partitions of {@code topic}, or -1
+ * if none - same helper contract as {@code RealBrokerIntegrationTest}.
+ */
+ private static long maxAckedOffset(UniRuntime runtime, String topic, String clientId) {
+ long max = -1L;
+ String prefix = clientId + "#";
+ for (java.util.Map.Entry<String, Long> e
+ : runtime.ingress().getOffsetStore().readAllOffsets(topic).entrySet()) {
+ if (e.getKey().startsWith(prefix)) {
+ max = Math.max(max, e.getValue());
+ }
+ }
+ return max;
+ }
+}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-memory/build.gradle b/eventmesh-storage-plugin/eventmesh-storage-memory/build.gradle
new file mode 100644
index 000000000..5880da344
--- /dev/null
+++ b/eventmesh-storage-plugin/eventmesh-storage-memory/build.gradle
@@ -0,0 +1,33 @@
+/*
+ * 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.
+ */
+
+dependencies {
+ implementation project(":eventmesh-storage-plugin:eventmesh-storage-api")
+ implementation project(":eventmesh-common")
+ implementation 'org.slf4j:slf4j-api'
+
+ testImplementation 'org.junit.jupiter:junit-jupiter'
+ testImplementation project(":eventmesh-storage-plugin:eventmesh-storage-api")
+ testImplementation project(":eventmesh-common")
+
+ compileOnly 'org.projectlombok:lombok'
+ compileOnly 'com.google.code.findbugs:jsr305:3.0.2'
+ annotationProcessor 'org.projectlombok:lombok'
+
+ testCompileOnly 'org.projectlombok:lombok'
+ testAnnotationProcessor 'org.projectlombok:lombok'
+}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-memory/gradle.properties b/eventmesh-storage-plugin/eventmesh-storage-memory/gradle.properties
new file mode 100644
index 000000000..575876414
--- /dev/null
+++ b/eventmesh-storage-plugin/eventmesh-storage-memory/gradle.properties
@@ -0,0 +1,19 @@
+# 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.
+#
+
+
+pluginType=storage
+pluginName=memory
\ No newline at end of file
diff --git a/eventmesh-storage-plugin/eventmesh-storage-memory/src/main/java/org/apache/eventmesh/storage/memory/MemoryMeshStoragePlugin.java b/eventmesh-storage-plugin/eventmesh-storage-memory/src/main/java/org/apache/eventmesh/storage/memory/MemoryMeshStoragePlugin.java
new file mode 100644
index 000000000..cb7071931
--- /dev/null
+++ b/eventmesh-storage-plugin/eventmesh-storage-memory/src/main/java/org/apache/eventmesh/storage/memory/MemoryMeshStoragePlugin.java
@@ -0,0 +1,259 @@
+/*
+ * 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.storage.memory;
+
+import org.apache.eventmesh.api.SendCallback;
+import org.apache.eventmesh.api.storage.MeshStoragePlugin;
+import org.apache.eventmesh.api.storage.StorageCapabilities;
+import org.apache.eventmesh.common.wire.EventMeshFrame;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * In-memory {@link MeshStoragePlugin} that implements <b>all 7</b>
+ * {@link StorageCapabilities} sub-interfaces.
+ *
+ * <p>A production-scope promotion of the TCK self-test fixture that lives in the
+ * {@code test} source set of {@code eventmesh-storage-api}. Intended for local dev
+ * sandboxes and CI smoke runs where provisioning an external broker (Kafka /
+ * RocketMQ) is undesirable — select it with {@code eventmesh.storage.type=memory}
+ * (env {@code EVENTMESH_STORAGE_TYPE=memory}).</p>
+ *
+ * <p><b>Not for production.</b> State is process-local and lost on shutdown;
+ * there is no persistence of any kind.</p>
+ *
+ * <h2>State</h2>
+ * <ul>
+ * <li>{@code topics} — per-topic list of appended frames (WAL).</li>
+ * <li>{@code offsets} — per-{@code (topic, partition)} committed offset.</li>
+ * <li>{@code partitions} — per-topic set of assigned partitions.</li>
+ * <li>{@code pendingAcks} — per-{@code (topic, ackKey)} pending POP ACK callback.</li>
+ * <li>{@code liteTopics} — per-{@code (parent, lite)} list of frames.</li>
+ * </ul>
+ */
+public class MemoryMeshStoragePlugin
+ implements MeshStoragePlugin,
+ StorageCapabilities.TopicManagement,
+ StorageCapabilities.PartitionAssignment,
+ StorageCapabilities.ExplicitOffsetCommit,
+ StorageCapabilities.EndOffsetQuery,
+ StorageCapabilities.AlignPullOffset,
+ StorageCapabilities.DeferredPopAck,
+ StorageCapabilities.LiteTopic {
+
+ private final ConcurrentMap<String, List<EventMeshFrame>> topics = new ConcurrentHashMap<>();
+ private final ConcurrentMap<String, Long> offsets = new ConcurrentHashMap<>();
+ private final ConcurrentMap<String, List<Integer>> partitions = new ConcurrentHashMap<>();
+ private final ConcurrentMap<String, Runnable> pendingAcks = new ConcurrentHashMap<>();
+ private final ConcurrentMap<String, List<EventMeshFrame>> liteTopics = new ConcurrentHashMap<>();
+
+ // Per-(topic, partition) pull cursor. startOffset < 0 means "self-resume from the
+ // cursor" (the runtime pull-loop contract); alignPullOffset rewinds it on ACK-timeout
+ // redelivery, giving at-least-once without re-scanning the whole log every tick.
+ private final ConcurrentMap<String, AtomicLong> pullCursors = new ConcurrentHashMap<>();
+ private final AtomicLong litePullCursors = new AtomicLong(0);
+
+ private volatile boolean started = false;
+ private volatile boolean closed = false;
+ private volatile int defaultLiteQueueCount = 4;
+
+ @Override
+ public void init(Properties properties) {
+ // No-op: state is in-memory. If a caller sets "inmem.liteQueueCount" we honor it.
+ if (properties != null && properties.getProperty("inmem.liteQueueCount") != null) {
+ try {
+ defaultLiteQueueCount = Integer.parseInt(properties.getProperty("inmem.liteQueueCount"));
+ } catch (NumberFormatException ignored) {
+ // keep default
+ }
+ }
+ }
+
+ @Override
+ public void send(String topic, EventMeshFrame frame, SendCallback callback) {
+ if (closed) {
+ throw new IllegalStateException("plugin is closed");
+ }
+ topics.computeIfAbsent(topic, k -> Collections.synchronizedList(new ArrayList<>())).add(frame);
+ if (callback != null) {
+ callback.onSuccess(new org.apache.eventmesh.api.SendResult());
+ }
+ }
+
+ @Override
+ public List<EventMeshFrame> poll(String topic, int partition, long startOffset, int maxEvents, long timeoutMs) {
+ List<EventMeshFrame> topicFrames = topics.get(topic);
+ if (topicFrames == null) {
+ return Collections.emptyList();
+ }
+ synchronized (topicFrames) {
+ AtomicLong cursor = pullCursors.computeIfAbsent(key(topic, partition), k -> new AtomicLong());
+ int from = startOffset < 0 ? (int) Math.min(cursor.get(), Integer.MAX_VALUE) : (int) startOffset;
+ if (from >= topicFrames.size()) {
+ return Collections.emptyList();
+ }
+ int to = Math.min(from + maxEvents, topicFrames.size());
+ cursor.set(to);
+ return new ArrayList<>(topicFrames.subList(from, to));
+ }
+ }
+
+ @Override
+ public void assignPartitions(String topic, List<Integer> ps) {
+ partitions.put(topic, new ArrayList<>(ps));
+ }
+
+ @Override
+ public void commitOffset(String topic, int partition, long offset) {
+ offsets.put(key(topic, partition), offset);
+ }
+
+ @Override
+ public int partitionCount(String topic) {
+ List<Integer> ps = partitions.get(topic);
+ return ps == null ? -1 : ps.size();
+ }
+
+ @Override
+ public long endOffset(String topic, int partition) {
+ List<EventMeshFrame> frames = topics.get(topic);
+ if (frames == null) {
+ return -1L;
+ }
+ synchronized (frames) {
+ return frames.size();
+ }
+ }
+
+ @Override
+ public boolean alignPullOffset(String topic, int partition, long ackOffset) {
+ if (ackOffset < 0) {
+ return false;
+ }
+ // Rewind the per-(topic, partition) pull cursor so the next poll re-delivers from
+ // ackOffset — the redelivery path for un-ACKed events (at-least-once).
+ pullCursors.computeIfAbsent(key(topic, partition), k -> new AtomicLong()).set(ackOffset);
+ return true;
+ }
+
+ @Override
+ public boolean ackPulledMessage(String topic, String ackKey) {
+ Runnable r = pendingAcks.remove(ackKey);
+ if (r == null) {
+ return false;
+ }
+ r.run();
+ return true;
+ }
+
+ @Override
+ public void createTopic(String topic, int partitions) {
+ topics.computeIfAbsent(topic, k -> Collections.synchronizedList(new ArrayList<>()));
+ }
+
+ // ============================== LiteTopic ==============================
+
+ @Override
+ public void createLiteTopic(String parentTopic, String liteTopic) {
+ createLiteTopic(parentTopic, liteTopic, defaultLiteQueueCount);
+ }
+
+ @Override
+ public void createLiteTopic(String parentTopic, String liteTopic, int queueCount) {
+ String key = liteKey(parentTopic, liteTopic);
+ liteTopics.computeIfAbsent(key, k -> Collections.synchronizedList(new ArrayList<>()));
+ }
+
+ @Override
+ public void sendLite(String parentTopic, String liteTopic, EventMeshFrame frame, SendCallback callback) {
+ if (closed) {
+ throw new IllegalStateException("plugin is closed");
+ }
+ String key = liteKey(parentTopic, liteTopic);
+ liteTopics.computeIfAbsent(key, k -> Collections.synchronizedList(new ArrayList<>())).add(frame);
+ // Also register a pending POP ACK so ackPulledMessage can demonstrate true.
+ String ackKey = "ack-" + key + "-" + System.nanoTime();
+ pendingAcks.put(ackKey, () -> { /* no-op for in-memory */ });
+ if (callback != null) {
+ callback.onSuccess(new org.apache.eventmesh.api.SendResult());
+ }
+ }
+
+ @Override
+ public List<EventMeshFrame> pullLite(String parentTopic, String liteTopic, int maxEvents, long timeoutMs) {
+ String key = liteKey(parentTopic, liteTopic);
+ List<EventMeshFrame> frames = liteTopics.get(key);
+ if (frames == null) {
+ return Collections.emptyList();
+ }
+ synchronized (frames) {
+ int from = (int) Math.min(litePullCursors.get(), frames.size());
+ int to = Math.min(from + maxEvents, frames.size());
+ if (from >= to) {
+ return Collections.emptyList();
+ }
+ return new ArrayList<>(frames.subList(from, to));
+ }
+ }
+
+ // ============================== LifeCycle ==============================
+
+ @Override
+ public boolean isStarted() {
+ return started;
+ }
+
+ @Override
+ public boolean isClosed() {
+ return closed;
+ }
+
+ @Override
+ public void start() {
+ started = true;
+ }
+
+ @Override
+ public void shutdown() {
+ closed = true;
+ }
+
+ // ============================== helpers ==============================
+
+ private static String key(String topic, int partition) {
+ return topic + "#" + partition;
+ }
+
+ private static String liteKey(String parent, String lite) {
+ return parent + "$" + lite;
+ }
+
+ /**
+ * Test-only accessor for verifying round-trip behavior (send then poll returns the frame).
+ */
+ public Map<String, List<EventMeshFrame>> debugTopics() {
+ return topics;
+ }
+}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-memory/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.storage.MeshStoragePlugin b/eventmesh-storage-plugin/eventmesh-storage-memory/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.storage.MeshStoragePlugin
new file mode 100644
index 000000000..63090f591
--- /dev/null
+++ b/eventmesh-storage-plugin/eventmesh-storage-memory/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.storage.MeshStoragePlugin
@@ -0,0 +1,16 @@
+# 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.
+
+memory=org.apache.eventmesh.storage.memory.MemoryMeshStoragePlugin
diff --git a/eventmesh-storage-plugin/eventmesh-storage-memory/src/test/java/org/apache/eventmesh/storage/memory/MemoryMeshStoragePluginTCKTest.java b/eventmesh-storage-plugin/eventmesh-storage-memory/src/test/java/org/apache/eventmesh/storage/memory/MemoryMeshStoragePluginTCKTest.java
new file mode 100644
index 000000000..d1b148d36
--- /dev/null
+++ b/eventmesh-storage-plugin/eventmesh-storage-memory/src/test/java/org/apache/eventmesh/storage/memory/MemoryMeshStoragePluginTCKTest.java
@@ -0,0 +1,94 @@
+/*
+ * 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.storage.memory;
+
+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.api.storage.StorageCapabilities;
+import org.apache.eventmesh.api.storage.tck.MeshStoragePluginTCK;
+import org.apache.eventmesh.common.wire.EventMeshFrame;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+
+import org.junit.jupiter.api.Test;
+
+/**
+ * TCK run for {@link MemoryMeshStoragePlugin}. Extends the abstract
+ * {@link MeshStoragePluginTCK} and declares all 7 capabilities.
+ */
+public class MemoryMeshStoragePluginTCKTest extends MeshStoragePluginTCK<MemoryMeshStoragePlugin> {
+
+ @Override
+ protected MemoryMeshStoragePlugin newPlugin() {
+ return new MemoryMeshStoragePlugin();
+ }
+
+ @Override
+ protected Set<Class<?>> expectedCapabilities() {
+ return Set.of(
+ StorageCapabilities.TopicManagement.class,
+ StorageCapabilities.PartitionAssignment.class,
+ StorageCapabilities.ExplicitOffsetCommit.class,
+ StorageCapabilities.EndOffsetQuery.class,
+ StorageCapabilities.AlignPullOffset.class,
+ StorageCapabilities.DeferredPopAck.class,
+ StorageCapabilities.LiteTopic.class
+ );
+ }
+
+ @Test
+ void sendThenPollRoundTrips() {
+ plugin.init(new Properties());
+ plugin.createTopic("round-trip", 1);
+ EventMeshFrame frame = EventMeshFrame.event(Map.of(), "hello".getBytes());
+ plugin.send("round-trip", frame, null);
+ List<EventMeshFrame> polled = plugin.poll("round-trip", 0, 0L, 10, 0L);
+ assertEquals(1, polled.size());
+ }
+
+ @Test
+ void commitOffsetRoundTrips() {
+ plugin.init(new Properties());
+ plugin.createTopic("offsets", 1);
+ plugin.commitOffset("offsets", 0, 42L);
+ assertTrue(plugin.endOffset("offsets", 0) >= 0L);
+ }
+
+ @Test
+ void endOffsetUnknownTopicIsMinusOne() {
+ plugin.init(new Properties());
+ assertEquals(-1L, plugin.endOffset("no-such-topic", 0));
+ }
+
+ @Test
+ void alignPullOffsetRejectsNegativeAckOffset() {
+ plugin.init(new Properties());
+ assertFalse(plugin.alignPullOffset("any", 0, -1L));
+ }
+
+ @Test
+ void ackPulledMessageUnknownKeyIsFalse() {
+ plugin.init(new Properties());
+ assertFalse(plugin.ackPulledMessage("any", "no-such-ack"));
+ }
+}
diff --git a/settings.gradle b/settings.gradle
index 9dbe8273a..8933236a1 100644
--- a/settings.gradle
+++ b/settings.gradle
@@ -55,6 +55,7 @@ include 'eventmesh-sdks:eventmesh-sdk-java'
// Storage 后端 (新架构适配了 kafka/rocketmq)
include 'eventmesh-storage-plugin:eventmesh-storage-api'
include 'eventmesh-storage-plugin:eventmesh-storage-kafka'
+include 'eventmesh-storage-plugin:eventmesh-storage-memory'
include 'eventmesh-storage-plugin:eventmesh-storage-rocketmq'
include 'eventmesh-storage-plugin:eventmesh-storage-rocketmq5'
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: commits-help@eventmesh.apache.org