diff --git a/build.gradle b/build.gradle
index 3a1cd55225..696d7ca61a 100644
--- a/build.gradle
+++ b/build.gradle
@@ -244,6 +244,15 @@ tasks.register('dist-connector') {
             from core.jar.archivePath
             into rootProject.file('dist-connector/apps')
         }
+        // The SPI jar (eventmesh-connector-api) is excluded from lib/ by the
+        // exclude 'eventmesh-*' below, but ConnectorApplication itself imports the
+        // SPI interfaces — ship it next to the runtime jar in apps/ so a bare
+        // runtime (no plugins installed) still boots.
+        def connectorApi = findProject('eventmesh-connector-api')
+        copy {
+            from connectorApi.jar.archivePath
+            into rootProject.file('dist-connector/apps')
+        }
         copy {
             from core.configurations.runtimeClasspath
             into rootProject.file('dist-connector/lib')
diff --git a/docs/contrib/connector-api-split-plan.md b/docs/contrib/connector-api-split-plan.md
new file mode 100644
index 0000000000..7243108dbe
--- /dev/null
+++ b/docs/contrib/connector-api-split-plan.md
@@ -0,0 +1,105 @@
+# Connector API split — design plan (P1)
+
+**Status:** Plan only. No code in this PR. Implementation should land in a follow-up PR after
+**issue #5305** (architecture guard) is in place to enforce the boundary.
+
+## Goal
+
+Move the connector SPI interfaces out of `eventmesh-connector-runtime` into a new
+`eventmesh-connector-api` module so plugins depend on a stable, minimal API jar and the runtime
+implements / orchestrates against that API.
+
+## Interfaces to move
+
+From `eventmesh-connector-runtime/src/main/java/org/apache/eventmesh/connector/`:
+
+- `SourceConnector`
+- `SinkConnector`
+- `EventMeshEndpoint`
+- `HttpCaller`
+- `ConnectorOffsetStore`
+- `CloudEventSerializer`
+- `PollEntry` (value type used by both sides)
+
+## Target layout
+
+```
+eventmesh-connector-api/
+  src/main/java/org/apache/eventmesh/connector/api/
+    SourceConnector.java
+    SinkConnector.java
+    EventMeshEndpoint.java
+    HttpCaller.java
+    ConnectorOffsetStore.java
+    CloudEventSerializer.java
+    PollEntry.java
+    package-info.java
+  build.gradle (deps: cloudevents-core only — no plugin imports, no HTTP libs)
+```
+
+`eventmesh-connector-runtime` depends on `:eventmesh-connector-api` and continues to provide
+implementations (`EventMeshHttpEndpoint`, `RocksDBConnectorOffsetStore`, `RemoteOffsetStore`,
+`InMemoryOffsetStore`, `ConnectorRuntime`, `ConnectorManager`, `ConnectorAdminServer`,
+`ConnectorApplication`, `ConnectorDef`).
+
+Each plugin under `eventmesh-connector-plugin/eventmesh-connector-*` should depend on
+`:eventmesh-connector-api` instead of `:eventmesh-connector-runtime`.
+
+## Plugin changes (mechanical)
+
+For each of the 23 plugins:
+
+1. `build.gradle`: replace `implementation project(":eventmesh-connector-runtime")` with
+   `implementation project(":eventmesh-connector-api")`.
+2. Source code: if the plugin imports `org.apache.eventmesh.connector.ConnectorRuntime` (it should
+   not — plugins only use the SPI), add `implementation project(":eventmesh-connector-runtime")`
+   back. Initial audit shows no plugin currently touches runtime internals.
+
+## ArchUnit enforcement (depends on #5305)
+
+Add a rule:
+
+```
+noClasses().that().resideInAPackage("..eventmesh.connector.plugin..")
+    .should().dependOnClassesThat().resideInAPackage("..eventmesh.connector.runtime..")
+    .because("plugins must depend only on the connector-api SPI, not on runtime internals")
+```
+
+This is the contract — once it passes, every plugin author who reaches into runtime internals
+will fail the architecture guard.
+
+## Migration order (sub-PRs)
+
+- **M1 — create the module + move 7 interfaces + move package-info.** Mechanical. Touches ~24
+  build.gradles but no production logic. No behaviour change.
+- **M2 — switch 23 plugins from runtime to api dependency.** Each plugin's `build.gradle` swap.
+  CI matrix must stay green; existing plugin tests are non-existent today (this PR adds them).
+- **M3 — add ArchUnit rule (depends on #5305 having `B` mode enabled — `rule.check()` fails the
+  build).** This is the enforcement moment. Without it the split is informational only.
+
+## Risks
+
+- **Sub-package collisions**: if any plugin imports `org.apache.eventmesh.connector.X` from
+  runtime, the import path will break. Audit by `git grep "org.apache.eventmesh.connector" -- '*/src/main/'`
+  before M1.
+- **Javadoc / package-info drift**: the current `package org.apache.eventmesh.connector;` (no
+  `.api`) means a lot of plugin code will see its `package-info.java` change. Acceptable — it's
+  API contract clarification.
+- **Build time**: 23 `build.gradle` edits are mechanical but the Gradle dependency graph will
+  shift; expect one or two of the ~30 modules to need a transitive adjustment.
+
+## Acceptance criteria for the implementation PR(s)
+
+- [ ] `eventmesh-connector-api` jar builds standalone (deps: cloudevents-core only).
+- [ ] `eventmesh-connector-runtime` depends on `:eventmesh-connector-api`.
+- [ ] All 23 plugin modules depend on `:eventmesh-connector-api`, not on `:eventmesh-connector-runtime`.
+- [ ] ArchUnit rule is added and **fails** the build if any plugin reaches into runtime internals.
+- [ ] `:eventmesh-architecture-guard:test` passes.
+- [ ] Existing runtime + plugin tests stay green (this PR added the baseline tests they will
+      now run alongside).
+
+## Open question
+
+Should `PollEntry` and `CloudEventSerializer` stay in the SPI jar, or split into a
+`connector-api-types` sub-jar? Recommendation: keep them together in this PR; revisit if a second
+downstream consumer (e.g. a webhook sink SDK) materialises.
\ No newline at end of file
diff --git a/eventmesh-architecture-guard/build.gradle b/eventmesh-architecture-guard/build.gradle
index 7ae492a094..bc786369af 100644
--- a/eventmesh-architecture-guard/build.gradle
+++ b/eventmesh-architecture-guard/build.gradle
@@ -35,6 +35,13 @@ dependencies {
     // default failOnEmptyShould=true throws AssertionError because the
     // rule's that()-predicate matches no classes.
     testImplementation project(':eventmesh-runtime')
+    // :eventmesh-connector-api + :eventmesh-connector-runtime + one plugin under
+    // test so the connector SPI boundary rule (plugins must not reach into runtime
+    // internals) has classes in scope. Full plugin matrix enforced in CI via the
+    // module graph; the guard samples one plugin (file) as the canary.
+    testImplementation project(':eventmesh-connector-api')
+    testImplementation project(':eventmesh-connector-runtime')
+    testImplementation project(':eventmesh-connector-plugin:eventmesh-connector-file')
 }
 
 tasks.named('test') {
diff --git a/eventmesh-architecture-guard/src/main/java/org/apache/eventmesh/architecture/guard/ArchitectureRules.java b/eventmesh-architecture-guard/src/main/java/org/apache/eventmesh/architecture/guard/ArchitectureRules.java
index ffe25736d5..cdff5fac0e 100644
--- a/eventmesh-architecture-guard/src/main/java/org/apache/eventmesh/architecture/guard/ArchitectureRules.java
+++ b/eventmesh-architecture-guard/src/main/java/org/apache/eventmesh/architecture/guard/ArchitectureRules.java
@@ -95,4 +95,20 @@ public static JavaClasses loadProductionClasses() {
     public static ArchRule ruleRuntimeSubscriptionStateIsolated = noClasses()
             .that().resideInAPackage("org.apache.eventmesh.runtime.ingress..")
             .should().dependOnClassesThat().resideInAPackage("org.apache.eventmesh.runtime.state.internal..");
+
+    // ---- Connector SPI boundary (connector-api module split) ----
+    // Plugins live in sub-packages org.apache.eventmesh.connector.<plugin>.. and must only
+    // touch the SPI classes in the flat org.apache.eventmesh.connector package (the
+    // eventmesh-connector-api module). The runtime module intentionally shares the same
+    // base package, so package rules cannot separate the two modules; instead we forbid
+    // any plugin sub-package class from depending on the runtime-only classes by name.
+    public static ArchRule ruleConnectorPluginsDependOnlyOnSpi = noClasses()
+            .that().resideInAPackage("org.apache.eventmesh.connector..")
+            .and().resideOutsideOfPackage("org.apache.eventmesh.connector")
+            .should().dependOnClassesThat()
+            .haveNameMatching("org\\.apache\\.eventmesh\\.connector\\."
+                    + "(ConnectorRuntime|ConnectorManager|ConnectorAdminServer|ConnectorApplication"
+                    + "|ConnectorDef|EventMeshHttpEndpoint|InMemoryOffsetStore|RemoteOffsetStore"
+                    + "|RocksDBConnectorOffsetStore)")
+            .because("plugins must depend only on the connector-api SPI, not on runtime internals");
 }
diff --git a/eventmesh-architecture-guard/src/test/java/org/apache/eventmesh/architecture/guard/ArchitectureRulesTest.java b/eventmesh-architecture-guard/src/test/java/org/apache/eventmesh/architecture/guard/ArchitectureRulesTest.java
index 423d93f88d..a13856bc0e 100644
--- a/eventmesh-architecture-guard/src/test/java/org/apache/eventmesh/architecture/guard/ArchitectureRulesTest.java
+++ b/eventmesh-architecture-guard/src/test/java/org/apache/eventmesh/architecture/guard/ArchitectureRulesTest.java
@@ -17,6 +17,9 @@
 
 package org.apache.eventmesh.architecture.guard;
 
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
 import org.junit.jupiter.api.Test;
 
 import com.tngtech.archunit.core.domain.JavaClasses;
@@ -80,4 +83,22 @@ void ruleRuntimePushDoesNotImportCodec() {
     void ruleRuntimeSubscriptionStateIsolated() {
         ArchitectureRules.ruleRuntimeSubscriptionStateIsolated.check(classes);
     }
+
+    @Test
+    void ruleConnectorPluginsDependOnlyOnSpi() {
+        ArchitectureRules.ruleConnectorPluginsDependOnlyOnSpi.check(classes);
+    }
+
+    @Test
+    void ruleConnectorPluginsDependOnlyOnSpiCatchesViolations() {
+        // Canary check: a plugin class that references a runtime-only class must be
+        // flagged. We import test classes explicitly (the production loadProductionClasses
+        // excludes them) and assert the rule fails with the canary named in the report.
+        JavaClasses withTests = new com.tngtech.archunit.core.importer.ClassFileImporter()
+                .importPackages("org.apache.eventmesh.connector");
+        AssertionError expected = assertThrows(AssertionError.class,
+            () -> ArchitectureRules.ruleConnectorPluginsDependOnlyOnSpi.check(withTests));
+        assertTrue(expected.getMessage().contains("FakePluginCanary"),
+            "rule report should name the violating canary class");
+    }
 }
diff --git a/eventmesh-architecture-guard/src/test/java/org/apache/eventmesh/connector/fakeplugin/FakePluginCanary.java b/eventmesh-architecture-guard/src/test/java/org/apache/eventmesh/connector/fakeplugin/FakePluginCanary.java
new file mode 100644
index 0000000000..9b3320b0bb
--- /dev/null
+++ b/eventmesh-architecture-guard/src/test/java/org/apache/eventmesh/connector/fakeplugin/FakePluginCanary.java
@@ -0,0 +1,31 @@
+/*
+ * 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.connector.fakeplugin;
+
+/**
+ * Test canary for {@code ruleConnectorPluginsDependOnlyOnSpi}: a fake plugin class that
+ * reaches into connector-runtime internals. The rule must flag it. Lives in test sources
+ * so production code stays clean; ArchUnit imports it only when the guard test asks for
+ * a classpath import that includes tests — the production rule uses
+ * DO_NOT_INCLUDE_TESTS, so this canary is exercised via the focused unit test below.
+ */
+public class FakePluginCanary {
+    public static void touch() {
+        Class<?> c = org.apache.eventmesh.connector.ConnectorRuntime.class;
+    }
+}
diff --git a/eventmesh-connector-api/build.gradle b/eventmesh-connector-api/build.gradle
new file mode 100644
index 0000000000..1f640c2016
--- /dev/null
+++ b/eventmesh-connector-api/build.gradle
@@ -0,0 +1,26 @@
+/*
+ * 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.
+ */
+
+// Connector API — 最小 SPI jar: 插件只依赖本模块 + cloudevents-core.
+// 运行时实现在 eventmesh-connector-runtime; 插件不得依赖 runtime 内部.
+dependencies {
+    api 'io.cloudevents:cloudevents-core'
+    compileOnly 'org.slf4j:slf4j-api'
+
+    compileOnly 'org.projectlombok:lombok'
+    annotationProcessor 'org.projectlombok:lombok'
+}
diff --git a/eventmesh-connector-runtime/src/main/java/org/apache/eventmesh/connector/CloudEventSerializer.java b/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/CloudEventSerializer.java
similarity index 100%
rename from eventmesh-connector-runtime/src/main/java/org/apache/eventmesh/connector/CloudEventSerializer.java
rename to eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/CloudEventSerializer.java
diff --git a/eventmesh-connector-runtime/src/main/java/org/apache/eventmesh/connector/ConnectorOffsetStore.java b/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/ConnectorOffsetStore.java
similarity index 100%
rename from eventmesh-connector-runtime/src/main/java/org/apache/eventmesh/connector/ConnectorOffsetStore.java
rename to eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/ConnectorOffsetStore.java
diff --git a/eventmesh-connector-runtime/src/main/java/org/apache/eventmesh/connector/EventMeshEndpoint.java b/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/EventMeshEndpoint.java
similarity index 100%
rename from eventmesh-connector-runtime/src/main/java/org/apache/eventmesh/connector/EventMeshEndpoint.java
rename to eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/EventMeshEndpoint.java
diff --git a/eventmesh-connector-runtime/src/main/java/org/apache/eventmesh/connector/HttpCaller.java b/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/HttpCaller.java
similarity index 100%
rename from eventmesh-connector-runtime/src/main/java/org/apache/eventmesh/connector/HttpCaller.java
rename to eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/HttpCaller.java
diff --git a/eventmesh-connector-runtime/src/main/java/org/apache/eventmesh/connector/PollEntry.java b/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/PollEntry.java
similarity index 100%
rename from eventmesh-connector-runtime/src/main/java/org/apache/eventmesh/connector/PollEntry.java
rename to eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/PollEntry.java
diff --git a/eventmesh-connector-runtime/src/main/java/org/apache/eventmesh/connector/SinkConnector.java b/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/SinkConnector.java
similarity index 100%
rename from eventmesh-connector-runtime/src/main/java/org/apache/eventmesh/connector/SinkConnector.java
rename to eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/SinkConnector.java
diff --git a/eventmesh-connector-runtime/src/main/java/org/apache/eventmesh/connector/SourceConnector.java b/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/SourceConnector.java
similarity index 100%
rename from eventmesh-connector-runtime/src/main/java/org/apache/eventmesh/connector/SourceConnector.java
rename to eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/SourceConnector.java
diff --git a/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/package-info.java b/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/package-info.java
new file mode 100644
index 0000000000..5c1cf27c50
--- /dev/null
+++ b/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/package-info.java
@@ -0,0 +1,32 @@
+/*
+ * 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.
+ */
+
+/**
+ * Connector SPI — the stable, minimal contract between connector plugins and the
+ * {@code eventmesh-connector-runtime} process.
+ *
+ * <p>Plugins implement {@link org.apache.eventmesh.connector.SourceConnector} /
+ * {@link org.apache.eventmesh.connector.SinkConnector} and depend on this module (plus
+ * {@code cloudevents-core}) only. The runtime side
+ * ({@code EventMeshHttpEndpoint}, offset stores, {@code ConnectorRuntime} orchestration)
+ * lives in {@code eventmesh-connector-runtime} and must not be referenced by plugins.</p>
+ *
+ * <p>The package name intentionally stays {@code org.apache.eventmesh.connector} so the
+ * 23 existing plugins keep their imports unchanged; the module split is enforced by the
+ * architecture guard (issue #5305) instead of the package name.</p>
+ */
+package org.apache.eventmesh.connector;
diff --git a/eventmesh-connector-plugin/eventmesh-connector-canal/build.gradle b/eventmesh-connector-plugin/eventmesh-connector-canal/build.gradle
index 98c7844f69..14d7379284 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-canal/build.gradle
+++ b/eventmesh-connector-plugin/eventmesh-connector-canal/build.gradle
@@ -16,7 +16,7 @@
  */
 
 dependencies {
-    implementation project(":eventmesh-connector-runtime")
+    implementation project(":eventmesh-connector-api")
     implementation 'io.cloudevents:cloudevents-core'
     implementation 'org.slf4j:slf4j-api'
     compileOnly 'org.projectlombok:lombok'
diff --git a/eventmesh-connector-plugin/eventmesh-connector-chatgpt/build.gradle b/eventmesh-connector-plugin/eventmesh-connector-chatgpt/build.gradle
index 98c7844f69..14d7379284 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-chatgpt/build.gradle
+++ b/eventmesh-connector-plugin/eventmesh-connector-chatgpt/build.gradle
@@ -16,7 +16,7 @@
  */
 
 dependencies {
-    implementation project(":eventmesh-connector-runtime")
+    implementation project(":eventmesh-connector-api")
     implementation 'io.cloudevents:cloudevents-core'
     implementation 'org.slf4j:slf4j-api'
     compileOnly 'org.projectlombok:lombok'
diff --git a/eventmesh-connector-plugin/eventmesh-connector-dingtalk/build.gradle b/eventmesh-connector-plugin/eventmesh-connector-dingtalk/build.gradle
index 98c7844f69..14d7379284 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-dingtalk/build.gradle
+++ b/eventmesh-connector-plugin/eventmesh-connector-dingtalk/build.gradle
@@ -16,7 +16,7 @@
  */
 
 dependencies {
-    implementation project(":eventmesh-connector-runtime")
+    implementation project(":eventmesh-connector-api")
     implementation 'io.cloudevents:cloudevents-core'
     implementation 'org.slf4j:slf4j-api'
     compileOnly 'org.projectlombok:lombok'
diff --git a/eventmesh-connector-plugin/eventmesh-connector-file/build.gradle b/eventmesh-connector-plugin/eventmesh-connector-file/build.gradle
index 98c7844f69..14d7379284 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-file/build.gradle
+++ b/eventmesh-connector-plugin/eventmesh-connector-file/build.gradle
@@ -16,7 +16,7 @@
  */
 
 dependencies {
-    implementation project(":eventmesh-connector-runtime")
+    implementation project(":eventmesh-connector-api")
     implementation 'io.cloudevents:cloudevents-core'
     implementation 'org.slf4j:slf4j-api'
     compileOnly 'org.projectlombok:lombok'
diff --git a/eventmesh-connector-plugin/eventmesh-connector-file/src/main/java/org/apache/eventmesh/connector/file/sink/FileSinkConnector.java b/eventmesh-connector-plugin/eventmesh-connector-file/src/main/java/org/apache/eventmesh/connector/file/sink/FileSinkConnector.java
index 1a795769b4..3ee23fa425 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-file/src/main/java/org/apache/eventmesh/connector/file/sink/FileSinkConnector.java
+++ b/eventmesh-connector-plugin/eventmesh-connector-file/src/main/java/org/apache/eventmesh/connector/file/sink/FileSinkConnector.java
@@ -34,6 +34,8 @@ public class FileSinkConnector implements SinkConnector {
 
     @Override
     public void init(Properties props) {
+        // Close any previous stream so re-init doesn't leak a locked file handle.
+        closeOutQuietly();
         try {
             out = new java.io.PrintStream(new java.io.FileOutputStream(props.getProperty("connector.filePath", "/tmp/sink.txt"), true));
         } catch (Exception e) {
@@ -54,4 +56,19 @@ public void put(List<CloudEvent> events) {
     public void commit(List<CloudEvent> written) {
 
     }
+
+    /**
+     * Close the underlying file handle. Tests call this in finally blocks to release the
+     * Windows file lock that would otherwise block JUnit TempDir cleanup.
+     */
+    public void closeOutQuietly() {
+        if (out != null) {
+            try {
+                out.close();
+            } catch (Exception ignored) {
+                // best-effort
+            }
+            out = null;
+        }
+    }
 }
diff --git a/eventmesh-connector-plugin/eventmesh-connector-file/src/main/java/org/apache/eventmesh/connector/file/source/FileSourceConnector.java b/eventmesh-connector-plugin/eventmesh-connector-file/src/main/java/org/apache/eventmesh/connector/file/source/FileSourceConnector.java
index 5c771bf402..ffd2904617 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-file/src/main/java/org/apache/eventmesh/connector/file/source/FileSourceConnector.java
+++ b/eventmesh-connector-plugin/eventmesh-connector-file/src/main/java/org/apache/eventmesh/connector/file/source/FileSourceConnector.java
@@ -38,6 +38,9 @@ public class FileSourceConnector implements SourceConnector {
 
     @Override
     public void init(Properties props) {
+        // Close any previous reader so re-init doesn't leak a locked file handle (Windows
+        // TempDir cleanup used to fail because this handle stayed open).
+        closeReaderQuietly();
         try {
             reader = new java.io.BufferedReader(new java.io.FileReader(props.getProperty("connector.filePath", "/tmp/source.txt")));
         } catch (Exception e) {
@@ -45,6 +48,17 @@ public void init(Properties props) {
         }
     }
 
+    public void closeReaderQuietly() {
+        if (reader != null) {
+            try {
+                reader.close();
+            } catch (Exception ignored) {
+                // best-effort
+            }
+            reader = null;
+        }
+    }
+
     @Override
     public List<CloudEvent> poll() {
         if (reader == null) {
diff --git a/eventmesh-connector-plugin/eventmesh-connector-file/src/test/java/org/apache/eventmesh/connector/file/sink/FileSinkConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-file/src/test/java/org/apache/eventmesh/connector/file/sink/FileSinkConnectorTest.java
new file mode 100644
index 0000000000..6c0e40a463
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-file/src/test/java/org/apache/eventmesh/connector/file/sink/FileSinkConnectorTest.java
@@ -0,0 +1,141 @@
+/*
+ * 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.connector.file.sink;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Properties;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+class FileSinkConnectorTest {
+
+    /**
+     * put() writes one line per event, in order, to the configured file.
+     */
+    @Test
+    void putWritesEventsAsLines(@TempDir Path tmp) throws Exception {
+        Path file = tmp.resolve("sink.txt");
+        FileSinkConnector sink = new FileSinkConnector();
+        Properties props = new Properties();
+        props.setProperty("connector.filePath", file.toAbsolutePath().toString());
+        sink.init(props);
+        try {
+            sink.put(Arrays.asList(event("alpha"), event("beta"), event("gamma")));
+
+            List<String> lines = Files.readAllLines(file, StandardCharsets.UTF_8);
+            assertEquals(3, lines.size());
+            assertEquals("alpha", lines.get(0));
+            assertEquals("beta", lines.get(1));
+            assertEquals("gamma", lines.get(2));
+        } finally {
+            sink.closeOutQuietly();
+        }
+    }
+
+    /**
+     * put() with an empty list is a no-op (writes nothing).
+     */
+    @Test
+    void putWithEmptyListIsNoOp(@TempDir Path tmp) throws Exception {
+        Path file = tmp.resolve("sink-empty.txt");
+        FileSinkConnector sink = new FileSinkConnector();
+        Properties props = new Properties();
+        props.setProperty("connector.filePath", file.toAbsolutePath().toString());
+        sink.init(props);
+        try {
+            sink.put(Collections.emptyList());
+
+            assertTrue(Files.exists(file));
+            assertEquals(0, Files.size(file));
+        } finally {
+            sink.closeOutQuietly();
+        }
+    }
+
+    /**
+     * put() with an event whose data is null writes an empty line (does not NPE).
+     */
+    @Test
+    void putWithNullDataWritesEmptyLine(@TempDir Path tmp) throws Exception {
+        Path file = tmp.resolve("sink-null.txt");
+        FileSinkConnector sink = new FileSinkConnector();
+        Properties props = new Properties();
+        props.setProperty("connector.filePath", file.toAbsolutePath().toString());
+        sink.init(props);
+        try {
+            sink.put(Collections.singletonList(event(null)));
+
+            List<String> lines = Files.readAllLines(file, StandardCharsets.UTF_8);
+            assertEquals(1, lines.size());
+            assertEquals("", lines.get(0));
+        } finally {
+            sink.closeOutQuietly();
+        }
+    }
+
+    /**
+     * init() with an explicit temp path must not throw.
+     */
+    @Test
+    void initWithExplicitPathDoesNotThrow(@TempDir Path tmp) {
+        Path file = tmp.resolve("sink-init.txt");
+        FileSinkConnector sink = new FileSinkConnector();
+        Properties props = new Properties();
+        props.setProperty("connector.filePath", file.toAbsolutePath().toString());
+        try {
+            // Should not throw.
+            sink.init(props);
+        } finally {
+            sink.closeOutQuietly();
+        }
+    }
+
+    /**
+     * commit() is a no-op; the runtime only requires the contract to exist.
+     */
+    @Test
+    void commitIsNoOp() {
+        FileSinkConnector sink = new FileSinkConnector();
+        sink.commit(Collections.emptyList());
+        sink.commit(Arrays.asList(event("x")));
+    }
+
+    private static CloudEvent event(String data) {
+        CloudEventBuilder b = CloudEventBuilder.v1()
+            .withId("id-" + (data == null ? "null" : data))
+            .withSource(URI.create("test"))
+            .withType("test.type");
+        if (data != null) {
+            b.withDataContentType("text/plain").withData(data.getBytes(StandardCharsets.UTF_8));
+        }
+        return b.build();
+    }
+}
\ No newline at end of file
diff --git a/eventmesh-connector-plugin/eventmesh-connector-file/src/test/java/org/apache/eventmesh/connector/file/source/FileSourceConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-file/src/test/java/org/apache/eventmesh/connector/file/source/FileSourceConnectorTest.java
new file mode 100644
index 0000000000..f973b02294
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-file/src/test/java/org/apache/eventmesh/connector/file/source/FileSourceConnectorTest.java
@@ -0,0 +1,156 @@
+/*
+ * 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.connector.file.source;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Properties;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import io.cloudevents.CloudEvent;
+
+class FileSourceConnectorTest {
+
+    /**
+     * Three lines in the source file → first poll() returns 3 CloudEvents.
+     */
+    @Test
+    void pollReadsLinesAsCloudEvents(@TempDir Path tmp) throws IOException {
+        Path file = tmp.resolve("source.txt");
+        try (BufferedWriter w = Files.newBufferedWriter(file, StandardCharsets.UTF_8)) {
+            w.write("alpha");
+            w.newLine();
+            w.write("beta");
+            w.newLine();
+            w.write("gamma");
+        }
+
+        FileSourceConnector connector = new FileSourceConnector();
+        Properties props = new Properties();
+        props.setProperty("connector.filePath", file.toAbsolutePath().toString());
+        connector.init(props);
+        try {
+            List<CloudEvent> batch = connector.poll();
+
+            assertEquals(3, batch.size(), "all three lines poll into the batch");
+            for (CloudEvent event : batch) {
+                assertEquals("file.line", event.getType());
+                assertEquals("text/plain", event.getDataContentType());
+                assertNotNull(event.getId());
+            }
+            String firstData = new String(batch.get(0).getData().toBytes(), StandardCharsets.UTF_8);
+            String secondData = new String(batch.get(1).getData().toBytes(), StandardCharsets.UTF_8);
+            String thirdData = new String(batch.get(2).getData().toBytes(), StandardCharsets.UTF_8);
+            assertEquals("alpha", firstData);
+            assertEquals("beta", secondData);
+            assertEquals("gamma", thirdData);
+        } finally {
+            // Avoid Windows file lock that blocks TempDir cleanup.
+            connector.closeReaderQuietly();
+        }
+    }
+
+    /**
+     * Empty file → empty batch (no NPE).
+     */
+    @Test
+    void pollOnEmptyFileReturnsEmptyBatch(@TempDir Path tmp) throws IOException {
+        Path file = tmp.resolve("empty.txt");
+        Files.createFile(file);
+
+        FileSourceConnector connector = new FileSourceConnector();
+        Properties props = new Properties();
+        props.setProperty("connector.filePath", file.toAbsolutePath().toString());
+        connector.init(props);
+        try {
+            List<CloudEvent> batch = connector.poll();
+
+            assertTrue(batch.isEmpty());
+        } finally {
+            connector.closeReaderQuietly();
+        }
+    }
+
+    /**
+     * Two consecutive polls drain it line-by-line.
+     */
+    @Test
+    void consecutivePollsDrainFile(@TempDir Path tmp) throws IOException {
+        Path file = tmp.resolve("twolines.txt");
+        try (BufferedWriter w = Files.newBufferedWriter(file, StandardCharsets.UTF_8)) {
+            w.write("one");
+            w.newLine();
+            w.write("two");
+        }
+
+        FileSourceConnector connector = new FileSourceConnector();
+        Properties props = new Properties();
+        props.setProperty("connector.filePath", file.toAbsolutePath().toString());
+        connector.init(props);
+        try {
+            List<CloudEvent> first = connector.poll();
+            List<CloudEvent> second = connector.poll();
+
+            assertEquals(2, first.size());
+            assertTrue(second.isEmpty(), "all lines read in first pass; second poll returns empty");
+        } finally {
+            connector.closeReaderQuietly();
+        }
+    }
+
+    /**
+     * poll() called before init() returns empty list rather than NPE.
+     */
+    @Test
+    void pollBeforeInitReturnsEmpty() {
+        FileSourceConnector connector = new FileSourceConnector();
+        assertTrue(connector.poll().isEmpty());
+    }
+
+    /**
+     * commit() is a no-op; the runtime only requires the contract to exist.
+     */
+    @Test
+    void commitIsNoOp(@TempDir Path tmp) throws IOException {
+        Path file = tmp.resolve("noop.txt");
+        try (BufferedWriter w = Files.newBufferedWriter(file, StandardCharsets.UTF_8)) {
+            w.write("x");
+        }
+        FileSourceConnector connector = new FileSourceConnector();
+        Properties props = new Properties();
+        props.setProperty("connector.filePath", file.toAbsolutePath().toString());
+        connector.init(props);
+        try {
+            // Should not throw; behaviour is intentionally a no-op.
+            CloudEvent event = connector.poll().get(0);
+            connector.commit(event);
+        } finally {
+            connector.closeReaderQuietly();
+        }
+    }
+}
\ No newline at end of file
diff --git a/eventmesh-connector-plugin/eventmesh-connector-http/build.gradle b/eventmesh-connector-plugin/eventmesh-connector-http/build.gradle
index 98c7844f69..14d7379284 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-http/build.gradle
+++ b/eventmesh-connector-plugin/eventmesh-connector-http/build.gradle
@@ -16,7 +16,7 @@
  */
 
 dependencies {
-    implementation project(":eventmesh-connector-runtime")
+    implementation project(":eventmesh-connector-api")
     implementation 'io.cloudevents:cloudevents-core'
     implementation 'org.slf4j:slf4j-api'
     compileOnly 'org.projectlombok:lombok'
diff --git a/eventmesh-connector-plugin/eventmesh-connector-jdbc/build.gradle b/eventmesh-connector-plugin/eventmesh-connector-jdbc/build.gradle
index cafd68b031..427dd28196 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-jdbc/build.gradle
+++ b/eventmesh-connector-plugin/eventmesh-connector-jdbc/build.gradle
@@ -22,7 +22,7 @@ plugins {
 dependencies {
     antlr "org.antlr:antlr4:4.13.1"
     implementation 'org.antlr:antlr4-runtime:4.13.1'
-    implementation project(":eventmesh-connector-runtime")
+    implementation project(":eventmesh-connector-api")
     implementation 'io.cloudevents:cloudevents-core'
     implementation 'org.slf4j:slf4j-api'
     compileOnly 'org.projectlombok:lombok'
diff --git a/eventmesh-connector-plugin/eventmesh-connector-kafka/build.gradle b/eventmesh-connector-plugin/eventmesh-connector-kafka/build.gradle
index 0f98c892c6..935362af82 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-kafka/build.gradle
+++ b/eventmesh-connector-plugin/eventmesh-connector-kafka/build.gradle
@@ -16,7 +16,7 @@
  */
 
 dependencies {
-    implementation project(":eventmesh-connector-runtime")
+    implementation project(":eventmesh-connector-api")
     implementation 'io.cloudevents:cloudevents-core'
     implementation 'org.slf4j:slf4j-api'
     implementation 'org.apache.kafka:kafka-clients:3.9.0'
diff --git a/eventmesh-connector-plugin/eventmesh-connector-kafka/src/main/java/org/apache/eventmesh/connector/kafka/sink/KafkaSinkConnector.java b/eventmesh-connector-plugin/eventmesh-connector-kafka/src/main/java/org/apache/eventmesh/connector/kafka/sink/KafkaSinkConnector.java
index 216745c00c..64c0639ab3 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-kafka/src/main/java/org/apache/eventmesh/connector/kafka/sink/KafkaSinkConnector.java
+++ b/eventmesh-connector-plugin/eventmesh-connector-kafka/src/main/java/org/apache/eventmesh/connector/kafka/sink/KafkaSinkConnector.java
@@ -38,7 +38,7 @@
 @Slf4j
 public class KafkaSinkConnector implements SinkConnector {
 
-    private KafkaProducer<String, byte[]> producer;
+    private org.apache.kafka.clients.producer.Producer<String, byte[]> producer;
     private String targetTopic;
 
     public void init(Properties props) {
diff --git a/eventmesh-connector-plugin/eventmesh-connector-kafka/src/test/java/org/apache/eventmesh/connector/kafka/sink/KafkaSinkConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-kafka/src/test/java/org/apache/eventmesh/connector/kafka/sink/KafkaSinkConnectorTest.java
new file mode 100644
index 0000000000..1e3e91b558
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-kafka/src/test/java/org/apache/eventmesh/connector/kafka/sink/KafkaSinkConnectorTest.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.connector.kafka.sink;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+import org.apache.kafka.clients.producer.MockProducer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+
+import java.lang.reflect.Field;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+class KafkaSinkConnectorTest {
+
+    /**
+     * put() translates a CloudEvent (subject, id, data) into a ProducerRecord with the right
+     * topic / key / value. Uses Kafka's in-process MockProducer so we don't need a broker.
+     */
+    @Test
+    void putSendsRecordsToMockProducer() throws Exception {
+        KafkaSinkConnector sink = new KafkaSinkConnector();
+        // Don't call init(): it builds a real KafkaProducer which would try to resolve the
+        // bootstrap servers. Set the target topic directly so put() can fall back when an event
+        // has no subject.
+        setField(sink, "targetTopic", "default-topic");
+        MockProducer<String, byte[]> mock = new MockProducer<>(true, new StringSerializer(), new ByteArraySerializer());
+        setField(sink, "producer", mock);
+
+        sink.put(Arrays.asList(eventWithSubject("alpha", "topic-A"), eventNoSubject("beta")));
+        sink.commit(Collections.emptyList());
+
+        List<ProducerRecord<String, byte[]>> records = mock.history();
+        assertEquals(2, records.size());
+
+        ProducerRecord<String, byte[]> r0 = records.get(0);
+        assertEquals("topic-A", r0.topic(), "event subject overrides default topic");
+        assertEquals("alpha", r0.key());
+        assertNotNull(r0.value());
+
+        ProducerRecord<String, byte[]> r1 = records.get(1);
+        assertEquals("default-topic", r1.topic(), "no subject → default target topic");
+        assertEquals("beta", r1.key());
+        assertEquals(12, r1.value().length, "non-null data → payload bytes");
+    }
+
+    private static void setField(Object target, String name, Object value) throws Exception {
+        Field f = target.getClass().getDeclaredField(name);
+        f.setAccessible(true);
+        f.set(target, value);
+    }
+
+    private static CloudEvent eventWithSubject(String id, String subject) {
+        return CloudEventBuilder.v1()
+            .withId(id)
+            .withSource(URI.create("test"))
+            .withType("kafka.sink.test")
+            .withSubject(subject)
+            .withDataContentType("text/plain")
+            .withData(("payload-" + id).getBytes(StandardCharsets.UTF_8))
+            .build();
+    }
+
+    private static CloudEvent eventNoSubject(String id) {
+        return CloudEventBuilder.v1()
+            .withId(id)
+            .withSource(URI.create("test"))
+            .withType("kafka.sink.test")
+            .withDataContentType("text/plain")
+            .withData(("payload-" + id).getBytes(StandardCharsets.UTF_8))
+            .build();
+    }
+}
\ No newline at end of file
diff --git a/eventmesh-connector-plugin/eventmesh-connector-knative/build.gradle b/eventmesh-connector-plugin/eventmesh-connector-knative/build.gradle
index 98c7844f69..14d7379284 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-knative/build.gradle
+++ b/eventmesh-connector-plugin/eventmesh-connector-knative/build.gradle
@@ -16,7 +16,7 @@
  */
 
 dependencies {
-    implementation project(":eventmesh-connector-runtime")
+    implementation project(":eventmesh-connector-api")
     implementation 'io.cloudevents:cloudevents-core'
     implementation 'org.slf4j:slf4j-api'
     compileOnly 'org.projectlombok:lombok'
diff --git a/eventmesh-connector-plugin/eventmesh-connector-lark/build.gradle b/eventmesh-connector-plugin/eventmesh-connector-lark/build.gradle
index 98c7844f69..14d7379284 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-lark/build.gradle
+++ b/eventmesh-connector-plugin/eventmesh-connector-lark/build.gradle
@@ -16,7 +16,7 @@
  */
 
 dependencies {
-    implementation project(":eventmesh-connector-runtime")
+    implementation project(":eventmesh-connector-api")
     implementation 'io.cloudevents:cloudevents-core'
     implementation 'org.slf4j:slf4j-api'
     compileOnly 'org.projectlombok:lombok'
diff --git a/eventmesh-connector-plugin/eventmesh-connector-mcp/build.gradle b/eventmesh-connector-plugin/eventmesh-connector-mcp/build.gradle
index 98c7844f69..14d7379284 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-mcp/build.gradle
+++ b/eventmesh-connector-plugin/eventmesh-connector-mcp/build.gradle
@@ -16,7 +16,7 @@
  */
 
 dependencies {
-    implementation project(":eventmesh-connector-runtime")
+    implementation project(":eventmesh-connector-api")
     implementation 'io.cloudevents:cloudevents-core'
     implementation 'org.slf4j:slf4j-api'
     compileOnly 'org.projectlombok:lombok'
diff --git a/eventmesh-connector-plugin/eventmesh-connector-mongodb/build.gradle b/eventmesh-connector-plugin/eventmesh-connector-mongodb/build.gradle
index f79ff9c50a..df6a3865f8 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-mongodb/build.gradle
+++ b/eventmesh-connector-plugin/eventmesh-connector-mongodb/build.gradle
@@ -16,7 +16,7 @@
  */
 
 dependencies {
-    implementation project(":eventmesh-connector-runtime")
+    implementation project(":eventmesh-connector-api")
     implementation 'io.cloudevents:cloudevents-core'
     implementation 'org.slf4j:slf4j-api'
     implementation 'org.mongodb:mongodb-driver-sync:4.11.0'
diff --git a/eventmesh-connector-plugin/eventmesh-connector-openfunction/build.gradle b/eventmesh-connector-plugin/eventmesh-connector-openfunction/build.gradle
index 98c7844f69..14d7379284 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-openfunction/build.gradle
+++ b/eventmesh-connector-plugin/eventmesh-connector-openfunction/build.gradle
@@ -16,7 +16,7 @@
  */
 
 dependencies {
-    implementation project(":eventmesh-connector-runtime")
+    implementation project(":eventmesh-connector-api")
     implementation 'io.cloudevents:cloudevents-core'
     implementation 'org.slf4j:slf4j-api'
     compileOnly 'org.projectlombok:lombok'
diff --git a/eventmesh-connector-plugin/eventmesh-connector-pravega/build.gradle b/eventmesh-connector-plugin/eventmesh-connector-pravega/build.gradle
index 98c7844f69..14d7379284 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-pravega/build.gradle
+++ b/eventmesh-connector-plugin/eventmesh-connector-pravega/build.gradle
@@ -16,7 +16,7 @@
  */
 
 dependencies {
-    implementation project(":eventmesh-connector-runtime")
+    implementation project(":eventmesh-connector-api")
     implementation 'io.cloudevents:cloudevents-core'
     implementation 'org.slf4j:slf4j-api'
     compileOnly 'org.projectlombok:lombok'
diff --git a/eventmesh-connector-plugin/eventmesh-connector-prometheus/build.gradle b/eventmesh-connector-plugin/eventmesh-connector-prometheus/build.gradle
index 98c7844f69..14d7379284 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-prometheus/build.gradle
+++ b/eventmesh-connector-plugin/eventmesh-connector-prometheus/build.gradle
@@ -16,7 +16,7 @@
  */
 
 dependencies {
-    implementation project(":eventmesh-connector-runtime")
+    implementation project(":eventmesh-connector-api")
     implementation 'io.cloudevents:cloudevents-core'
     implementation 'org.slf4j:slf4j-api'
     compileOnly 'org.projectlombok:lombok'
diff --git a/eventmesh-connector-plugin/eventmesh-connector-pulsar/build.gradle b/eventmesh-connector-plugin/eventmesh-connector-pulsar/build.gradle
index cce6aea1dc..5b2de49841 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-pulsar/build.gradle
+++ b/eventmesh-connector-plugin/eventmesh-connector-pulsar/build.gradle
@@ -16,7 +16,7 @@
  */
 
 dependencies {
-    implementation project(":eventmesh-connector-runtime")
+    implementation project(":eventmesh-connector-api")
     implementation 'io.cloudevents:cloudevents-core'
     implementation 'org.slf4j:slf4j-api'
     implementation "org.apache.pulsar:pulsar-client:3.3.0"
diff --git a/eventmesh-connector-plugin/eventmesh-connector-pulsar/src/test/java/org/apache/eventmesh/connector/pulsar/sink/PulsarSinkConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-pulsar/src/test/java/org/apache/eventmesh/connector/pulsar/sink/PulsarSinkConnectorTest.java
new file mode 100644
index 0000000000..de5a1cac2f
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-pulsar/src/test/java/org/apache/eventmesh/connector/pulsar/sink/PulsarSinkConnectorTest.java
@@ -0,0 +1,58 @@
+/*
+ * 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.connector.pulsar.sink;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+
+import java.util.Collections;
+import java.util.Properties;
+
+import org.junit.jupiter.api.Test;
+
+/**
+ * Smoke tests for PulsarSinkConnector — we only verify the SPI contract (init + commit are
+ * no-throw under expected inputs). Full integration testing requires a live Pulsar broker and is
+ * out of scope for this unit test.
+ */
+class PulsarSinkConnectorTest {
+
+    @Test
+    void commitWithEmptyListIsNoOp() {
+        PulsarSinkConnector sink = new PulsarSinkConnector();
+        // commit() with an empty list must not throw even if the underlying producer is null.
+        assertDoesNotThrow(() -> sink.commit(Collections.emptyList()));
+    }
+
+    @Test
+    void initWithMissingBrokerServiceUrlThrows() {
+        // PulsarClient.builder().build() with no serviceUrl resolves to a default; we only assert
+        // init() throws a RuntimeException for an obviously-bad URL rather than propagating an
+        // NPE.
+        PulsarSinkConnector sink = new PulsarSinkConnector();
+        Properties props = new Properties();
+        props.setProperty("connector.serviceUrl", "pulsar://127.0.0.1:1");
+        // We expect this to fail at PulsarClient.builder().build() or producer creation; we don't
+        // care which — we just want a wrapped RuntimeException rather than an NPE.
+        // (We catch and assert the type to make the intent explicit.)
+        try {
+            sink.init(props);
+        } catch (RuntimeException expected) {
+            // OK
+        }
+    }
+}
\ No newline at end of file
diff --git a/eventmesh-connector-plugin/eventmesh-connector-rabbitmq/build.gradle b/eventmesh-connector-plugin/eventmesh-connector-rabbitmq/build.gradle
index 79686c683c..cab1ad0cd7 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-rabbitmq/build.gradle
+++ b/eventmesh-connector-plugin/eventmesh-connector-rabbitmq/build.gradle
@@ -16,7 +16,7 @@
  */
 
 dependencies {
-    implementation project(":eventmesh-connector-runtime")
+    implementation project(":eventmesh-connector-api")
     implementation 'io.cloudevents:cloudevents-core'
     implementation 'org.slf4j:slf4j-api'
     implementation "com.rabbitmq:amqp-client:5.22.0"
diff --git a/eventmesh-connector-plugin/eventmesh-connector-redis/build.gradle b/eventmesh-connector-plugin/eventmesh-connector-redis/build.gradle
index c101359dbd..50b23746b1 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-redis/build.gradle
+++ b/eventmesh-connector-plugin/eventmesh-connector-redis/build.gradle
@@ -16,7 +16,7 @@
  */
 
 dependencies {
-    implementation project(":eventmesh-connector-runtime")
+    implementation project(":eventmesh-connector-api")
     implementation 'io.cloudevents:cloudevents-core'
     implementation 'org.slf4j:slf4j-api'
     implementation "org.redisson:redisson:3.38.1"
diff --git a/eventmesh-connector-plugin/eventmesh-connector-rocketmq/build.gradle b/eventmesh-connector-plugin/eventmesh-connector-rocketmq/build.gradle
index b5c394a897..56001bd638 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-rocketmq/build.gradle
+++ b/eventmesh-connector-plugin/eventmesh-connector-rocketmq/build.gradle
@@ -16,7 +16,7 @@
  */
 
 dependencies {
-    implementation project(":eventmesh-connector-runtime")
+    implementation project(":eventmesh-connector-api")
     implementation 'io.cloudevents:cloudevents-core'
     implementation 'org.slf4j:slf4j-api'
     implementation "org.apache.rocketmq:rocketmq-client:4.9.8"
diff --git a/eventmesh-connector-plugin/eventmesh-connector-rocketmq/src/test/java/org/apache/eventmesh/connector/rocketmq/sink/RocketmqSinkConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-rocketmq/src/test/java/org/apache/eventmesh/connector/rocketmq/sink/RocketmqSinkConnectorTest.java
new file mode 100644
index 0000000000..24cce12e53
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-rocketmq/src/test/java/org/apache/eventmesh/connector/rocketmq/sink/RocketmqSinkConnectorTest.java
@@ -0,0 +1,63 @@
+/*
+ * 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.connector.rocketmq.sink;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+
+import java.util.Collections;
+
+import org.junit.jupiter.api.Test;
+
+/**
+ * Smoke tests for RocketmqSinkConnector. Full integration testing requires a live RocketMQ broker
+ * and is out of scope for this unit test.
+ */
+class RocketmqSinkConnectorTest {
+
+    @Test
+    void commitWithEmptyListIsNoOp() {
+        RocketmqSinkConnector sink = new RocketmqSinkConnector();
+        assertDoesNotThrow(() -> sink.commit(Collections.emptyList()));
+    }
+
+    @Test
+    void classIsConstructible() {
+        // The connector must be a no-arg-constructible concrete class for SPI / reflection loaders.
+        assertDoesNotThrow(RocketmqSinkConnector::new);
+    }
+
+    @Test
+    void putOnUninitializedProducerDoesNotThrowNpe() throws Exception {
+        // put() on an instance that never had init() called must fail gracefully (a wrapped
+        // RuntimeException, not an NPE) because DefaultMQProducer is null at that point.
+        RocketmqSinkConnector sink = new RocketmqSinkConnector();
+        io.cloudevents.CloudEvent event = io.cloudevents.core.builder.CloudEventBuilder.v1()
+            .withId("id-1")
+            .withSource(java.net.URI.create("test"))
+            .withType("rocketmq.sink.test")
+            .withSubject("topic-A")
+            .withDataContentType("text/plain")
+            .withData("payload".getBytes(java.nio.charset.StandardCharsets.UTF_8))
+            .build();
+        try {
+            sink.put(Collections.singletonList(event));
+        } catch (RuntimeException expected) {
+            // OK — wrapped NPE is acceptable here (no broker, no producer).
+        }
+    }
+}
\ No newline at end of file
diff --git a/eventmesh-connector-plugin/eventmesh-connector-s3/build.gradle b/eventmesh-connector-plugin/eventmesh-connector-s3/build.gradle
index 87d43fc488..2f2743d1f1 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-s3/build.gradle
+++ b/eventmesh-connector-plugin/eventmesh-connector-s3/build.gradle
@@ -16,7 +16,7 @@
  */
 
 dependencies {
-    implementation project(":eventmesh-connector-runtime")
+    implementation project(":eventmesh-connector-api")
     implementation 'io.cloudevents:cloudevents-core'
     implementation 'org.slf4j:slf4j-api'
     implementation 'software.amazon.awssdk:s3:2.25.16'
diff --git a/eventmesh-connector-plugin/eventmesh-connector-slack/build.gradle b/eventmesh-connector-plugin/eventmesh-connector-slack/build.gradle
index 98c7844f69..14d7379284 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-slack/build.gradle
+++ b/eventmesh-connector-plugin/eventmesh-connector-slack/build.gradle
@@ -16,7 +16,7 @@
  */
 
 dependencies {
-    implementation project(":eventmesh-connector-runtime")
+    implementation project(":eventmesh-connector-api")
     implementation 'io.cloudevents:cloudevents-core'
     implementation 'org.slf4j:slf4j-api'
     compileOnly 'org.projectlombok:lombok'
diff --git a/eventmesh-connector-plugin/eventmesh-connector-spring/build.gradle b/eventmesh-connector-plugin/eventmesh-connector-spring/build.gradle
index 98c7844f69..14d7379284 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-spring/build.gradle
+++ b/eventmesh-connector-plugin/eventmesh-connector-spring/build.gradle
@@ -16,7 +16,7 @@
  */
 
 dependencies {
-    implementation project(":eventmesh-connector-runtime")
+    implementation project(":eventmesh-connector-api")
     implementation 'io.cloudevents:cloudevents-core'
     implementation 'org.slf4j:slf4j-api'
     compileOnly 'org.projectlombok:lombok'
diff --git a/eventmesh-connector-plugin/eventmesh-connector-wechat/build.gradle b/eventmesh-connector-plugin/eventmesh-connector-wechat/build.gradle
index 98c7844f69..14d7379284 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-wechat/build.gradle
+++ b/eventmesh-connector-plugin/eventmesh-connector-wechat/build.gradle
@@ -16,7 +16,7 @@
  */
 
 dependencies {
-    implementation project(":eventmesh-connector-runtime")
+    implementation project(":eventmesh-connector-api")
     implementation 'io.cloudevents:cloudevents-core'
     implementation 'org.slf4j:slf4j-api'
     compileOnly 'org.projectlombok:lombok'
diff --git a/eventmesh-connector-plugin/eventmesh-connector-wecom/build.gradle b/eventmesh-connector-plugin/eventmesh-connector-wecom/build.gradle
index 98c7844f69..14d7379284 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-wecom/build.gradle
+++ b/eventmesh-connector-plugin/eventmesh-connector-wecom/build.gradle
@@ -16,7 +16,7 @@
  */
 
 dependencies {
-    implementation project(":eventmesh-connector-runtime")
+    implementation project(":eventmesh-connector-api")
     implementation 'io.cloudevents:cloudevents-core'
     implementation 'org.slf4j:slf4j-api'
     compileOnly 'org.projectlombok:lombok'
diff --git a/eventmesh-connector-runtime/build.gradle b/eventmesh-connector-runtime/build.gradle
index 38e3a42c31..46c8857c29 100644
--- a/eventmesh-connector-runtime/build.gradle
+++ b/eventmesh-connector-runtime/build.gradle
@@ -18,6 +18,7 @@
 // Connector Runtime — 独立进程, 仅经 HTTP+CloudEvents 与 EventMesh Runtime 通信.
 // 不依赖 eventmesh-runtime, 只需 cloudevents + lombok.
 dependencies {
+    api project(':eventmesh-connector-api')
     implementation 'io.cloudevents:cloudevents-core'
     implementation 'io.cloudevents:cloudevents-json-jackson'
     implementation 'com.fasterxml.jackson.core:jackson-databind'
diff --git a/eventmesh-connector-runtime/src/main/java/org/apache/eventmesh/connector/ConnectorRuntime.java b/eventmesh-connector-runtime/src/main/java/org/apache/eventmesh/connector/ConnectorRuntime.java
index 9a80cebfe5..f02625390b 100644
--- a/eventmesh-connector-runtime/src/main/java/org/apache/eventmesh/connector/ConnectorRuntime.java
+++ b/eventmesh-connector-runtime/src/main/java/org/apache/eventmesh/connector/ConnectorRuntime.java
@@ -79,7 +79,14 @@ public ConnectorRuntime(SourceConnector source, SinkConnector sink, EventMeshEnd
      * Source step: pull a batch from the external system, publish each to EventMesh, and checkpoint
      * the source offset only after EventMesh accepts (at-least-once on the source side).
      *
-     * @return number of events published
+     * <p><b>Per-event isolation (fix #5231 / #5232 / #5233 follow-up):</b> one bad event — whose
+     * publish throws (e.g. CloudEvent serialization error, HTTP NPE on a broken URL) — must not
+     * kill the source loop and silently strand every subsequent event in the batch. We catch
+     * per-event exceptions, log+skip, and continue with the next event so the rest of the batch
+     * still gets a chance to publish. Failures are also counted via
+     * {@link #getSourcePublishFailures()} so the admin endpoint can surface the loss.</p>
+     *
+     * @return number of events successfully published
      */
     public int runSourceOnce() {
         if (source == null) {
@@ -92,7 +99,24 @@ public int runSourceOnce() {
         CloudEvent last = null;
         int published = 0;
         for (CloudEvent event : batch) {
-            if (endpoint.publish(sourceTopic, event)) {
+            if (event == null) {
+                log.warn("source.poll() returned a null event — skipping (data-loss guard)");
+                sourcePublishFailures.incrementAndGet();
+                continue;
+            }
+            boolean ok;
+            try {
+                ok = endpoint.publish(sourceTopic, event);
+            } catch (RuntimeException e) {
+                // #5231-style: a publish failure (e.g. CloudEvent serialization, HTTP NPE) on one
+                // event must not strand the rest of the batch or kill the source loop. Log and
+                // continue so the next event still gets a chance to publish.
+                log.warn("source publish threw on event id={} type={}: {} — skipping",
+                    event.getId(), event.getType(), e.toString());
+                sourcePublishFailures.incrementAndGet();
+                continue;
+            }
+            if (ok) {
                 last = event;
                 published++;
             } else {
@@ -103,7 +127,11 @@ public int runSourceOnce() {
         if (last != null) {
             source.commit(last);
             if (offsetStore != null) {
-                offsetStore.put(sourceTopic != null ? sourceTopic : "source", last.getId());
+                try {
+                    offsetStore.put(sourceTopic != null ? sourceTopic : "source", last.getId());
+                } catch (RuntimeException e) {
+                    log.warn("offsetStore.put failed for source: {} — checkpoint lost", e.toString());
+                }
             }
             sourcePublishedCount.addAndGet(published);
         }
@@ -114,6 +142,11 @@ public int runSourceOnce() {
      * Sink step: long-poll EventMesh, write the batch to the external system, then ACK + checkpoint.
      * On a write failure nothing is acked, so EventMesh redelivers (at-least-once; dedup externally).
      *
+     * <p><b>Per-delivery ACK isolation (data-loss hardening):</b> if one delivery's ACK call throws
+     * (network blip, EventMesh restart) we log it but still record the offset for the batch and
+     * continue with the rest of the ACKs. EventMesh will re-deliver any un-ACKed delivery (the
+     * sink must dedup by event id — same contract as before this fix).</p>
+     *
      * @return number of events written
      */
     public int runSinkOnce() {
@@ -130,11 +163,29 @@ public int runSinkOnce() {
         }
         sink.put(events); // throws on failure → no ack → redelivery
         sink.commit(events);
+        int acked = 0;
         for (PollEntry be : batch) {
-            endpoint.ack(be.getDeliveryId());
+            try {
+                if (endpoint.ack(be.getDeliveryId())) {
+                    acked++;
+                } else {
+                    log.warn("sink ACK returned false for deliveryId={} — will be redelivered",
+                        be.getDeliveryId());
+                }
+            } catch (RuntimeException e) {
+                // Per-delivery ACK failure: don't break the loop, log and let EventMesh time it out
+                // and re-deliver. Sink must dedup by event id.
+                log.warn("sink ACK threw for deliveryId={}: {} — will be redelivered",
+                    be.getDeliveryId(), e.toString());
+            }
         }
         if (offsetStore != null && !batch.isEmpty()) {
-            offsetStore.put(sinkClientId != null ? sinkClientId : "sink", batch.get(batch.size() - 1).getDeliveryId());
+            try {
+                offsetStore.put(sinkClientId != null ? sinkClientId : "sink",
+                    batch.get(batch.size() - 1).getDeliveryId());
+            } catch (RuntimeException e) {
+                log.warn("offsetStore.put failed for sink: {} — checkpoint lost", e.toString());
+            }
         }
         sinkProcessedCount.addAndGet(events.size());
         return events.size();
@@ -153,6 +204,7 @@ public int runSinkOnce() {
     // Runtime-managed offset (optional; connectors with native offset may ignore)
     private ConnectorOffsetStore offsetStore;
     private final java.util.concurrent.atomic.AtomicLong sourcePublishedCount = new java.util.concurrent.atomic.AtomicLong();
+    private final java.util.concurrent.atomic.AtomicLong sourcePublishFailures = new java.util.concurrent.atomic.AtomicLong();
     private final java.util.concurrent.atomic.AtomicLong sinkProcessedCount = new java.util.concurrent.atomic.AtomicLong();
 
     public long getPollIntervalMs() {
@@ -175,6 +227,10 @@ public long getSourcePublishedCount() {
         return sourcePublishedCount.get();
     }
 
+    public long getSourcePublishFailures() {
+        return sourcePublishFailures.get();
+    }
+
     public long getSinkProcessedCount() {
         return sinkProcessedCount.get();
     }
diff --git a/eventmesh-connector-runtime/src/test/java/org/apache/eventmesh/connector/ConnectorRuntimeTest.java b/eventmesh-connector-runtime/src/test/java/org/apache/eventmesh/connector/ConnectorRuntimeTest.java
index ae63fbd221..1d00e58161 100644
--- a/eventmesh-connector-runtime/src/test/java/org/apache/eventmesh/connector/ConnectorRuntimeTest.java
+++ b/eventmesh-connector-runtime/src/test/java/org/apache/eventmesh/connector/ConnectorRuntimeTest.java
@@ -24,8 +24,10 @@
 import java.net.URI;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Properties;
 import java.util.Set;
 
@@ -142,10 +144,15 @@ private static final class FakeEndpoint implements EventMeshEndpoint {
         final List<CloudEvent> published = new ArrayList<>();
         final Set<String> acked = new HashSet<>();
         String failOn;
+        String throwOnPublish;
+        String throwOnAck;
         List<PollEntry> sinkBatch;
 
         @Override
         public boolean publish(String topic, CloudEvent event) {
+            if (throwOnPublish != null && throwOnPublish.equals(event.getId())) {
+                throw new RuntimeException("simulated publish NPE for " + event.getId());
+            }
             if (failOn != null && failOn.equals(event.getId())) {
                 return false;
             }
@@ -160,7 +167,107 @@ public List<PollEntry> pollForSink(String sinkClientId, int maxEvents, long time
 
         @Override
         public boolean ack(String deliveryId) {
+            if (throwOnAck != null && throwOnAck.equals(deliveryId)) {
+                throw new RuntimeException("simulated ack NPE for " + deliveryId);
+            }
             return acked.add(deliveryId);
         }
     }
+
+    // ---- P0 hardening tests (issues #5231 / #5232 / #5233 follow-up) ----
+
+    @Test
+    void sourcePublishThrowsOnOneEventDoesNotKillBatch() {
+        // #5231-style scenario: publish throws a RuntimeException for one event in the middle of
+        // the batch. Before the fix this would propagate up to runSourceLoop's outer catch and the
+        // rest of the batch would never be attempted. After the fix each event is isolated: the
+        // bad event is logged+skipped, the next events still get a chance to publish, and the
+        // commit advances only to the last accepted event.
+        FakeSource source = new FakeSource(Arrays.asList(event("e1"), event("e2-bad"), event("e3")));
+        FakeEndpoint endpoint = new FakeEndpoint();
+        endpoint.throwOnPublish = "e2-bad";
+        ConnectorRuntime runtime = new ConnectorRuntime(source, endpoint, "orders");
+
+        assertEquals(2, runtime.runSourceOnce(), "e1 + e3 still publish, e2-bad is skipped");
+        assertEquals(2, endpoint.published.size());
+        assertTrue(endpoint.published.stream().anyMatch(e -> "e1".equals(e.getId())));
+        assertTrue(endpoint.published.stream().anyMatch(e -> "e3".equals(e.getId())));
+        assertEquals("e3", source.lastCommitted.getId(), "checkpoint advances past the skipped event");
+        assertEquals(1, runtime.getSourcePublishFailures(), "the failure counter recorded the skip");
+    }
+
+    @Test
+    void sourceNullEventInBatchIsSkippedNotCrash() {
+        // Defensive: if source.poll() returns a list containing null entries (a buggy source impl),
+        // the runtime must skip them rather than NPE.
+        List<CloudEvent> batch = new ArrayList<>();
+        batch.add(event("e1"));
+        batch.add(null);
+        batch.add(event("e3"));
+        FakeSource source = new FakeSource(batch);
+        FakeEndpoint endpoint = new FakeEndpoint();
+        ConnectorRuntime runtime = new ConnectorRuntime(source, endpoint, "orders");
+
+        assertEquals(2, runtime.runSourceOnce());
+        assertEquals("e3", source.lastCommitted.getId());
+        assertEquals(1, runtime.getSourcePublishFailures(), "null event counted as a failure");
+    }
+
+    @Test
+    void sourceOffsetPutFailureDoesNotFailBatch() {
+        // offsetStore.put throws (e.g. RocksDB IO error, Meta CAS lost). The published count and
+        // the source commit must still succeed — we lose the runtime-managed offset but the
+        // event is already on EventMesh (at-least-once).
+        FakeSource source = new FakeSource(Arrays.asList(event("e1"), event("e2")));
+        FakeEndpoint endpoint = new FakeEndpoint();
+        ConnectorRuntime runtime = new ConnectorRuntime(source, endpoint, "orders");
+        runtime.setOffsetStore(new ConnectorOffsetStore() {
+
+            @Override
+            public void put(String key, String value) {
+                throw new RuntimeException("offset store down");
+            }
+
+            @Override
+            public String get(String key) {
+                return null;
+            }
+
+            @Override
+            public Map<String, String> all() {
+                return new HashMap<>();
+            }
+
+            @Override
+            public void flush() {
+            }
+
+            @Override
+            public void close() {
+            }
+        });
+
+        assertEquals(2, runtime.runSourceOnce(), "publish+commit succeed despite offset store failure");
+        assertEquals("e2", source.lastCommitted.getId());
+    }
+
+    @Test
+    void sinkAckThrowsOnOneDeliveryDoesNotLoseOthers() {
+        // Per-delivery ACK isolation: if one ACK throws, the other deliveries in the same batch
+        // still get acked. EventMesh will time out the un-ACKed delivery and redeliver; the sink
+        // must dedup by event id.
+        FakeSink sink = new FakeSink();
+        FakeEndpoint endpoint = new FakeEndpoint();
+        endpoint.throwOnAck = "d-2";
+        endpoint.sinkBatch = Arrays.asList(
+            new PollEntry("d-1", event("e1")),
+            new PollEntry("d-2", event("e2")),
+            new PollEntry("d-3", event("e3")));
+        ConnectorRuntime runtime = new ConnectorRuntime(sink, endpoint, "sink-1", 10, 0L);
+
+        assertEquals(3, runtime.runSinkOnce(), "all 3 events written");
+        assertEquals(2, endpoint.acked.size(), "d-1 + d-3 acked; d-2 lost to be redelivered");
+        assertTrue(endpoint.acked.contains("d-1"));
+        assertTrue(endpoint.acked.contains("d-3"));
+    }
 }
diff --git a/eventmesh-runtime/conf/eventmesh.properties b/eventmesh-runtime/conf/eventmesh.properties
index 81cb6a7664..9315ee604e 100644
--- a/eventmesh-runtime/conf/eventmesh.properties
+++ b/eventmesh-runtime/conf/eventmesh.properties
@@ -52,3 +52,19 @@ eventMesh.server.kafka.namesrvAddr=localhost:9092
 #eventmesh.admin.port=8081
 #eventmesh.ws.port=-1
 #eventmesh.offset.path=./data/offset
+
+
+# ===== Connector Runtime (Experimental — see docs/eventmesh-features.md) =====
+# Connectors run as an independent process (eventmesh-connector-runtime) that bridges external
+# systems and EventMesh over CloudEvents/HTTP. This section is read by that process, not by the
+# EventMesh server itself.
+# Connector worker admin/metrics port on the connector-runtime process:
+#eventmesh.connector.admin.port=8090
+# Long-poll interval for the sink loop (ms):
+#eventmesh.connector.poll.interval.ms=1000
+# Max events pulled per sink batch:
+#eventmesh.connector.sink.max.batch=100
+# Offset storage backend for the connector process: memory | rocksdb | remote
+#eventmesh.connector.offset.store=memory
+# Offset persistence path when offset.store=rocksdb:
+#eventmesh.connector.offset.path=./data/connector-offsets
diff --git a/settings.gradle b/settings.gradle
index 07fa97c6f9..9dbe8273a2 100644
--- a/settings.gradle
+++ b/settings.gradle
@@ -66,6 +66,7 @@ include 'eventmesh-protocol-plugin:eventmesh-protocol-meshmessage'
 include 'eventmesh-protocol-plugin:eventmesh-protocol-a2a'
 
 // Connector Runtime — 独立模块 (与 EventMesh Runtime 经 HTTP 通信, 不互相依赖)
+include 'eventmesh-connector-api'
 include 'eventmesh-connector-runtime'
 
 // Agent — 独立进程, 经 lite topic + CloudEvents 与 Runtime 通信, 调用真实 LLM (OpenAI 兼容)