Azure Servicebus Source Connector

java-connectors/.gitignore

@@ -1,5 +1,6 @@
 .gradle
 build/
+target/
 !gradle/wrapper/gradle-wrapper.jar
 !**/src/main/**/build/
 !**/src/test/**/build/
@@ -45,4 +46,4 @@ bin/
 ### Lenses-specific ###
 release/
 gradle-modules.txt
-**/src/main/gen/*
+gradle-test-modules.txt

java-connectors/build.gradle

@@ -8,7 +8,7 @@ plugins {
 allprojects {
 
     group = "io.lenses.streamreactor"
-    version = "7.0.1-SNAPSHOT"
+    version = "7.2.1-SNAPSHOT"
     description = "stream-reactor"
 
     apply plugin: 'java'
@@ -23,7 +23,7 @@ allprojects {
 
     ext {
         //DEPENDENCY VERSIONS
-        lombokVersion = '1.18.30'
+        lombokVersion = '1.18.32'
         kafkaVersion = '3.7.0'
         logbackVersion = '1.4.14'
         jUnitVersion = '5.9.1'

java-connectors/kafka-connect-azure-eventhubs/src/main/java/io/lenses/streamreactor/connect/azure/eventhubs/source/AzureEventHubsSourceConnector.java

@@ -38,7 +38,7 @@
 public class AzureEventHubsSourceConnector extends SourceConnector {
 
   private final JarManifest jarManifest =
-      new JarManifest(getClass().getProtectionDomain().getCodeSource().getLocation());
+      JarManifest.produceFromClass(getClass());
   private Map<String, String> configProperties;
 
   @Override

java-connectors/kafka-connect-azure-eventhubs/src/main/java/io/lenses/streamreactor/connect/azure/eventhubs/source/AzureEventHubsSourceTask.java

@@ -46,7 +46,7 @@ public class AzureEventHubsSourceTask extends SourceTask {
   private BlockingQueueProducerProvider blockingQueueProducerProvider;
 
   public AzureEventHubsSourceTask() {
-    jarManifest = new JarManifest(getClass().getProtectionDomain().getCodeSource().getLocation());
+    jarManifest = JarManifest.produceFromClass(getClass());
   }
 
   public AzureEventHubsSourceTask(JarManifest jarManifest) {

java-connectors/kafka-connect-azure-servicebus/build.gradle

@@ -0,0 +1,18 @@
+project(':kafka-connect-azure-servicebus') {
+
+
+    test {
+        maxParallelForks = 1
+    }
+
+    dependencies {
+        implementation project(':kafka-connect-common')
+        implementation project(':kafka-connect-query-language')
+
+        // Azure Service Bus dependencies
+        implementation group: 'com.azure', name: 'azure-core', version: '1.49.0'
+        implementation group: 'com.azure', name: 'azure-messaging-servicebus', version: '7.17.0'
+
+
+    }
+}

java-connectors/kafka-connect-azure-servicebus/src/main/java/io/lenses/streamreactor/connect/azure/servicebus/config/AzureServiceBusConfigConstants.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2017-2024 Lenses.io Ltd
+ *
+ * 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.
+ */
+package io.lenses.streamreactor.connect.azure.servicebus.config;
+
+import io.lenses.streamreactor.connect.azure.servicebus.source.AzureServiceBusSourceConnector;
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+/**
+ * Class to indicate String constants used in Service Bus Connectors Config.
+ */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class AzureServiceBusConfigConstants {
+
+  private static final String DOT = ".";
+  public static final String CONNECTOR_PREFIX = "connect.servicebus";
+  public static final String SOURCE_CONNECTOR_PREFIX = CONNECTOR_PREFIX + DOT + "source";
+
+  public static final String CONNECTOR_NAME = "name";
+  public static final String CONNECTOR_NAME_DOC = "Connector's name";
+  public static final String CONNECTOR_NAME_DEFAULT = AzureServiceBusSourceConnector.class.getSimpleName();
+  public static final String CONNECTION_STRING = CONNECTOR_PREFIX + DOT + "connection.string";
+  public static final String CONNECTION_STRING_DOC = "Azure ServiceBus Connection String";
+  public static final String KCQL_CONFIG = CONNECTOR_PREFIX + DOT + "kcql";
+  public static final String KCQL_DOC =
+      "KCQL expression describing field selection and data routing to the target.";
+
+  public static final String TASK_RECORDS_QUEUE_SIZE = SOURCE_CONNECTOR_PREFIX + "task.records.queue.size";
+  public static final String TASK_RECORDS_QUEUE_SIZE_DOC = "Task's records queue size.";
+  public static final int TASK_RECORDS_QUEUE_SIZE_DEFAULT = 20;
+
+}

java-connectors/kafka-connect-azure-servicebus/src/main/java/io/lenses/streamreactor/connect/azure/servicebus/config/AzureServiceBusSourceConfig.java

@@ -0,0 +1,87 @@
+/*
+ * Copyright 2017-2024 Lenses.io Ltd
+ *
+ * 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.
+ */
+package io.lenses.streamreactor.connect.azure.servicebus.config;
+
+import io.lenses.streamreactor.common.config.base.BaseConfig;
+import io.lenses.streamreactor.common.config.base.intf.ConnectorPrefixed;
+import java.util.Map;
+import lombok.Getter;
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigDef.Importance;
+import org.apache.kafka.common.config.ConfigDef.Type;
+import org.apache.kafka.common.config.ConfigDef.Width;
+
+/**
+ * Service Bus Connector Configuration class.
+ */
+public class AzureServiceBusSourceConfig extends BaseConfig implements ConnectorPrefixed {
+
+  @Getter
+  private static final ConfigDef configDefinition;
+
+  public static final String CONNECTION_GROUP = "Connection";
+  public static final String BASE_GROUP = "Base";
+
+  static {
+    configDefinition =
+        new ConfigDef()
+            .define(AzureServiceBusConfigConstants.CONNECTOR_NAME,
+                Type.STRING,
+                AzureServiceBusConfigConstants.CONNECTOR_NAME_DEFAULT,
+                Importance.HIGH,
+                AzureServiceBusConfigConstants.CONNECTOR_NAME_DOC,
+                BASE_GROUP,
+                1,
+                ConfigDef.Width.LONG,
+                AzureServiceBusConfigConstants.CONNECTOR_NAME
+            ).define(AzureServiceBusConfigConstants.TASK_RECORDS_QUEUE_SIZE,
+                Type.INT,
+                AzureServiceBusConfigConstants.TASK_RECORDS_QUEUE_SIZE_DEFAULT,
+                Importance.MEDIUM,
+                AzureServiceBusConfigConstants.TASK_RECORDS_QUEUE_SIZE_DOC,
+                BASE_GROUP,
+                2,
+                Width.SHORT,
+                AzureServiceBusConfigConstants.TASK_RECORDS_QUEUE_SIZE
+            ).define(AzureServiceBusConfigConstants.CONNECTION_STRING,
+                Type.STRING,
+                Importance.HIGH,
+                AzureServiceBusConfigConstants.CONNECTION_STRING_DOC,
+                CONNECTION_GROUP,
+                2,
+                ConfigDef.Width.LONG,
+                AzureServiceBusConfigConstants.CONNECTION_STRING
+            ).define(AzureServiceBusConfigConstants.KCQL_CONFIG,
+                Type.STRING,
+                Importance.HIGH,
+                AzureServiceBusConfigConstants.KCQL_DOC,
+                "Mappings",
+                1,
+                ConfigDef.Width.LONG,
+                AzureServiceBusConfigConstants.KCQL_CONFIG
+            );
+  }
+
+  public AzureServiceBusSourceConfig(Map<?, ?> properties) {
+    super(AzureServiceBusConfigConstants.CONNECTOR_PREFIX, getConfigDefinition(), properties);
+  }
+
+  @Override
+  public String connectorPrefix() {
+    return connectorPrefix;
+  }
+
+}

java-connectors/kafka-connect-azure-servicebus/src/main/java/io/lenses/streamreactor/connect/azure/servicebus/mapping/AzureServiceBusSourceRecord.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2017-2024 Lenses.io Ltd
+ *
+ * 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.
+ */
+package io.lenses.streamreactor.connect.azure.servicebus.mapping;
+
+import java.util.Map;
+import lombok.ToString;
+import org.apache.kafka.connect.data.Schema;
+import org.apache.kafka.connect.header.ConnectHeaders;
+import org.apache.kafka.connect.source.SourceRecord;
+
+/**
+ * Implementation of {@link SourceRecord} for Microsoft Azure EventHubs.
+ */
+@ToString(callSuper = true)
+public class AzureServiceBusSourceRecord extends SourceRecord {
+
+  public AzureServiceBusSourceRecord(Map<String, ?> sourcePartition, Map<String, ?> sourceOffset,
+      String topic, Object key, Schema valueSchema, Object value, Long timestamp) {
+    super(sourcePartition, sourceOffset, topic, null,
+        Schema.STRING_SCHEMA, key, valueSchema, value, timestamp, new ConnectHeaders());
+  }
+}

java-connectors/kafka-connect-azure-servicebus/src/main/java/io/lenses/streamreactor/connect/azure/servicebus/mapping/ServiceBusToSourceRecordMapper.java

@@ -0,0 +1,98 @@
+/*
+ * Copyright 2017-2024 Lenses.io Ltd
+ *
+ * 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.
+ */
+package io.lenses.streamreactor.connect.azure.servicebus.mapping;
+
+import static io.lenses.streamreactor.connect.azure.servicebus.mapping.ServiceBusValueSchemaField.CONTENT_TYPE;
+import static io.lenses.streamreactor.connect.azure.servicebus.mapping.ServiceBusValueSchemaField.CORRELATION_ID;
+import static io.lenses.streamreactor.connect.azure.servicebus.mapping.ServiceBusValueSchemaField.DEAD_LETTER_SOURCE;
+import static io.lenses.streamreactor.connect.azure.servicebus.mapping.ServiceBusValueSchemaField.DELIVERY_COUNT;
+import static io.lenses.streamreactor.connect.azure.servicebus.mapping.ServiceBusValueSchemaField.ENQUEUED_TIME_UTC;
+import static io.lenses.streamreactor.connect.azure.servicebus.mapping.ServiceBusValueSchemaField.GET_TO;
+import static io.lenses.streamreactor.connect.azure.servicebus.mapping.ServiceBusValueSchemaField.LABEL;
+import static io.lenses.streamreactor.connect.azure.servicebus.mapping.ServiceBusValueSchemaField.LOCKED_UNTIL_UTC;
+import static io.lenses.streamreactor.connect.azure.servicebus.mapping.ServiceBusValueSchemaField.LOCK_TOKEN;
+import static io.lenses.streamreactor.connect.azure.servicebus.mapping.ServiceBusValueSchemaField.MESSAGE_BODY;
+import static io.lenses.streamreactor.connect.azure.servicebus.mapping.ServiceBusValueSchemaField.PARTITION_KEY;
+import static io.lenses.streamreactor.connect.azure.servicebus.mapping.ServiceBusValueSchemaField.REPLY_TO;
+import static io.lenses.streamreactor.connect.azure.servicebus.mapping.ServiceBusValueSchemaField.REPLY_TO_SESSION_ID;
+import static io.lenses.streamreactor.connect.azure.servicebus.mapping.ServiceBusValueSchemaField.SEQUENCE_NUMBER;
+import static io.lenses.streamreactor.connect.azure.servicebus.mapping.ServiceBusValueSchemaField.SESSION_ID;
+import static io.lenses.streamreactor.connect.azure.servicebus.mapping.ServiceBusValueSchemaField.TIME_TO_LIVE;
+
+import com.azure.messaging.servicebus.ServiceBusReceivedMessage;
+import io.lenses.streamreactor.connect.azure.servicebus.source.AzureServiceBusSourceConnector;
+import java.util.Map;
+import org.apache.kafka.connect.data.Schema;
+import org.apache.kafka.connect.data.SchemaBuilder;
+import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.source.SourceRecord;
+
+/**
+ * Class that maps {@link ServiceBusReceivedMessage} to Kafka Connect {@link SourceRecord}.
+ */
+public class ServiceBusToSourceRecordMapper {
+
+  public static final Schema VALUE_SCHEMA;
+
+  static {
+    SchemaBuilder structSchemaBuilder = SchemaBuilder.struct();
+    ServiceBusValueSchemaField.getAllFields()
+        .forEach(field -> structSchemaBuilder.field(field.name(), field.schema()));
+    VALUE_SCHEMA = structSchemaBuilder.build();
+  }
+
+  private ServiceBusToSourceRecordMapper() {
+  }
+
+  /**
+   * Method to make SourceRecord out of ServiceBusReceivedMessage with key and body as structure.
+   *
+   * @param serviceBusMessage original Service Bus message
+   * @param outputTopic       Output topic for record
+   * @param partitionKey      AzureTopicPartitionKey to indicate topic and partition
+   * @param offsetMap         AzureOffsetMarker to indicate offset
+   * @return mapped SourceRecord
+   */
+  public static SourceRecord mapSingleServiceBusMessage(ServiceBusReceivedMessage serviceBusMessage, String outputTopic,
+      Map<String, String> partitionKey, Map<String, Object> offsetMap) {
+    String key = serviceBusMessage.getMessageId();
+
+    Struct valueObject = createStructFromServiceBusMessage(serviceBusMessage);
+    return new AzureServiceBusSourceRecord(partitionKey, offsetMap, outputTopic, key,
+        VALUE_SCHEMA, valueObject, serviceBusMessage.getEnqueuedTime().toEpochSecond());
+  }
+
+  private static Struct createStructFromServiceBusMessage(final ServiceBusReceivedMessage serviceBusMessage) {
+    return new Struct(VALUE_SCHEMA)
+        .put(DELIVERY_COUNT, serviceBusMessage.getDeliveryCount())
+        .put(ENQUEUED_TIME_UTC, serviceBusMessage.getEnqueuedTime().toEpochSecond())
+        .put(CONTENT_TYPE, serviceBusMessage.getContentType())
+        .put(LABEL, AzureServiceBusSourceConnector.class.getSimpleName())
+        .put(CORRELATION_ID, serviceBusMessage.getCorrelationId())
+        .put(PARTITION_KEY, serviceBusMessage.getPartitionKey())
+        .put(REPLY_TO, serviceBusMessage.getReplyTo())
+        .put(REPLY_TO_SESSION_ID, serviceBusMessage.getReplyToSessionId())
+        .put(DEAD_LETTER_SOURCE, serviceBusMessage.getDeadLetterSource())
+        .put(TIME_TO_LIVE, serviceBusMessage.getTimeToLive().toMillis())
+        .put(LOCKED_UNTIL_UTC, serviceBusMessage.getLockedUntil().toEpochSecond())
+        .put(SEQUENCE_NUMBER, serviceBusMessage.getSequenceNumber())
+        .put(SESSION_ID, serviceBusMessage.getSessionId())
+        .put(LOCK_TOKEN, serviceBusMessage.getLockToken())
+        .put(MESSAGE_BODY, serviceBusMessage.getBody().toBytes())
+        .put(GET_TO, serviceBusMessage.getTo());
+  }
+
+}