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:20:07 UTC
(eventmesh) branch fix/5364-chaos-upgrade-admin created (now 596ba0b55)
This is an automated email from the ASF dual-hosted git repository.
qqeasonchen pushed a change to branch fix/5364-chaos-upgrade-admin
in repository https://gitbox.apache.org/repos/asf/eventmesh.git
at 596ba0b55 fix(#5364): admin token guard (fail-closed) + chaos in-process cases
This branch includes the following new commits:
new 596ba0b55 fix(#5364): admin token guard (fail-closed) + chaos in-process cases
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: commits-help@eventmesh.apache.org
(eventmesh) 01/01: fix(#5364): admin token guard (fail-closed) + chaos in-process cases
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
qqeasonchen pushed a commit to branch fix/5364-chaos-upgrade-admin
in repository https://gitbox.apache.org/repos/asf/eventmesh.git
commit 596ba0b557a45828b6bbd0f2fc12a59611e2cc05
Author: Eason Chen <qq...@gmail.com>
AuthorDate: Wed Sep 9 13:19:55 2026 +0800
fix(#5364): admin token guard (fail-closed) + chaos in-process cases
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).
Co-authored-by: qqeasonchen <qq...@gmail.com>
---
.../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