fix(connectors): reject duplicate SDK open calls

core/connectors/sdk/src/sink.rs

@@ -236,6 +236,13 @@ macro_rules! sink_connector {
             config_len: usize,
             log_callback: LogCallback,
         ) -> i32 {
+            if INSTANCES.contains_key(&id) {
+                // Duplicate id: caller did not close before reopening. Without
+                // this guard the existing entry would be silently overwritten,
+                // discarding any in-flight buffered data and orphaning tasks.
+                return -1;
+            }
+
             let mut container = SinkContainer::new(id);
             let result = container.open(id, config_ptr, config_len, log_callback, <$type>::new);
             INSTANCES.insert(id, container);

core/connectors/sdk/src/source.rs

@@ -226,6 +226,13 @@ macro_rules! source_connector {
             state_len: usize,
             log_callback: LogCallback,
         ) -> i32 {
+            if INSTANCES.contains_key(&id) {
+                // Duplicate id: caller did not close before reopening. Without
+                // this guard the existing entry would be silently overwritten,
+                // discarding any in-flight buffered data and orphaning tasks.
+                return -1;
+            }
+
             let mut container = SourceContainer::new(id);
             let result = container.open(
                 id,