[ISSUE #5231#5232#5233] Harden ConnectorRuntime against data loss; split connector-api SPI module; add plugin unit tests (#4642)

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')

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.

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') {

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");
 }

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");
+    }
 }

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;
+    }
+}

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'
+}

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;

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'

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'

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'

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'

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;
+        }
+    }
 }

eventmesh-connector-plugin/eventmesh-connector-file/src/main/java/org/apache/eventmesh/connector/file/source/FileSourceConnector.java

@@ -38,13 +38,27 @@ 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) {
             throw new RuntimeException(e);
         }
     }
 
+    public void closeReaderQuietly() {
+        if (reader != null) {
+            try {
+                reader.close();
+            } catch (Exception ignored) {
+                // best-effort
+            }
+            reader = null;
+        }
+    }
+
     @Override
     public List<CloudEvent> poll() {
         if (reader == null) {