Add data-exchange interface and async source support (#1335)

* Add data-exchange interface and async source support

Introduce KNetConnectDataExchange for CLR/JVM object exchange and implement it across connectors, tasks, transformations and predicates. Enhance KNetSourceTask with async record push support (PushRecordAsync, CreateAndPushRecordAsync), UseOnlyAsync flag and poll logic to consume an internal queue. Refactor .NET bridge: add IKNetCommon, make remote ExecuteOnRemote/DataToExchange use a unified reflectedConnectorOrTask, expose internal Start/TaskConfigs/Stop entrypoints, add DeallocateTask, and introduce several IKNet* interfaces and API visibility changes to support the new async/out-of-sync behavior and cleaner interop.

* Update documentation

* Fix #1321 updating missing changes in #1325

Author: masesdevelopers

src/documentation/articles/connectSDK.md

@@ -14,6 +14,9 @@ This is only a quick start guide for KNet Connect SDK, other information related
 
 The code for a connector based on KNet Connect SDK follows the same rules for both .NET and the JVM hosted runtimes.
 
+> [!TIP]
+> Except for some specific method signature or additional properties, the rules used in a JVM developed connectors/transformer/predicate applies to KNet Connect SDK connectors/transformer/predicate.
+
 ### Source connector
 
 A source connector shall be defined extending the following class:
@@ -35,13 +38,18 @@ The mandatory method to be implemented is:
 ```c#
 public abstract System.Collections.Generic.IList<SourceRecord> Poll();
 ```
-which returns a set of `SourceRecord`, each `SourceRecord` can be created directly or by using the `CreateRecord` helper methods available.
+which returns a set of `SourceRecord`, each `SourceRecord` can be created directly or by using the available `CreateRecord` helper methods.
 
 > [!TIP]
 > Starting from KNet 3.0.3, beside the standard invocation where the `SourceRecord`s will be returned once at the end of the `Poll` invocation and then pushed to the JVM,
 by using the `CreateAndPushRecord` helper methods available each `SourceRecord` is created and directly pushed to the JVM: 
 this new way can be used to take advantage of the idle time if the `KNetSourceTask<TTask>` is waiting to receive data, e.g. socket, disk access, etc.
 
+> [!TIP]
+> KNet 3.1.2 introduces, beside the `CreateAndPushRecord` helper methods, some new `CreateAndPushRecordAsync` helper methods which can push records in the JVM outside the `Poll` invocation:
+these new helpers decouple `Poll` invocation from `SourceRecord` generation and can be useful if the .NET code is taking adavantage of async/await pattern or similar; if `UseOnlyAsync` is set to true 
+the `Poll` method is never invoked and everything happens in async from the .NET point-of-view (JVM side continues to invoke the `poll` method, but the returned list was filled in the meantime).
+
 ### Sink connector
 
 A sink connector shall be defined extending the following class:
@@ -146,7 +154,7 @@ knet.dotnet.classname=MyConnectorNamespace.MySourceConnector, MySourceConnectorA
 ```
 
 From KNet 3.1.2, the JVM side exposes two new methods `getAssemblyLocation` and `getClassName`. 
-Overriding them in a child class of 
+Overriding the methods in a child class of 
 - `org.mases.knet.developed.connect.sink.KNetSinkConnector`
 - `org.mases.knet.developed.connect.source.KNetSourceConnector`
 - `org.mases.knet.developed.connect.transforms.KNetTransformation`

src/jvm/knet/src/main/java/org/mases/knet/developed/connect/KNetConnectDataExchange.java

@@ -0,0 +1,34 @@
+/*
+ *  Copyright (c) 2021-2025 MASES s.r.l.
+ *
+ *  Licensed 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.
+ *
+ *  Refer to LICENSE for more information.
+ */
+
+package org.mases.knet.developed.connect;
+
+/*
+ * All KNet Connect SDK object implements this interface for data exchange between JVM and CLR
+ */
+public interface KNetConnectDataExchange {
+    /* Invoked from CLR to retrieve the object stored from CLR before a method invocation
+     * @return the object stored from JVM can be used from CLR
+     */
+    Object getDataToExchange();
+
+    /* Set the object from CLR to be used from the JVM
+     * @param dte the object set from CLR
+     */
+    void setDataToExchange(Object dte);
+}

src/jvm/knet/src/main/java/org/mases/knet/developed/connect/KNetConnectInitializer.java

@@ -18,12 +18,16 @@
 
 package org.mases.knet.developed.connect;
 
+/*
+ * All KNet Connect SDK main object implements this interface to define mandatory information used from infrastructure
+ */
 public interface KNetConnectInitializer {
     /* Overrides in a derived class to define the Assembly location
      *
      * @returns the location where the assembly is available, return null to use the configuration property
      */
     String getAssemblyLocation();
+
     /* Overrides in a derived class to define the .NET class to be loaded
      *
      * @returns the location where the assembly is available, return null to use the configuration property

src/jvm/knet/src/main/java/org/mases/knet/developed/connect/sink/KNetSinkConnector.java

@@ -25,6 +25,7 @@
 import org.apache.kafka.connect.sink.SinkConnector;
 import org.apache.kafka.connect.sink.SinkConnectorContext;
 import org.mases.jcobridge.*;
+import org.mases.knet.developed.connect.KNetConnectDataExchange;
 import org.mases.knet.developed.connect.KNetConnectInitializer;
 import org.mases.knet.developed.connect.KNetConnectLogging;
 import org.mases.knet.developed.connect.KNetConnectProxy;
@@ -37,7 +38,7 @@
 import java.util.List;
 import java.util.Map;
 
-public class KNetSinkConnector extends SinkConnector implements KNetConnectLogging, KNetConnectInitializer {
+public class KNetSinkConnector extends SinkConnector implements KNetConnectLogging, KNetConnectInitializer, KNetConnectDataExchange {
     private static final Logger log = LoggerFactory.getLogger(KNetSinkConnector.class);
 
     private static final String registrationName = "KNetSinkConnector";

src/jvm/knet/src/main/java/org/mases/knet/developed/connect/sink/KNetSinkTask.java

@@ -23,6 +23,7 @@
 import org.apache.kafka.connect.sink.SinkTask;
 import org.apache.kafka.connect.sink.SinkTaskContext;
 import org.mases.jcobridge.*;
+import org.mases.knet.developed.connect.KNetConnectDataExchange;
 import org.mases.knet.developed.connect.KNetConnectLogging;
 import org.mases.knet.developed.connect.KNetConnectProxy;
 import org.slf4j.Logger;
@@ -32,7 +33,7 @@
 import java.util.Collection;
 import java.util.Map;
 
-public class KNetSinkTask extends SinkTask implements KNetConnectLogging {
+public class KNetSinkTask extends SinkTask implements KNetConnectLogging, KNetConnectDataExchange {
     private static final Logger log = LoggerFactory.getLogger(KNetSinkTask.class);
 
     private static final String registrationName = "KNetSinkTask";

src/jvm/knet/src/main/java/org/mases/knet/developed/connect/source/KNetSourceConnector.java

@@ -24,6 +24,7 @@
 import org.apache.kafka.connect.errors.ConnectException;
 import org.apache.kafka.connect.source.*;
 import org.mases.jcobridge.*;
+import org.mases.knet.developed.connect.KNetConnectDataExchange;
 import org.mases.knet.developed.connect.KNetConnectInitializer;
 import org.mases.knet.developed.connect.KNetConnectLogging;
 import org.mases.knet.developed.connect.KNetConnectProxy;
@@ -36,7 +37,7 @@
 import java.util.List;
 import java.util.Map;
 
-public class KNetSourceConnector extends SourceConnector implements KNetConnectLogging, KNetConnectInitializer {
+public class KNetSourceConnector extends SourceConnector implements KNetConnectLogging, KNetConnectInitializer, KNetConnectDataExchange {
     private static final Logger log = LoggerFactory.getLogger(KNetSourceConnector.class);
 
     private static final String registrationName = "KNetSourceConnector";

src/jvm/knet/src/main/java/org/mases/knet/developed/connect/source/KNetSourceTask.java

@@ -23,6 +23,7 @@
 import org.apache.kafka.connect.source.SourceTask;
 import org.apache.kafka.connect.source.SourceTaskContext;
 import org.mases.jcobridge.*;
+import org.mases.knet.developed.connect.KNetConnectDataExchange;
 import org.mases.knet.developed.connect.KNetConnectLogging;
 import org.mases.knet.developed.connect.KNetConnectProxy;
 import org.slf4j.Logger;
@@ -31,7 +32,7 @@
 import java.io.IOException;
 import java.util.*;
 
-public class KNetSourceTask extends SourceTask implements KNetConnectLogging {
+public class KNetSourceTask extends SourceTask implements KNetConnectLogging, KNetConnectDataExchange {
     private static final Logger log = LoggerFactory.getLogger(KNetSourceTask.class);
 
     private static final String registrationName = "KNetSourceTask";
@@ -44,6 +45,9 @@ public class KNetSourceTask extends SourceTask implements KNetConnectLogging {
 
     Object dataToExchange = null;
 
+    boolean _useOnlyAsyncEnable = false;
+    List<SourceRecord> _sourceList;
+
     public Object getDataToExchange() {
         return dataToExchange;
     }
@@ -57,6 +61,7 @@ public KNetSourceTask() throws ConnectException, JCException, IOException {
         log.debug("Invoking ctor of KNetSourceTask");
         taskId = KNetConnectProxy.getNewTaskId();
         indexedRegistrationName = String.format("%s_%d", registrationName, taskId);
+        _sourceList = new ArrayList<>();
         if (JCOBridge.isCLRHostingProcess()) {
             JCOBridge.RegisterJVMGlobal(indexedRegistrationName, this);
             JCObject source = KNetConnectProxy.getSourceConnector();
@@ -102,27 +107,59 @@ public void start(Map<String, String> map) {
             } finally {
                 dataToExchange = null;
             }
+            requestUseOnlyAsync();
         } catch (JCException | IOException jcne) {
             log.error("Failed Invoke of \"start\"", jcne);
             throw new ConnectException("Failed Invoke of \"start\"", jcne);
         }
     }
 
+    public void requestUseOnlyAsync() {
+        try {
+            try {
+                dataToExchange = _useOnlyAsyncEnable;
+                sourceTask.Invoke("UseOnlyAsyncInternal");
+                _useOnlyAsyncEnable = (boolean) dataToExchange;
+            } finally {
+                dataToExchange = null;
+            }
+        } catch (JCNativeException jcne) {
+            log.error("Failed Invoke of \"UseOnlyAsyncInternal\"", jcne);
+        }
+    }
+
+    public void pushRecordAsync(SourceRecord record) {
+        synchronized (this) {
+            _sourceList.add(record);
+        }
+    }
+
     @Override
     public List<SourceRecord> poll() throws InterruptedException {
         log.debug("Invoking poll");
-        try {
+        synchronized (this) {
+            int size = 0;
             try {
-                dataToExchange = new ArrayList<SourceRecord>();
-                sourceTask.Invoke("PollInternal");
-                return (List<SourceRecord>) dataToExchange;
+                if (_useOnlyAsyncEnable) {
+                    if (_sourceList.isEmpty()) return null;
+                } else {
+                    try {
+                        dataToExchange = _sourceList;
+                        sourceTask.Invoke("PollInternal");
+                    } catch (JCNativeException jcne) {
+                        log.error("Failed Invoke of \"PollInternal\"", jcne);
+                    }
+                }
+                size = _sourceList.size();
+                return (size != 0) ? _sourceList : null;
             } finally {
                 dataToExchange = null;
+                if (size != 0) {
+                    log.debug("Current size: {} - next list capacity will use this value.", size);
+                    _sourceList = new ArrayList<>(size);
+                }
             }
-        } catch (JCNativeException jcne) {
-            log.error("Failed Invoke of \"poll\"", jcne);
         }
-        return null;
     }
 
     public <V> Map<String, Object> offsetAt(String key, V value) {

src/jvm/knet/src/main/java/org/mases/knet/developed/connect/transforms/KNetTransformation.java

@@ -23,6 +23,7 @@
 import org.apache.kafka.connect.components.Versioned;
 import org.apache.kafka.connect.connector.ConnectRecord;
 import org.apache.kafka.connect.transforms.Transformation;
+import org.mases.knet.developed.connect.KNetConnectDataExchange;
 import org.mases.knet.developed.connect.KNetConnectInitializer;
 import org.mases.knet.developed.connect.KNetConnectLogging;
 import org.mases.knet.developed.connect.KNetConnectProxy;
@@ -33,7 +34,7 @@
 import java.io.IOException;
 import java.util.Map;
 
-public class KNetTransformation<R extends ConnectRecord<R>> implements Transformation<R>, Versioned, KNetConnectLogging, KNetConnectInitializer {
+public class KNetTransformation<R extends ConnectRecord<R>> implements Transformation<R>, Versioned, KNetConnectLogging, KNetConnectInitializer, KNetConnectDataExchange {
     private static final Logger log = LoggerFactory.getLogger(KNetTransformation.class);
 
     private static final String registrationName = "KNetTransformation";

src/jvm/knet/src/main/java/org/mases/knet/developed/connect/transforms/predicates/KNetPredicate.java

@@ -23,9 +23,7 @@
 import org.apache.kafka.connect.components.Versioned;
 import org.apache.kafka.connect.connector.ConnectRecord;
 import org.apache.kafka.connect.transforms.predicates.Predicate;
-import org.mases.knet.developed.connect.KNetConnectInitializer;
-import org.mases.knet.developed.connect.KNetConnectLogging;
-import org.mases.knet.developed.connect.KNetConnectProxy;
+import org.mases.knet.developed.connect.*;
 import org.mases.jcobridge.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -35,7 +33,7 @@
 
 import org.mases.knet.developed.connect.KNetConnectInitializer;
 
-public class KNetPredicate<R extends ConnectRecord<R>> implements Predicate<R>, Versioned, KNetConnectLogging, KNetConnectInitializer {
+public class KNetPredicate<R extends ConnectRecord<R>> implements Predicate<R>, Versioned, KNetConnectLogging, KNetConnectInitializer, KNetConnectDataExchange {
     private static final Logger log = LoggerFactory.getLogger(KNetPredicate.class);
 
     private static final String registrationName = "KNetPredicate";

src/jvm/knet/src/main/resources/META-INF/services/org.apache.kafka.connect.transforms.Transformation

@@ -0,0 +1 @@
+org.mases.knet.developed.connect.transforms.KNetTransformation