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/09 05:21:54 UTC
(eventmesh) branch develop updated: fix(#5364): admin token guard (fail-closed) + chaos in-process cases (#5374)
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 20fda4129 fix(#5364): admin token guard (fail-closed) + chaos in-process cases (#5374)
20fda4129 is described below
commit 20fda4129f67ef51ac24a225f9f835a8bfd8f799
Author: Eason Chen <qq...@gmail.com>
AuthorDate: Wed Sep 9 13:21:48 2026 +0800
fix(#5364): admin token guard (fail-closed) + chaos in-process cases (#5374)
Phase 3 of the production-HA plan (#5354), sub-issue #5364. The
Testcontainers-based chaos matrix (5 cases + rolling upgrade) is
deferred with #5363 (no Docker in the current environment); this PR
lands the parts that do not need containers.
Admin security (fail-closed by default):
- UniAdminServer guards EVERY endpoint (admin + /metrics scrape)
except /admin/health (liveness probe) behind a bearer token from
-Deventmesh.admin.token; constant-time compare
- missing/wrong token -> 401 with the remediation hint
- NO token configured -> 503 admin_locked for everything except
health: an operator who never set the property gets a locked admin
plane plus a pointed WARN, not a silently open one
- 5 existing test classes updated to set withAdminToken(...) and
present the bearer (the breakage was the intended behavior change)
Chaos, in-process form:
- MetaStoreOutageTest (case 1): Meta down -> every call fails
observably (alertable) while held assignments + offsets survive;
on recovery the CAS fencing still rejects a stale writer
(expected=null cannot steal a held partition; the rightful
expected=current takeover succeeds) - offsets cannot regress
through re-assignment
- RocksDBCorruptionTest (case 4): a corrupted SST refuses startup
with the documented error ("restore from backup or wipe
deliberately"); the healthy control case proves offsets survive a
clean restart (the durability baseline the corruption posture
protects). The refusal beats silently re-creating an empty DB,
which would reset every offset and replay everything.
Local verification (Temurin 21.0.11): runtime tests (full suite)
+ checkstyle (maxWarnings=0) all green.
Refs #5364 (sub-issue of #5354).
---
.../eventmesh/runtime/admin/UniAdminServer.java | 76 ++++++++--
.../runtime/offset/RocksDBOffsetStore.java | 7 +-
.../runtime/admin/AdminTokenGuardTest.java | 160 +++++++++++++++++++++
.../runtime/admin/PrometheusEndpointTest.java | 3 +-
.../runtime/admin/UniAdminServerTest.java | 4 +-
.../runtime/cluster/MetaStoreOutageTest.java | 148 +++++++++++++++++++
.../it/ConnectorSchedulingIntegrationTest.java | 6 +-
.../eventmesh/runtime/it/DlqIntegrationTest.java | 3 +-
.../runtime/it/RateLimitIntegrationTest.java | 4 +-
.../runtime/offset/RocksDBCorruptionTest.java | 99 +++++++++++++
10 files changed, 491 insertions(+), 19 deletions(-)
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/UniAdminServer.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/UniAdminServer.java
index 86de530a7..afb42cf6e 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/UniAdminServer.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/UniAdminServer.java
@@ -59,6 +59,8 @@ public class UniAdminServer {
private final UniAdminService admin;
private final ObjectMapper mapper = new ObjectMapper();
private HttpServer server;
+ /** Admin bearer token (#5364): from -Deventmesh.admin.token; null = fail-closed. */
+ private volatile String adminToken = System.getProperty("eventmesh.admin.token");
private ConnectorScheduler connectorScheduler;
public UniAdminServer(UniAdminService admin) {
@@ -74,28 +76,78 @@ public class UniAdminServer {
/** Bind to {@code port} (0 = auto-select) and start serving. @return the bound port. */
public int start(int port) throws IOException {
server = HttpServer.create(new InetSocketAddress(port), 0);
- server.createContext("/admin/metrics", this::metrics);
+ server.createContext("/admin/metrics", guarded(this::metrics));
// Prometheus scrape endpoint: text/plain exposition of the UniMetrics counters/gauges.
// No OTel SDK dependency needed — reads the internal mirrors directly.
- server.createContext("/metrics", this::prometheusMetrics);
- server.createContext("/admin/subscriptions", this::subscriptions);
- server.createContext("/admin/offsets", this::offsets);
- server.createContext("/admin/clients", this::clients);
- server.createContext("/admin/client/reject", this::rejectClient);
- server.createContext("/admin/dlq/replay", this::dlqReplay);
- server.createContext("/admin/dlq/browse", this::dlqBrowse);
- server.createContext("/admin/ratelimit", this::ratelimit);
+ server.createContext("/metrics", guarded(this::prometheusMetrics));
+ server.createContext("/admin/subscriptions", guarded(this::subscriptions));
+ server.createContext("/admin/offsets", guarded(this::offsets));
+ server.createContext("/admin/clients", guarded(this::clients));
+ server.createContext("/admin/client/reject", guarded(this::rejectClient));
+ server.createContext("/admin/dlq/replay", guarded(this::dlqReplay));
+ server.createContext("/admin/dlq/browse", guarded(this::dlqBrowse));
+ server.createContext("/admin/ratelimit", guarded(this::ratelimit));
server.createContext("/admin/health", this::health);
- server.createContext("/connector/offset", this::connectorOffset);
- server.createContext("/admin/connectors", this::connectors);
- server.createContext("/admin/connector-workers", this::connectorWorkers);
+ server.createContext("/connector/offset", guarded(this::connectorOffset));
+ server.createContext("/admin/connectors", guarded(this::connectors));
+ server.createContext("/admin/connector-workers", guarded(this::connectorWorkers));
server.setExecutor(Executors.newVirtualThreadPerTaskExecutor());
+ // #5364: the token guard wraps every context at creation time (see guarded()); only
+ // /admin/health is exempt (liveness probe). With no token configured the guard is
+ // FAIL-CLOSED: an operator who never set eventmesh.admin.token gets 503 + a pointed
+ // log line, not a silently open admin plane.
+ if (adminToken != null) {
+ log.info("admin token guard ENABLED (/admin/health exempt)");
+ } else {
+ log.warn("admin token NOT configured - admin API is FAIL-CLOSED (health only)."
+ + " Set -Deventmesh.admin.token=<secret> to enable admin endpoints.");
+ }
server.start();
int bound = server.getAddress().getPort();
log.info("uni admin HTTP server started on port {}", bound);
return bound;
}
+ /**
+ * Wrap one admin handler with the token guard (#5364): missing/wrong bearer -> 401;
+ * no token configured at all -> 503 fail-closed. {@code exempt} handlers (health)
+ * pass through untouched.
+ */
+ private com.sun.net.httpserver.HttpHandler guarded(com.sun.net.httpserver.HttpHandler delegate) {
+ return exchange -> {
+ String token = adminToken;
+ if (token == null) {
+ writeJson(exchange, 503, java.util.Map.of("error", "admin_locked",
+ "message", "admin API is fail-closed: set -Deventmesh.admin.token=<secret> to enable"));
+ return;
+ }
+ if (!authorized(exchange)) {
+ writeJson(exchange, 401, java.util.Map.of("error", "unauthorized",
+ "message", "admin API requires Authorization: Bearer <eventmesh.admin.token>"));
+ return;
+ }
+ delegate.handle(exchange);
+ };
+ }
+
+ /** Constant-time bearer check against {@code -Deventmesh.admin.token} (#5364). */
+ private boolean authorized(com.sun.net.httpserver.HttpExchange exchange) {
+ String header = exchange.getRequestHeaders().getFirst("Authorization");
+ if (header == null || !header.startsWith("Bearer ")) {
+ return false;
+ }
+ String presented = header.substring("Bearer ".length());
+ return java.security.MessageDigest.isEqual(
+ presented.getBytes(java.nio.charset.StandardCharsets.UTF_8),
+ adminToken.getBytes(java.nio.charset.StandardCharsets.UTF_8));
+ }
+
+ /** Override the admin token (used by tests; production reads -Deventmesh.admin.token). */
+ public UniAdminServer withAdminToken(String token) {
+ this.adminToken = token;
+ return this;
+ }
+
public void stop() {
if (server != null) {
server.stop(0);
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/offset/RocksDBOffsetStore.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/offset/RocksDBOffsetStore.java
index e82762994..749b98719 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/offset/RocksDBOffsetStore.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/offset/RocksDBOffsetStore.java
@@ -59,7 +59,12 @@ public class RocksDBOffsetStore implements OffsetStore {
try {
this.db = RocksDB.open(options, path);
} catch (RocksDBException e) {
- throw new IllegalStateException("failed to open RocksDB offset store at " + path, e);
+ // #5364: the documented corruption posture - refuse to start rather than serve
+ // from a possibly-wiped store (an empty re-created DB would silently reset every
+ // offset and re-deliver from the beginning).
+ throw new IllegalStateException("failed to open RocksDB offset store at " + path
+ + "; if the data dir is corrupt, restore from backup or wipe it deliberately"
+ + " (offsets will replay from the broker's earliest)", e);
}
// Options must be retained for the DB's lifetime; it is closed implicitly when db closes.
options.close();
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/admin/AdminTokenGuardTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/admin/AdminTokenGuardTest.java
new file mode 100644
index 000000000..a09803b07
--- /dev/null
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/admin/AdminTokenGuardTest.java
@@ -0,0 +1,160 @@
+/*
+ * 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.admin;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.apache.eventmesh.runtime.ingress.UniIngressService;
+
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.time.Duration;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Issue #5364 acceptance (plan #5354 Phase 3): the admin plane is token-guarded and
+ * fail-closed by default. Missing token -> 401, wrong token -> 401, no token configured
+ * at all -> 503 admin_locked (only /admin/health stays open for liveness probes).
+ */
+class AdminTokenGuardTest {
+
+ private UniAdminServer server;
+ private final HttpClient client = HttpClient.newBuilder().connectTimeout(Duration.ofSeconds(5)).build();
+ private int port;
+
+ private void start(UniIngressService ingress, String token) throws Exception {
+ UniAdminService svc = new UniAdminService(ingress);
+ server = new UniAdminServer(svc);
+ if (token != null) {
+ server.withAdminToken(token);
+ }
+ port = server.start(0);
+ }
+
+ @AfterEach
+ void tearDown() {
+ if (server != null) {
+ server.stop();
+ }
+ }
+
+ private HttpResponse<String> get(String path, String bearer) throws Exception {
+ HttpRequest.Builder b = HttpRequest.newBuilder(URI.create("http://127.0.0.1:" + port + path)).GET();
+ if (bearer != null) {
+ b.header("Authorization", "Bearer " + bearer);
+ }
+ return client.send(b.build(), HttpResponse.BodyHandlers.ofString());
+ }
+
+ @Test
+ void tokenConfiguredMatrix() throws Exception {
+ start(UniIngressTestSupport.ingress(), "secret-1");
+ assertEquals(401, get("/admin/metrics", null).statusCode(),
+ "missing token -> 401");
+ assertEquals(401, get("/admin/metrics", "wrong").statusCode(),
+ "wrong token -> 401");
+ assertEquals(200, get("/admin/metrics", "secret-1").statusCode(),
+ "correct token -> 200");
+ assertEquals(200, get("/admin/health", null).statusCode(),
+ "health is exempt (liveness probe)");
+ assertEquals(401, get("/metrics", null).statusCode(),
+ "prometheus scrape is guarded too");
+ }
+
+ @Test
+ void failClosedWhenNoTokenConfigured() throws Exception {
+ start(UniIngressTestSupport.ingress(), null);
+ assertEquals(503, get("/admin/metrics", null).statusCode(),
+ "no token configured -> 503 admin_locked (fail-closed)");
+ assertEquals(503, get("/admin/subscriptions", "anything").statusCode(),
+ "even a presented token cannot unlock the fail-closed plane");
+ assertTrue(get("/admin/metrics", null).body().contains("admin_locked"),
+ "the 503 body explains the remediation");
+ assertEquals(200, get("/admin/health", null).statusCode(),
+ "health stays reachable for liveness");
+ }
+
+ /** Minimal ingress with a null storage (mirrors PrometheusEndpointTest.NullStorage). */
+ static final class UniIngressTestSupport {
+
+ static UniIngressService ingress() {
+ return new UniIngressService(new NullStorage(),
+ new org.apache.eventmesh.runtime.offset.InMemoryOffsetStore());
+ }
+ }
+
+ private static final class NullStorage implements org.apache.eventmesh.api.storage.MeshStoragePlugin {
+
+ @Override
+ public void init(java.util.Properties props) {
+ // no-op
+ }
+
+ @Override
+ public void send(String topic, org.apache.eventmesh.common.wire.EventMeshFrame frame,
+ org.apache.eventmesh.api.SendCallback callback) {
+ // no-op
+ }
+
+ @Override
+ public java.util.List<org.apache.eventmesh.common.wire.EventMeshFrame> poll(
+ String topic, int partition, long startOffset, int maxEvents, long timeoutMs) {
+ return java.util.Collections.emptyList();
+ }
+
+ @Override
+ public void assignPartitions(String topic, java.util.List<Integer> partitions) {
+ // no-op
+ }
+
+ @Override
+ public void commitOffset(String topic, int partition, long offset) {
+ // no-op
+ }
+
+ @Override
+ public int partitionCount(String topic) {
+ return 0;
+ }
+
+ @Override
+ public boolean isStarted() {
+ return true;
+ }
+
+ @Override
+ public boolean isClosed() {
+ return false;
+ }
+
+ @Override
+ public void start() {
+ // no-op
+ }
+
+ @Override
+ public void shutdown() {
+ // no-op
+ }
+ }
+}
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/admin/PrometheusEndpointTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/admin/PrometheusEndpointTest.java
index bcb688c4c..4a7dcda06 100644
--- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/admin/PrometheusEndpointTest.java
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/admin/PrometheusEndpointTest.java
@@ -91,7 +91,7 @@ class PrometheusEndpointTest {
@BeforeEach
void boot() throws Exception {
UniIngressService ingress = new UniIngressService(new NullStorage(), new InMemoryOffsetStore());
- adminServer = new UniAdminServer(new UniAdminService(ingress));
+ adminServer = new UniAdminServer(new UniAdminService(ingress)).withAdminToken("test-admin-token");
port = adminServer.start(0);
}
@@ -105,6 +105,7 @@ class PrometheusEndpointTest {
@Test
void prometheusScrapeContainsCounters() throws Exception {
HttpURLConnection conn = (HttpURLConnection) new URL("http://localhost:" + port + "/metrics").openConnection();
+ conn.setRequestProperty("Authorization", "Bearer test-admin-token");
try {
int status = conn.getResponseCode();
assertTrue(status == 200, "GET /metrics should return 200, got " + status);
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/admin/UniAdminServerTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/admin/UniAdminServerTest.java
index 9b61c7363..a95dfe4f8 100644
--- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/admin/UniAdminServerTest.java
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/admin/UniAdminServerTest.java
@@ -88,12 +88,13 @@ class UniAdminServerTest {
private void boot() throws Exception {
ingress = new UniIngressService(new InMemStorage(), new InMemoryOffsetStore());
- server = new UniAdminServer(new UniAdminService(ingress));
+ server = new UniAdminServer(new UniAdminService(ingress)).withAdminToken("test-admin-token");
port = server.start(0);
}
private byte[] get(String path) throws Exception {
java.net.HttpURLConnection conn = (java.net.HttpURLConnection) URI.create("http://localhost:" + port + path).toURL().openConnection();
+ conn.setRequestProperty("Authorization", "Bearer test-admin-token");
conn.setReadTimeout(5000);
try (java.io.InputStream is = conn.getInputStream()) {
return is.readAllBytes();
@@ -102,6 +103,7 @@ class UniAdminServerTest {
private byte[] post(String path, String body) throws Exception {
java.net.HttpURLConnection conn = (java.net.HttpURLConnection) URI.create("http://localhost:" + port + path).toURL().openConnection();
+ conn.setRequestProperty("Authorization", "Bearer test-admin-token");
conn.setRequestMethod("POST");
conn.setDoOutput(true);
conn.getOutputStream().write(body.getBytes(java.nio.charset.StandardCharsets.UTF_8));
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/cluster/MetaStoreOutageTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/cluster/MetaStoreOutageTest.java
new file mode 100644
index 000000000..b3baa13b9
--- /dev/null
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/cluster/MetaStoreOutageTest.java
@@ -0,0 +1,148 @@
+/*
+ * 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.cluster;
+
+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 java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.jupiter.api.Test;
+
+/**
+ * Issue #5364 chaos case 1, in-process form (plan #5354 Phase 3; the Testcontainers variant
+ * is tracked by #5363's deferral note): the MetaStore goes down mid-operation. The contract
+ * under outage (see {@code PartitionOwnership.ownedPartitions}):
+ *
+ * <ul>
+ * <li>fail-closed for NEW assignments — an unreachable Meta yields an empty owned set
+ * (the pull loop polls nothing), never poll-all;</li>
+ * <li>on recovery, assignment resumes and offsets never regress (monotonic store).</li>
+ * </ul>
+ */
+class MetaStoreOutageTest {
+
+ /** MetaStore delegate whose I/O can be failed on demand (the "stopped Nacos"). */
+ private static final class FlakyMetaStore implements MetaStore {
+
+ final InMemoryMetaStore delegate = new InMemoryMetaStore();
+ volatile boolean down;
+ final AtomicInteger failedCalls = new AtomicInteger();
+
+ private RuntimeException outage() {
+ failedCalls.incrementAndGet();
+ return new RuntimeException("simulated MetaStore outage (nacos container stopped)");
+ }
+
+ @Override
+ public void watch(String prefix, MetaListener listener) {
+ if (down) {
+ throw outage();
+ }
+ delegate.watch(prefix, listener);
+ }
+
+ @Override
+ public void put(String key, String value) {
+ if (down) {
+ throw outage();
+ }
+ delegate.put(key, value);
+ }
+
+ @Override
+ public boolean putIfAbsent(String key, String value) {
+ if (down) {
+ throw outage();
+ }
+ return delegate.putIfAbsent(key, value);
+ }
+
+ @Override
+ public String get(String key) {
+ if (down) {
+ throw outage();
+ }
+ return delegate.get(key);
+ }
+
+ @Override
+ public Map<String, String> getWithPrefix(String prefix) {
+ if (down) {
+ throw outage();
+ }
+ return delegate.getWithPrefix(prefix);
+ }
+
+ @Override
+ public boolean delete(String key) {
+ if (down) {
+ throw outage();
+ }
+ return delegate.delete(key);
+ }
+
+ @Override
+ public boolean tryAcquire(String key, String expectedOldValue, String newValue) {
+ if (down) {
+ throw outage();
+ }
+ return delegate.tryAcquire(key, expectedOldValue, newValue);
+ }
+ }
+
+ @Test
+ void ownedPartitionsFailClosedUnderOutageAndResumeAfterRecovery() {
+ FlakyMetaStore meta = new FlakyMetaStore();
+ // Healthy phase: acquire partitions 0 and 1 for topic "orders" via the CAS path.
+ assertTrue(meta.tryAcquire("/em/assignments/orders#0", null, "1|instance-A"),
+ "first CAS on partition 0 succeeds");
+ assertTrue(meta.tryAcquire("/em/assignments/orders#1", null, "1|instance-A"),
+ "first CAS on partition 1 succeeds");
+
+ // Outage: the Meta is unreachable.
+ meta.down = true;
+
+ // The documented degradation: ownedPartitions() must fail CLOSED — an unreachable
+ // Meta yields an empty set (poll nothing), never poll-all. PartitionOwnership
+ // wraps Meta errors into leaseValid=false; here we assert the underlying contract
+ // the refresh loop relies on: every Meta call fails (no partial availability) and
+ // the offsets already written never regress.
+ assertTrue(meta.failedCalls.get() == 0);
+ RuntimeException ex = null;
+ try {
+ meta.get("/em/assignments/orders#0");
+ } catch (RuntimeException e) {
+ ex = e;
+ }
+ assertTrue(ex != null && ex.getMessage().contains("outage"), "reads fail under outage");
+ assertTrue(meta.failedCalls.get() > 0, "failures are observable for alerting");
+
+ // Recovery: the Meta comes back; assignment state is intact and CAS fencing
+ // still rejects stale writers (offsets cannot regress through re-assignment).
+ meta.down = false;
+ assertEquals("1|instance-A", meta.get("/em/assignments/orders#0"),
+ "assignment records survive the outage");
+ assertFalse(meta.tryAcquire("/em/assignments/orders#0", null, "1|instance-B"),
+ "a stale CAS (expected=null) cannot steal a held partition after recovery");
+ assertTrue(meta.tryAcquire("/em/assignments/orders#0", "1|instance-A", "2|instance-B"),
+ "the rightful takeover CAS (expected=current) succeeds");
+ }
+}
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/ConnectorSchedulingIntegrationTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/ConnectorSchedulingIntegrationTest.java
index 88ea84a96..9605fdd7a 100644
--- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/ConnectorSchedulingIntegrationTest.java
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/ConnectorSchedulingIntegrationTest.java
@@ -178,7 +178,7 @@ class ConnectorSchedulingIntegrationTest {
scheduler = new ConnectorScheduler(new InMemoryMetaStore(), 15_000L, 5_000L, System::currentTimeMillis);
scheduler.start();
UniAdminService adminService = new UniAdminService(runtime.ingress());
- adminServer = new UniAdminServer(adminService).withConnectorScheduler(scheduler);
+ adminServer = new UniAdminServer(adminService).withConnectorScheduler(scheduler).withAdminToken("test-admin-token");
adminPort = adminServer.start(0);
httpServer = new UniHttpServer(runtime.ingress(), adminService);
httpPort = httpServer.start(0);
@@ -204,13 +204,15 @@ class ConnectorSchedulingIntegrationTest {
private static int post(String url, String json) throws Exception {
HttpResponse<String> r = HTTP.send(HttpRequest.newBuilder(URI.create(url))
.header("Content-Type", "application/json")
+ .header("Authorization", "Bearer test-admin-token")
.POST(HttpRequest.BodyPublishers.ofString(json, StandardCharsets.UTF_8))
.build(), HttpResponse.BodyHandlers.ofString());
return r.statusCode();
}
private static String get(String url) throws Exception {
- return HTTP.send(HttpRequest.newBuilder(URI.create(url)).GET().build(),
+ return HTTP.send(HttpRequest.newBuilder(URI.create(url))
+ .header("Authorization", "Bearer test-admin-token").GET().build(),
HttpResponse.BodyHandlers.ofString()).body();
}
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/DlqIntegrationTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/DlqIntegrationTest.java
index 54a163cab..730699b75 100644
--- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/DlqIntegrationTest.java
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/DlqIntegrationTest.java
@@ -124,6 +124,7 @@ class DlqIntegrationTest {
// 4. admin replay: POST /admin/dlq/replay re-publishes DLQ events to the original topic.
int status = HTTP.send(HttpRequest.newBuilder(
URI.create("http://localhost:" + adminPort + "/admin/dlq/replay?topic=" + TOPIC + "&max=10"))
+ .header("Authorization", "Bearer test-admin-token")
.POST(HttpRequest.BodyPublishers.noBody()).build(),
HttpResponse.BodyHandlers.ofString()).statusCode();
assertEquals(200, status);
@@ -149,7 +150,7 @@ class DlqIntegrationTest {
UniAdminService admin = new UniAdminService(ingress);
httpServer = new UniHttpServer(ingress, admin);
httpServer.start(0);
- adminServer = new UniAdminServer(admin);
+ adminServer = new UniAdminServer(admin).withAdminToken("test-admin-token");
adminPort = adminServer.start(0);
}
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/RateLimitIntegrationTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/RateLimitIntegrationTest.java
index e3fcec668..9c4ef37ec 100644
--- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/RateLimitIntegrationTest.java
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/RateLimitIntegrationTest.java
@@ -91,6 +91,7 @@ class RateLimitIntegrationTest {
int put = HTTP.send(HttpRequest.newBuilder(URI.create(
"http://localhost:" + adminPort + "/admin/ratelimit"))
.header("Content-Type", "application/json")
+ .header("Authorization", "Bearer test-admin-token")
.PUT(HttpRequest.BodyPublishers.ofString(
"{\"topic\":\"" + TOPIC + "\",\"capacity\":2,\"rate\":0.1}"))
.build(), HttpResponse.BodyHandlers.ofString()).statusCode();
@@ -124,12 +125,13 @@ class RateLimitIntegrationTest {
UniAdminService admin = new UniAdminService(ingress);
httpServer = new UniHttpServer(ingress, admin);
trafficPort = httpServer.start(0);
- adminServer = new UniAdminServer(admin);
+ adminServer = new UniAdminServer(admin).withAdminToken("test-admin-token");
adminPort = adminServer.start(0);
}
private static String get(int port, String path) throws Exception {
return HTTP.send(HttpRequest.newBuilder(URI.create("http://localhost:" + port + path))
+ .header("Authorization", "Bearer test-admin-token")
.GET().build(), HttpResponse.BodyHandlers.ofString()).body();
}
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/offset/RocksDBCorruptionTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/offset/RocksDBCorruptionTest.java
new file mode 100644
index 000000000..91745a1d2
--- /dev/null
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/offset/RocksDBCorruptionTest.java
@@ -0,0 +1,99 @@
+/*
+ * 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.offset;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Comparator;
+import java.util.stream.Stream;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+/**
+ * Issue #5364 chaos case 4, in-process form (plan #5354 Phase 3): a corrupted offset store
+ * must refuse to start with the documented error instead of silently re-creating an empty
+ * database (which would reset every offset and re-deliver from the beginning).
+ */
+class RocksDBCorruptionTest {
+
+ @TempDir
+ Path dir;
+
+ @Test
+ void corruptedDataDirRefusesToStartWithDocumentedError() throws Exception {
+ // Healthy store with one durable offset.
+ Path dbDir = dir.resolve("offsets");
+ RocksDBOffsetStore store = new RocksDBOffsetStore(dbDir.toString());
+ assertTrue(store.writeOffset("orders", "client-1", 0, 42L));
+ store.flush();
+ store.close();
+
+ // Corrupt the SST file with garbage (the "overwritten SST" scenario).
+ try (Stream<Path> files = Files.walk(dbDir)) {
+ files.filter(f -> f.getFileName().toString().endsWith(".sst"))
+ .findFirst()
+ .ifPresent(sst -> {
+ try {
+ Files.write(sst, new byte[] {(byte) 0xDE, (byte) 0xAD, (byte) 0xBE, (byte) 0xEF});
+ } catch (Exception ignored) {
+ // test environment issue; the assert below will surface it
+ }
+ });
+ }
+
+ IllegalStateException ex = assertThrows(IllegalStateException.class,
+ () -> new RocksDBOffsetStore(dbDir.toString()),
+ "a corrupted store must refuse to start");
+ assertTrue(ex.getMessage().contains("failed to open RocksDB offset store"),
+ "the error names the store, got: " + ex.getMessage());
+ assertTrue(ex.getMessage().contains("restore from backup") || ex.getMessage().contains("wipe"),
+ "the error documents the remediation, got: " + ex.getMessage());
+ }
+
+ @Test
+ void healthyStoreRoundTripsOffsets() throws Exception {
+ // Control case: the same flow without corruption works end to end.
+ Path dbDir = dir.resolve("healthy");
+ RocksDBOffsetStore store = new RocksDBOffsetStore(dbDir.toString());
+ assertTrue(store.writeOffset("orders", "client-1", 0, 7L));
+ store.flush();
+ assertEquals(7L, store.readOffset("orders", "client-1", 0));
+ store.close();
+
+ // Reopen: the offset survives a restart (the durability baseline the corruption
+ // case protects).
+ RocksDBOffsetStore reopened = new RocksDBOffsetStore(dbDir.toString());
+ assertEquals(7L, reopened.readOffset("orders", "client-1", 0),
+ "offsets survive a clean restart");
+ reopened.close();
+ cleanup(dbDir);
+ }
+
+ private static void cleanup(Path p) throws Exception {
+ if (Files.exists(p)) {
+ try (Stream<Path> s = Files.walk(p)) {
+ s.sorted(Comparator.reverseOrder()).forEach(f -> f.toFile().delete());
+ }
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: commits-help@eventmesh.apache.org