First Connect SDK

src/config/connect-knet-sink.properties

@@ -0,0 +1,20 @@
+# 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.
+
+name=local-knet-sink
+connector.class=KNetSinkConnector
+tasks.max=1
+classname=MASES.KNetTemplate.KNetConnect.KNetConnectSink, knetConnectSink
+topics=topic-perf

src/config/connect-knet-source.properties

@@ -0,0 +1,20 @@
+# 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.
+
+name=local-knet-source
+connector.class=KNetSourceConnector
+tasks.max=1
+classname=MASES.KNetTemplate.KNetConnect.KNetConnectSource, knetConnectSource
+topic=topic-perf

src/java/knet/pom.xml

@@ -9,7 +9,7 @@
     <name>mases.knet</name>
     <description>Interface bridging implementation for Apache Kafka</description>
     <url>https://github.com/masesgroup/KNet</url>
-    <version>1.2.4.0</version>
+    <version>1.3.0.0</version>
 
     <licenses>
         <license>
@@ -40,7 +40,7 @@
         <maven.compiler.source>8</maven.compiler.source>
         <maven.compiler.target>8</maven.compiler.target>
         <classpathfile>${basedir}/classpathfile.classpath</classpathfile>
-        <kafkaVersion>3.1.0</kafkaVersion>
+        <kafkaVersion>3.1.1</kafkaVersion>
     </properties>
     <distributionManagement>
         <snapshotRepository>
@@ -228,6 +228,11 @@
             <version>4.13.1</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
+            <version>1.7.36</version>
+        </dependency>
         <dependency>
             <groupId>org.apache.kafka</groupId>
             <artifactId>kafka-clients</artifactId>

src/java/knet/src/main/java/org/mases/knet/connect/KNetConnectProxy.java

@@ -0,0 +1,89 @@
+/*
+ *  Copyright 2022 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.connect;
+
+import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigException;
+import org.mases.jcobridge.*;
+import org.mases.knet.connect.source.KNetSourceTask;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Map;
+
+public class KNetConnectProxy {
+    private static final Logger log = LoggerFactory.getLogger(KNetSourceTask.class);
+
+    public static final String DOTNET_CLASSNAME_CONFIG = "classname";
+
+    public static final ConfigDef CONFIG_DEF = new ConfigDef()
+            .define(DOTNET_CLASSNAME_CONFIG, ConfigDef.Type.STRING, null, ConfigDef.Importance.HIGH, ".NET class name.");
+
+    static JCObject knetConnectProxy = null;
+    static String sinkConnectorName = null;
+    static JCObject sinkConnector = null;
+    static String sourceConnectorName = null;
+    static JCObject sourceConnector = null;
+
+    static synchronized JCObject getConnectProxy() throws JCException, IOException {
+        if (knetConnectProxy == null) {
+            JCOBridge.Initialize();
+            knetConnectProxy = (JCObject) JCOBridge.GetCLRGlobal("KNetConnectProxy");
+        }
+        return knetConnectProxy;
+    }
+
+    public static synchronized boolean initializeSinkConnector(Map<String, String> props) throws JCException, IOException {
+        AbstractConfig parsedConfig = new AbstractConfig(CONFIG_DEF, props);
+        String className = parsedConfig.getString(DOTNET_CLASSNAME_CONFIG);
+        if (className == null)
+            throw new ConfigException("'classname' in KNetSinkConnector configuration requires a definition");
+        return (boolean) getConnectProxy().Invoke("AllocateSinkConnector", className);
+    }
+
+    public static synchronized boolean initializeSourceConnector(Map<String, String> props) throws JCException, IOException {
+        AbstractConfig parsedConfig = new AbstractConfig(CONFIG_DEF, props);
+        String className = parsedConfig.getString(DOTNET_CLASSNAME_CONFIG);
+        if (className == null)
+            throw new ConfigException("'classname' in KNetSinkConnector configuration requires a definition");
+        return (boolean) getConnectProxy().Invoke("AllocateSourceConnector", className);
+    }
+
+    public static synchronized JCObject getSinkConnector() throws JCException, IOException {
+        if (sinkConnector == null) {
+            sinkConnectorName = (String) getConnectProxy().Invoke("SinkConnectorName");
+            if (sinkConnectorName != null) {
+                sinkConnector = (JCObject) JCOBridge.GetCLRGlobal(sinkConnectorName);
+            }
+        }
+        return sinkConnector;
+    }
+
+    public static synchronized JCObject getSourceConnector() throws JCException, IOException {
+        if (sourceConnector == null) {
+            sourceConnectorName = (String) getConnectProxy().Invoke("SourceConnectorName");
+            if (sourceConnectorName != null) {
+                sourceConnector = (JCObject) JCOBridge.GetCLRGlobal(sourceConnectorName);
+            }
+        }
+        return sourceConnector;
+    }
+}

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

@@ -0,0 +1,122 @@
+/*
+ *  Copyright 2022 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.connect.sink;
+
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.connect.connector.Task;
+import org.apache.kafka.connect.sink.SinkConnector;
+import org.mases.jcobridge.*;
+import org.mases.knet.connect.KNetConnectProxy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class KNetSinkConnector extends SinkConnector {
+    private static final Logger log = LoggerFactory.getLogger(KNetSinkConnector.class);
+
+    Object dataToExchange = null;
+
+    public Object getDataToExchange() {
+        return dataToExchange;
+    }
+
+    public void setDataToExchange(Object dte) {
+        dataToExchange = dte;
+    }
+
+    @Override
+    public void start(Map<String, String> props) {
+        try {
+            if (!KNetConnectProxy.initializeSinkConnector(props)) {
+                log.error("Failed Invoke of \"initializeSinkConnector\"");
+                throw new ConfigException("Failed Invoke of \"initializeSinkConnector\"");
+            } else {
+                JCOBridge.RegisterJVMGlobal("KNetSinkConnector", this);
+                try {
+                    dataToExchange = props;
+                    KNetConnectProxy.getSinkConnector().Invoke("StartInternal");
+                } finally {
+                    dataToExchange = null;
+                }
+            }
+        } catch (JCException | IOException jcne) {
+            log.error("Failed Invoke of \"start\"", jcne);
+        }
+    }
+
+    @Override
+    public Class<? extends Task> taskClass() {
+        return KNetSinkTask.class;
+    }
+
+    @Override
+    public List<Map<String, String>> taskConfigs(int maxTasks) {
+        ArrayList<Map<String, String>> configs = new ArrayList<>();
+        for (int i = 0; i < maxTasks; i++) {
+            Map<String, String> config = new HashMap<>();
+            try {
+                dataToExchange = config;
+                KNetConnectProxy.getSinkConnector().Invoke("TaskConfigsInternal", i);
+            } catch (JCException | IOException jcne) {
+                log.error("Failed Invoke of \"start\"", jcne);
+            } finally {
+                dataToExchange = null;
+            }
+            configs.add(config);
+        }
+        return configs;
+    }
+
+    @Override
+    public void stop() {
+        try {
+            try {
+                KNetConnectProxy.getSinkConnector().Invoke("StopInternal");
+            } finally {
+                JCOBridge.UnregisterJVMGlobal("KNetSinkConnector");
+            }
+        } catch (JCException | IOException jcne) {
+            log.error("Failed Invoke of \"stop\"", jcne);
+        }
+    }
+
+    @Override
+    public ConfigDef config() {
+        return KNetConnectProxy.CONFIG_DEF;
+    }
+
+    @Override
+    public String version() {
+        try {
+            JCObject sink = KNetConnectProxy.getSinkConnector();
+            if (sink != null) {
+                return (String) sink.Invoke("VersionInternal");
+            }
+        } catch (JCException | IOException jcne) {
+            log.error("Failed Invoke of \"version\"", jcne);
+        }
+        return "NOT AVAILABLE";
+    }
+}

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

@@ -0,0 +1,105 @@
+/*
+ *  Copyright 2022 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.connect.sink;
+
+import org.apache.kafka.connect.sink.SinkRecord;
+import org.apache.kafka.connect.sink.SinkTask;
+import org.apache.kafka.connect.sink.SinkTaskContext;
+import org.mases.jcobridge.*;
+import org.mases.knet.connect.KNetConnectProxy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
+
+public class KNetSinkTask extends SinkTask {
+    private static final Logger log = LoggerFactory.getLogger(KNetSinkTask.class);
+
+    static final AtomicLong taskId = new AtomicLong(0);
+    JCObject sinkTask = null;
+
+    Object dataToExchange = null;
+
+    public Object getDataToExchange() {
+        return dataToExchange;
+    }
+
+    public void setDataToExchange(Object dte) {
+        dataToExchange = dte;
+    }
+
+    public KNetSinkTask() throws JCException, IOException {
+        super();
+        long taskid = taskId.incrementAndGet();
+        JCOBridge.RegisterJVMGlobal(String.format("KNetSinkTask_%d", taskid), this);
+        sinkTask = (JCObject) KNetConnectProxy.getSinkConnector().Invoke("AllocateTask", taskid);
+    }
+
+    @Override
+    public String version() {
+        try {
+            if (sinkTask != null) {
+                return (String) sinkTask.Invoke("VersionInternal");
+            }
+        } catch (JCNativeException jcne) {
+            log.error("Failed Invoke of \"version\"", jcne);
+        }
+        return "NOT AVAILABLE";
+    }
+
+    @Override
+    public void start(Map<String, String> map) {
+        try {
+            try {
+                dataToExchange = map;
+                sinkTask.Invoke("StartInternal");
+            } finally {
+                dataToExchange = null;
+            }
+        } catch (JCNativeException jcne) {
+            log.error("Failed Invoke of \"start\"", jcne);
+        }
+    }
+
+    @Override
+    public void put(Collection<SinkRecord> collection) {
+        try {
+            try {
+                dataToExchange = collection;
+                sinkTask.Invoke("PutInternal");
+            } finally {
+                dataToExchange = null;
+            }
+        } catch (JCNativeException jcne) {
+            log.error("Failed Invoke of \"put\"", jcne);
+        }
+    }
+
+    @Override
+    public void stop() {
+        try {
+            sinkTask.Invoke("StopInternal");
+        } catch (JCNativeException jcne) {
+            log.error("Failed Invoke of \"stop\"", jcne);
+        }
+    }
+}