kafka connect docker image setup

buildSrc/src/main/kotlin/dev/adamko/kafkatorio/infra/docker-compose.gradle.kts

@@ -6,7 +6,7 @@ plugins {
   id("dev.adamko.kafkatorio.infra.process-runner")
 }
 
-val dockerComposeTaskGroup : String by extra("docker-compose")
+val dockerComposeTaskGroup: String by extra("docker-compose")
 
 val composeProjectName: String by extra(rootProject.name)
 val srcDir: Directory by extra(layout.projectDirectory.dir("src"))
@@ -14,12 +14,16 @@ val srcDir: Directory by extra(layout.projectDirectory.dir("src"))
 
 val dockerUp by tasks.creating(Exec::class) {
   group = dockerComposeTaskGroup
+
+  dependsOn(tasks.assemble)
+
   workingDir = srcDir.asFile
   commandLine = parseSpaceSeparatedArgs(""" docker-compose up -d """)
 }
 
 val dockerDown by tasks.creating(Exec::class) {
   group = dockerComposeTaskGroup
+
   workingDir = srcDir.asFile
   commandLine = parseSpaceSeparatedArgs(""" docker-compose down """)
 }

buildSrc/src/main/kotlin/dev/adamko/kafkatorio/lang/node.gradle.kts

@@ -53,3 +53,5 @@ val updatePackageJson by tasks.registering(UpdatePackageJson::class) {
 
   }
 }
+
+tasks.assemble { dependsOn(updatePackageJson) }

gradle/libs.versions.toml

@@ -9,20 +9,22 @@ jvm        = "11"
 kotlin     = "1.6.0"
 
 ## libs ##
-http4k     = "4.17.2.0"
+camel-kafkaConnectors = "0.11.0"
-slf4j      = "1.7.30"
+http4k                = "4.17.2.0"
-log4j2     = "2.14.1"
+kafka                 = "3.0.0"
-qosLogback = "1.2.5"
-kafka      = "3.0.0"
 kotlinx-serialization = "1.3.1"
-kvision    = "5.6.1"
+kvision               = "5.6.1"
+log4j2                = "2.14.1"
+qosLogback            = "1.2.5"
+slf4j                 = "1.7.30"
 
 ## test ##
 junit      = "5.8.2"
 kotest     = "5.0.1"
 
 ## infra ##
-factorio = "1.1.49"
+factorio               = "1.1.49"
+confluent-kafkaConnect = "7.0.0"
 
 #########
 [plugins]

modules/infra-kafka-connect/build.gradle.kts

@@ -0,0 +1,127 @@
+import dev.adamko.kafkatorio.gradle.asConsumer
+import dev.adamko.kafkatorio.jsonMapper
+import kotlinx.serialization.encodeToString
+import kotlinx.serialization.json.contentOrNull
+import kotlinx.serialization.json.jsonArray
+import kotlinx.serialization.json.jsonObject
+import kotlinx.serialization.json.jsonPrimitive
+
+plugins {
+  id("dev.adamko.kafkatorio.infra.docker-compose")
+}
+
+val camelConnectorDependencies: Configuration by configurations.creating { asConsumer() }
+
+fun DependencyHandlerScope.kafkaConnector(
+  name: String,
+  group: String = "org.apache.camel.kafkaconnector",
+  version: String = libs.versions.camel.kafkaConnectors.get(),
+) {
+  camelConnectorDependencies(group, name, version)
+}
+
+dependencies {
+  kafkaConnector("camel-websocket-kafka-connector")
+  kafkaConnector("camel-ahc-ws-kafka-connector")
+}
+
+
+val connectorJarDir: Directory = layout.projectDirectory.dir("src/camel-kafka-connectors")
+
+val downloadCamelConnectors by tasks.registering(Sync::class) {
+  group = project.name
+  description = "Retrieve Camel Kafka Connectors from Maven, in preparation for the Docker build"
+
+  from(
+    provider { camelConnectorDependencies }
+      .map { eventsSchema ->
+        eventsSchema.incoming
+          .artifactView { lenient(true) }
+          .artifacts
+          .artifactFiles
+      }
+  )
+  into(connectorJarDir)
+}
+
+
+tasks.dockerUp {
+  dependsOn(":modules:infra-kafka-cluster:dockerUp")
+  dependsOn(downloadCamelConnectors)
+}
+
+tasks.dockerEnv {
+  properties(
+    "CONNECTOR_ID" to "kafkatorio-connect",
+    "KAFKA_CONNECT_VERSION" to libs.versions.confluent.kafkaConnect.get(),
+  )
+}
+
+
+val availableCamelVersions by tasks.registering {
+  // dicking around with some sort of 'all camel connectors' docker image.
+  // it's unfinished.
+
+  val camelGroup = "org.apache.camel.kafkaconnector"
+  val camelParent = "connectors"
+
+  val requestUrl = """
+    https://search.maven.org/solrsearch/select?q=g:"$camelGroup"+AND+a:"$camelParent"&core=gav&rows=20&wt=json
+  """.trimIndent()
+    .replace("\"", "%22")
+
+  doLast {
+
+    val connectorVersions = java.net.URL(requestUrl).readText()
+    logger.lifecycle("conn ==== $connectorVersions")
+    val versionsJson = jsonMapper.parseToJsonElement(connectorVersions).jsonObject
+
+    logger.lifecycle("json ==== \n ${jsonMapper.encodeToString(versionsJson)}")
+
+    val deps = versionsJson["response"]
+      ?.jsonObject
+      ?.get("docs")
+      ?.jsonArray
+      ?.map { it.jsonObject }
+      ?.mapNotNull { j ->
+        val group = j["g"]?.jsonPrimitive?.contentOrNull
+        val artifact = j["a"]?.jsonPrimitive?.contentOrNull
+        val version = j["v"]?.jsonPrimitive?.contentOrNull
+        val packaging = j["p"]?.jsonPrimitive?.contentOrNull?.toLowerCase()
+
+        if (listOf(group, artifact, version, packaging).all { it != null }) {
+          val d = project.dependencies.create(
+            group = group!!,
+            name = artifact!!,
+            version = version!!,
+            ext = packaging,
+          )
+          logger.lifecycle("dep = $d / $packaging")
+          d
+        } else {
+          null
+        }
+      } ?: emptyList()
+
+    val c = configurations.detachedConfiguration()
+    c.dependencies.addAll(deps)
+    c.resolve()
+      .joinToString("\n\n---\n\n") { it.readLines().take(5).joinToString("\n") }
+
+//    logger.lifecycle("ids: ${versions?.joinToString("\n", "\n\n", "\n\n")}")
+//
+//    versions?.filterNotNull()?.forEach { id ->
+//
+//      configurations.detachedConfiguration(
+//        project.dependencies.create(id)
+//      )
+//        .resolve()
+//        .joinToString("\n\n---\n\n") { it.readLines().take(5).joinToString("\n") }
+//
+//    }
+
+
+//  jsonMapper.parseToJsonElement()
+
+  }
+}

modules/infra-kafka-connect/src/.env

@@ -0,0 +1,4 @@
+# Do not edit manually. This file was created with task 'dockerEnv'
+COMPOSE_PROJECT_NAME=kafkatorio
+CONNECTOR_ID=kafkatorio-connect
+KAFKA_CONNECT_VERSION=7.0.0

modules/infra-kafka-connect/src/.gitignore

@@ -0,0 +1 @@
+camel-kafka-connectors/

modules/infra-kafka-connect/src/docker-compose.yml

@@ -0,0 +1,50 @@
+version: "3.9"
+
+services:
+
+  connect:
+    image: adamko-dev/kafkatorio-kafka-connect:${KAFKA_CONNECT_VERSION}
+    hostname: ${CONNECTOR_ID}
+    container_name: ${CONNECTOR_ID}
+    build:
+      context: "."
+      dockerfile: kafka-connect.Dockerfile
+      args:
+        KAFKA_CONNECT_VERSION: ${KAFKA_CONNECT_VERSION}
+#        FILE_PULSE_VERSION: ${CONNECTOR_FILE_PULSE_VERSION}
+#        CAMEL_KC_VERSION: ${CONNECTOR_CAMEL_KAFKA_VERSION}
+#        CAMEL_CONNECTOR: ${CAMEL_FILE_CONNECTOR}
+    ports:
+      - "28083:28083"
+#    volumes:
+#      - "./connect/input:/warehouse/input"
+#      - "./connect/components:/usr/share/confluent-hub-components"
+
+#    volumes:
+#      - "./camel-kafka-connectors:/usr/share/java"
+    networks:
+      - "kafkatorio-nw"
+    environment:
+      CONNECT_BOOTSTRAP_SERVERS: "kafka:29092"
+      CONNECT_REST_ADVERTISED_HOST_NAME: ${CONNECTOR_ID}
+#      CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL: "http://${SCHEMA_REGISTRY}:8081"
+      CONNECT_REST_PORT: 28083
+      CONNECT_GROUP_ID: ${CONNECTOR_ID}
+      CONNECT_CONFIG_STORAGE_TOPIC: ${CONNECTOR_ID}-configs
+      CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1
+      CONNECT_OFFSET_FLUSH_INTERVAL_MS: 10000
+      CONNECT_OFFSET_STORAGE_TOPIC: ${CONNECTOR_ID}-offsets
+      CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1
+      CONNECT_STATUS_STORAGE_TOPIC: ${CONNECTOR_ID}-status
+      CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1
+      CONNECT_KEY_CONVERTER: "org.apache.kafka.connect.storage.StringConverter"
+#      CONNECT_VALUE_CONVERTER: "io.confluent.connect.avro.AvroConverter"
+      CONNECT_VALUE_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
+      CONNECT_PLUGIN_PATH: "/usr/share/java,/usr/share/confluent-hub-components"
+      CONNECT_LOG4J_ROOT_LOGLEVEL: INFO
+      CONNECT_LOG4J_LOGGERS: "org.apache.zookeeper=INFO,org.I0Itec.zkclient=ERROR,org.reflections=ERROR"
+
+networks:
+  kafkatorio-nw:
+    name: "kafkatorio-nw"
+    external: true

modules/infra-kafka-connect/src/kafka-connect.Dockerfile

@@ -0,0 +1,19 @@
+ARG KAFKA_CONNECT_VERSION
+
+
+FROM confluentinc/cp-kafka-connect-base:${KAFKA_CONNECT_VERSION}
+
+ENV CONNECT_VERSION=${KAFKA_CONNECT_VERSION}
+
+USER 0:0
+
+COPY ./camel-kafka-connectors /usr/share/java/
+
+#WORKDIR /usr/share/java/
+#
+##ENV CAMEL_KC_VERSION=0.9.0
+##ENV CAMEL_CONNECTOR=camel-file-kafka-connector
+#
+#ENV URL=https://repo.maven.apache.org/maven2/org/apache/camel/kafkaconnector/camel-file-kafka-connector/0.9.0/camel-file-kafka-connector-0.9.0-package.tar.gz
+#
+#RUN curl -L $URL | tar -xzvf - --strip=0 $CAMEL_CONNECTOR

settings.gradle.kts

@@ -8,8 +8,10 @@ include(
 
   ":modules:infra-factorio-client",
   ":modules:infra-factorio-server",
+
   ":modules:infra-kafka-cluster",
   ":modules:infra-kafka-pipe",
+  ":modules:infra-kafka-connect",
 
   ":modules:kt-rcon",
   ":modules:web-map",