Azure Servicebus Source Connector

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-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,38 @@
+/*
+ * 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;
+
+/**
+ * Class to indicate String constants used in Service Bus Connectors Config.
+ */
+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.";
+
+}

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

@@ -0,0 +1,77 @@
+/*
+ * 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;
+
+/**
+ * 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.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.Header;
+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, Integer partition, Schema keySchema, Object key,
+      Schema valueSchema, Object value, Long timestamp, Iterable<Header> headers) {
+    super(sourcePartition, sourceOffset, topic, partition, keySchema, key, valueSchema, value, timestamp, headers);
+  }
+}

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

@@ -0,0 +1,99 @@
+/*
+ * 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.header.ConnectHeaders;
+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 = new Struct(VALUE_SCHEMA);
+
+    mapValuesFromServiceBusMessage(valueObject, serviceBusMessage);
+    return new AzureServiceBusSourceRecord(partitionKey, offsetMap, outputTopic, null, Schema.STRING_SCHEMA, key,
+        VALUE_SCHEMA, valueObject, serviceBusMessage.getEnqueuedTime().toEpochSecond(), new ConnectHeaders());
+  }
+
+  private static void mapValuesFromServiceBusMessage(Struct targetObject, ServiceBusReceivedMessage serviceBusMessage) {
+    targetObject.put(DELIVERY_COUNT, serviceBusMessage.getDeliveryCount());
+    targetObject.put(ENQUEUED_TIME_UTC, serviceBusMessage.getEnqueuedTime().toEpochSecond());
+    targetObject.put(CONTENT_TYPE, serviceBusMessage.getContentType());
+    targetObject.put(LABEL, AzureServiceBusSourceConnector.class.getSimpleName());
+    targetObject.put(CORRELATION_ID, serviceBusMessage.getCorrelationId());
+    targetObject.put(PARTITION_KEY, serviceBusMessage.getPartitionKey());
+    targetObject.put(REPLY_TO, serviceBusMessage.getReplyTo());
+    targetObject.put(REPLY_TO_SESSION_ID, serviceBusMessage.getReplyToSessionId());
+    targetObject.put(DEAD_LETTER_SOURCE, serviceBusMessage.getDeadLetterSource());
+    targetObject.put(TIME_TO_LIVE, serviceBusMessage.getTimeToLive().toMillis());
+    targetObject.put(LOCKED_UNTIL_UTC, serviceBusMessage.getLockedUntil().toEpochSecond());
+    targetObject.put(SEQUENCE_NUMBER, serviceBusMessage.getSequenceNumber());
+    targetObject.put(SESSION_ID, serviceBusMessage.getSessionId());
+    targetObject.put(LOCK_TOKEN, serviceBusMessage.getLockToken());
+    targetObject.put(MESSAGE_BODY, serviceBusMessage.getBody().toBytes());
+    targetObject.put(GET_TO, serviceBusMessage.getTo());
+  }
+
+}