* Splits Elastic module into elastic-common and elastic8

* Includes refactoring of all connectors to ensure that Maps are passed around and util.Maps are only used on entry and exit from scala code and converted soon after
* Adding Opensearch module
* Adding Opensearch unit tests
* Opensearch SSL Test - needs completing
* Replace Elastic6+7 with Elastic8

Removing deleted modules

Source package renamer

Package rename

Dir rename

Author: davidsloan

build.sbt

@@ -1,5 +1,7 @@
 import Dependencies.globalExcludeDeps
 import Dependencies.gson
+import Dependencies.bouncyCastle
+
 import Settings.*
 import sbt.Keys.libraryDependencies
 import sbt.*
@@ -17,8 +19,9 @@ lazy val subProjects: Seq[Project] = Seq(
   `azure-documentdb`,
   `azure-datalake`,
   cassandra,
-  elastic6,
-  elastic7,
+  `elastic-common`,
+  opensearch,
+  elastic8,
   ftp,
   `gcp-storage`,
   http,
@@ -201,17 +204,16 @@ lazy val cassandra = (project in file("kafka-connect-cassandra"))
   .configureFunctionalTests()
   .enablePlugins(PackPlugin)
 
-lazy val elastic6 = (project in file("kafka-connect-elastic6"))
+lazy val `elastic-common` = (project in file("kafka-connect-elastic-common"))
   .dependsOn(common)
   .dependsOn(`test-common` % "fun->compile")
   .settings(
     settings ++
       Seq(
-        name := "kafka-connect-elastic6",
+        name := "kafka-connect-elastic-common",
         description := "Kafka Connect compatible connectors to move data between Kafka and popular data stores",
-        libraryDependencies ++= baseDeps ++ kafkaConnectElastic6Deps,
+        libraryDependencies ++= baseDeps ++ kafkaConnectElasticBaseDeps,
         publish / skip := true,
-        FunctionalTest / baseDirectory := (LocalRootProject / baseDirectory).value,
         packExcludeJars := Seq(
           "scala-.*\\.jar",
           "zookeeper-.*\\.jar",
@@ -220,19 +222,20 @@ lazy val elastic6 = (project in file("kafka-connect-elastic6"))
   )
   .configureAssembly(true)
   .configureTests(baseTestDeps)
-  .configureIntegrationTests(kafkaConnectElastic6TestDeps)
+  .configureIntegrationTests(kafkaConnectElastic8TestDeps)
   .configureFunctionalTests()
-  .enablePlugins(PackPlugin)
+  .disablePlugins(PackPlugin)
 
-lazy val elastic7 = (project in file("kafka-connect-elastic7"))
+lazy val elastic8 = (project in file("kafka-connect-elastic8"))
   .dependsOn(common)
-  .dependsOn(`test-common` % "fun->compile")
+  .dependsOn(`elastic-common`)
+  .dependsOn(`test-common` % "fun->compile;it->compile")
   .settings(
     settings ++
       Seq(
-        name := "kafka-connect-elastic7",
+        name := "kafka-connect-elastic8",
         description := "Kafka Connect compatible connectors to move data between Kafka and popular data stores",
-        libraryDependencies ++= baseDeps ++ kafkaConnectElastic7Deps,
+        libraryDependencies ++= baseDeps ++ kafkaConnectElastic8Deps,
         publish / skip := true,
         packExcludeJars := Seq(
           "scala-.*\\.jar",
@@ -242,10 +245,33 @@ lazy val elastic7 = (project in file("kafka-connect-elastic7"))
   )
   .configureAssembly(true)
   .configureTests(baseTestDeps)
-  .configureIntegrationTests(kafkaConnectElastic7TestDeps)
+  .configureIntegrationTests(kafkaConnectElastic8TestDeps)
   .configureFunctionalTests()
   .enablePlugins(PackPlugin)
 
+lazy val opensearch = (project in file("kafka-connect-opensearch"))
+  .dependsOn(common)
+  .dependsOn(`elastic-common`)
+  .dependsOn(`test-common` % "fun->compile;it->compile")
+  .settings(
+    settings ++
+      Seq(
+        name := "kafka-connect-opensearch",
+        description := "Kafka Connect compatible connectors to move data between Kafka and popular data stores",
+        libraryDependencies ++= baseDeps ++ kafkaConnectOpenSearchDeps,
+        publish / skip := true,
+        packExcludeJars := Seq(
+          "scala-.*\\.jar",
+          "zookeeper-.*\\.jar",
+        ),
+      ),
+  )
+  .configureAssembly(false)
+  .configureTests(baseTestDeps)
+  //.configureIntegrationTests(kafkaConnectOpenSearchTestDeps)
+  .configureFunctionalTests(bouncyCastle)
+  .enablePlugins(PackPlugin)
+
 lazy val http = (project in file("kafka-connect-http"))
   .dependsOn(common)
   //.dependsOn(`test-common` % "fun->compile")

kafka-connect-aws-s3/src/main/scala/io/lenses/streamreactor/connect/aws/s3/sink/config/S3SinkConfig.scala

@@ -49,7 +49,7 @@ object S3SinkConfig {
         s3ConfigDefBuilder.getInt(SEEK_MAX_INDEX_FILES),
       )
     } yield S3SinkConfig(
-      S3Config(s3ConfigDefBuilder.getParsedValues),
+      S3Config(s3ConfigDefBuilder.props),
       sinkBucketOptions,
       offsetSeekerOptions,
       s3ConfigDefBuilder.getCompressionCodec(),

kafka-connect-aws-s3/src/main/scala/io/lenses/streamreactor/connect/aws/s3/sink/config/S3SinkConfigDefBuilder.scala

@@ -18,19 +18,15 @@ package io.lenses.streamreactor.connect.aws.s3.sink.config
 import io.lenses.streamreactor.common.config.base.traits._
 import io.lenses.streamreactor.connect.aws.s3.config.DeleteModeSettings
 import io.lenses.streamreactor.connect.aws.s3.config.S3ConfigSettings
+import io.lenses.streamreactor.connect.cloud.common.config.CompressionCodecSettings
 import io.lenses.streamreactor.connect.cloud.common.sink.config.CloudSinkConfigDefBuilder
 
-import scala.jdk.CollectionConverters.MapHasAsScala
-
 case class S3SinkConfigDefBuilder(props: Map[String, String])
     extends BaseConfig(S3ConfigSettings.CONNECTOR_PREFIX, S3SinkConfigDef.config, props)
     with CloudSinkConfigDefBuilder
     with ErrorPolicySettings
     with NumberRetriesSettings
     with UserSettings
     with ConnectionSettings
-    with DeleteModeSettings {
-
-  def getParsedValues: Map[String, _] = values().asScala.toMap
-
-}
+    with CompressionCodecSettings
+    with DeleteModeSettings {}

kafka-connect-aws-s3/src/main/scala/io/lenses/streamreactor/connect/aws/s3/source/S3SourceTask.scala

@@ -63,9 +63,8 @@ class S3SourceTask extends SourceTask with LazyLogging {
 
     logger.debug(s"Received call to S3SourceTask.start with ${props.size()} properties")
 
-    val contextProperties: Map[String, String] =
-      Option(context).flatMap(c => Option(c.configs()).map(_.asScala.toMap)).getOrElse(Map.empty)
-    val mergedProperties: Map[String, String] = MapUtils.mergeProps(contextProperties, props.asScala.toMap)
+    val contextProperties = Option(context).flatMap(c => Option(c.configs()).map(_.asScala.toMap)).getOrElse(Map.empty)
+    val mergedProperties  = MapUtils.mergeProps(contextProperties, props.asScala.toMap)
     (for {
       result <- S3SourceState.make(mergedProperties, contextOffsetFn)
       fiber  <- result.partitionDiscoveryLoop.start

kafka-connect-aws-s3/src/main/scala/io/lenses/streamreactor/connect/aws/s3/source/config/S3SourceConfig.scala

@@ -50,7 +50,7 @@ object S3SourceConfig {
     S3SourceConfig(S3SourceConfigDefBuilder(props))
 
   def apply(s3ConfigDefBuilder: S3SourceConfigDefBuilder): Either[Throwable, S3SourceConfig] = {
-    val parsedValues = s3ConfigDefBuilder.getParsedValues
+    val parsedValues = s3ConfigDefBuilder.props
     for {
       sbo <- SourceBucketOptions(
         s3ConfigDefBuilder,

kafka-connect-aws-s3/src/main/scala/io/lenses/streamreactor/connect/aws/s3/source/config/S3SourceConfigDefBuilder.scala

@@ -20,8 +20,6 @@ import io.lenses.streamreactor.connect.aws.s3.config.DeleteModeSettings
 import io.lenses.streamreactor.connect.aws.s3.config.S3ConfigSettings
 import io.lenses.streamreactor.connect.cloud.common.config.CompressionCodecSettings
 
-import scala.jdk.CollectionConverters.MapHasAsScala
-
 case class S3SourceConfigDefBuilder(props: Map[String, String])
     extends BaseConfig(S3ConfigSettings.CONNECTOR_PREFIX, S3SourceConfigDef.config, props)
     with KcqlSettings
@@ -31,8 +29,4 @@ case class S3SourceConfigDefBuilder(props: Map[String, String])
     with ConnectionSettings
     with CompressionCodecSettings
     with SourcePartitionSearcherSettings
-    with DeleteModeSettings {
-
-  def getParsedValues: Map[String, _] = values().asScala.toMap
-
-}
+    with DeleteModeSettings {}

kafka-connect-azure-datalake/src/main/scala/io/lenses/streamreactor/connect/datalake/sink/config/DatalakeSinkConfig.scala

@@ -49,7 +49,7 @@ object DatalakeSinkConfig {
         s3ConfigDefBuilder.getInt(SEEK_MAX_INDEX_FILES),
       )
     } yield DatalakeSinkConfig(
-      AzureConfig(s3ConfigDefBuilder.getParsedValues, authMode),
+      AzureConfig(s3ConfigDefBuilder.props, authMode),
       sinkBucketOptions,
       offsetSeekerOptions,
       s3ConfigDefBuilder.getCompressionCodec(),

kafka-connect-azure-datalake/src/main/scala/io/lenses/streamreactor/connect/datalake/sink/config/DatalakeSinkConfigDefBuilder.scala

@@ -16,21 +16,17 @@
 package io.lenses.streamreactor.connect.datalake.sink.config
 
 import io.lenses.streamreactor.common.config.base.traits._
+import io.lenses.streamreactor.connect.cloud.common.config.CompressionCodecSettings
 import io.lenses.streamreactor.connect.cloud.common.sink.config.CloudSinkConfigDefBuilder
 import io.lenses.streamreactor.connect.datalake.config.AuthModeSettings
 import io.lenses.streamreactor.connect.datalake.config.AzureConfigSettings
 
-import scala.jdk.CollectionConverters.MapHasAsScala
-
 case class DatalakeSinkConfigDefBuilder(props: Map[String, String])
     extends BaseConfig(AzureConfigSettings.CONNECTOR_PREFIX, DatalakeSinkConfigDef.config, props)
     with CloudSinkConfigDefBuilder
     with ErrorPolicySettings
     with NumberRetriesSettings
     with UserSettings
     with ConnectionSettings
-    with AuthModeSettings {
-
-  def getParsedValues: Map[String, _] = values().asScala.toMap
-
-}
+    with CompressionCodecSettings
+    with AuthModeSettings {}

kafka-connect-azure-documentdb/src/main/scala/io/lenses/streamreactor/connect/azure/documentdb/sink/DocumentDbSinkConnector.scala

@@ -15,6 +15,7 @@
  */
 package io.lenses.streamreactor.connect.azure.documentdb.sink
 
+import cats.implicits.toBifunctorOps
 import io.lenses.streamreactor.common.config.Helpers
 import io.lenses.streamreactor.common.utils.JarManifest
 import io.lenses.streamreactor.connect.azure.documentdb.DocumentClientProvider
@@ -100,7 +101,7 @@ class DocumentDbSinkConnector private[sink] (builder: DocumentDbSinkSettings =>
     configProps = props
 
     //check input topics
-    Helpers.checkInputTopics(DocumentDbConfigConstants.KCQL_CONFIG, props.asScala.toMap)
+    Helpers.checkInputTopics(DocumentDbConfigConstants.KCQL_CONFIG, props.asScala.toMap).leftMap(throw _)
 
     val settings = DocumentDbSinkSettings(config)
 

kafka-connect-cassandra/src/main/scala/io/lenses/streamreactor/connect/cassandra/CassandraConnection.scala

@@ -15,9 +15,9 @@
  */
 package io.lenses.streamreactor.connect.cassandra
 
-import io.lenses.streamreactor.common.config.SSLConfig
-import io.lenses.streamreactor.common.config.SSLConfigContext
 import io.lenses.streamreactor.connect.cassandra.config.CassandraConfigConstants
+import io.lenses.streamreactor.connect.cassandra.config.SSLConfig
+import io.lenses.streamreactor.connect.cassandra.config.SSLConfigContext
 import io.lenses.streamreactor.connect.cassandra.config.LoadBalancingPolicy
 import com.datastax.driver.core.Cluster.Builder
 import com.datastax.driver.core.policies.DCAwareRoundRobinPolicy