diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index 594f23d..4b09960 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -10,7 +10,7 @@ kotlin = "1.6.10" ## libs ## camel-kafkaConnectors = "0.11.0" -http4k = "4.17.4.0" +http4k = "4.17.6.0" kafka = "3.0.0" kotlinx-serialization = "1.3.1" kvision = "5.6.1" @@ -24,7 +24,7 @@ kotest = "5.0.1" ## infra ## factorio = "1.1.50" -confluent-kafkaConnect = "7.0.0" +confluent-kafkaConnect = "7.0.1" ######### [plugins] diff --git a/modules/infra-kafka-connect/build.gradle.kts b/modules/infra-kafka-connect/build.gradle.kts index 2899214..ff2da04 100644 --- a/modules/infra-kafka-connect/build.gradle.kts +++ b/modules/infra-kafka-connect/build.gradle.kts @@ -1,6 +1,7 @@ @file:Suppress("UnstableApiUsage") // VERSION_PLATFORM + platform(...) is unstable import dev.adamko.kafkatorio.gradle.asConsumer +import org.jetbrains.kotlin.util.parseSpaceSeparatedArgs plugins { id("dev.adamko.kafkatorio.infra.docker-compose") @@ -16,12 +17,13 @@ fun DependencyHandlerScope.kafkaConnector( name: String, group: String = "org.apache.camel.kafkaconnector", version: String = libs.versions.camel.kafkaConnectors.get(), + classifier: String = "package", + extension: String = "tar.gz", ) { - camelConnectorDependencies(group, name, version) + camelConnectorDependencies(group, name, version, ext = extension, classifier = classifier) } dependencies { - kafkaConnector("camel-websocket-kafka-connector") kafkaConnector("camel-ahc-ws-kafka-connector") implementation("com.github.adamko-dev:json5-kotlin:2.0.3") @@ -35,21 +37,46 @@ dependencies { implementation("org.jetbrains.kotlinx:kotlinx-serialization-json") } +val dockerBuildKafkaConnect by tasks.registering(Exec::class) { + group = "docker-compose" + dependsOn(downloadCamelConnectors, tasks.dockerEnv) + + logging.captureStandardOutput(LogLevel.LIFECYCLE) + + inputs.dir(dockerSrcDir) + + workingDir = dockerSrcDir.asFile + commandLine = parseSpaceSeparatedArgs(""" docker-compose build connect """) +} +tasks.assemble { dependsOn(dockerBuildKafkaConnect) } + val downloadCamelConnectors by tasks.registering(Sync::class) { group = project.name description = "Retrieve Camel Kafka Connectors from Maven, in preparation for the Docker build" + dependsOn(camelConnectorDependencies) + from( provider { camelConnectorDependencies } .map { eventsSchema -> eventsSchema.incoming - .artifactView { lenient(true) } .artifacts .artifactFiles } ) - into(connectorJarDir) + into(temporaryDir) + + doLast { + sync { + fileTree(temporaryDir) + .asFileTree + .forEach { zip -> + from(tarTree(zip)) + } + into(connectorJarDir) + } + } } tasks.dockerUp { diff --git a/modules/infra-kafka-connect/src/main/docker/docker-compose.yml b/modules/infra-kafka-connect/src/main/docker/docker-compose.yml index 7833ca4..f44f3af 100644 --- a/modules/infra-kafka-connect/src/main/docker/docker-compose.yml +++ b/modules/infra-kafka-connect/src/main/docker/docker-compose.yml @@ -15,7 +15,8 @@ services: # CAMEL_KC_VERSION: ${CONNECTOR_CAMEL_KAFKA_VERSION} # CAMEL_CONNECTOR: ${CAMEL_FILE_CONNECTOR} ports: - - "28083:28083" + - "8083:8083" + - "8095:8095" # volumes: # - "./connect/input:/warehouse/input" # - "./connect/components:/usr/share/confluent-hub-components" @@ -28,7 +29,8 @@ services: 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_REST_PORT: 28083 + CONNECT_REST_PORT: 8083 CONNECT_GROUP_ID: ${CONNECTOR_ID} CONNECT_CONFIG_STORAGE_TOPIC: ${CONNECTOR_ID}-configs CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1 @@ -40,9 +42,10 @@ services: 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_PLUGIN_PATH: "/usr/share/java,/usr/share/camel-kafka-connectors" CONNECT_LOG4J_ROOT_LOGLEVEL: INFO CONNECT_LOG4J_LOGGERS: "org.apache.zookeeper=INFO,org.I0Itec.zkclient=ERROR,org.reflections=ERROR" +# CONNECT_LISTENERS: "http://localhost:28083" networks: kafkatorio-nw: diff --git a/modules/infra-kafka-connect/src/main/docker/kafka-connect.Dockerfile b/modules/infra-kafka-connect/src/main/docker/kafka-connect.Dockerfile index 7659554..885c292 100644 --- a/modules/infra-kafka-connect/src/main/docker/kafka-connect.Dockerfile +++ b/modules/infra-kafka-connect/src/main/docker/kafka-connect.Dockerfile @@ -7,7 +7,7 @@ ENV CONNECT_VERSION=${KAFKA_CONNECT_VERSION} USER 0:0 -COPY ./camel-kafka-connectors /usr/share/java/ +COPY ./camel-kafka-connectors /usr/share/camel-kafka-connectors #WORKDIR /usr/share/java/ # diff --git a/modules/infra-kafka-connect/src/main/kotlin/dev/adamko/kafkatorio/kafkaconnect/Resources.kt b/modules/infra-kafka-connect/src/main/kotlin/dev/adamko/kafkatorio/kafkaconnect/Resources.kt new file mode 100644 index 0000000..0e6406c --- /dev/null +++ b/modules/infra-kafka-connect/src/main/kotlin/dev/adamko/kafkatorio/kafkaconnect/Resources.kt @@ -0,0 +1,10 @@ +package dev.adamko.kafkatorio.kafkaconnect + +import java.io.File + +object Resources { + fun loadResource(name: String): File = + File(requireNotNull(Resources::class.java.getResource(name)?.file) { + "could not load resource '$name'" + }) +} diff --git a/modules/infra-kafka-connect/src/main/kotlin/dev/adamko/kafkatorio/kafkaconnect/kafkaConnectAdmin.kt b/modules/infra-kafka-connect/src/main/kotlin/dev/adamko/kafkatorio/kafkaconnect/kafkaConnectAdmin.kt index a34e265..6a68965 100644 --- a/modules/infra-kafka-connect/src/main/kotlin/dev/adamko/kafkatorio/kafkaconnect/kafkaConnectAdmin.kt +++ b/modules/infra-kafka-connect/src/main/kotlin/dev/adamko/kafkatorio/kafkaconnect/kafkaConnectAdmin.kt @@ -3,10 +3,15 @@ package dev.adamko.kafkatorio.kafkaconnect import at.syntaxerror.json5.Json5Module import kotlinx.serialization.decodeFromString import kotlinx.serialization.encodeToString +import kotlinx.serialization.json.Json import kotlinx.serialization.json.JsonElement import kotlinx.serialization.json.JsonObject - -import kotlinx.serialization.json.Json +import kotlinx.serialization.json.jsonPrimitive +import org.http4k.client.OkHttp +import org.http4k.core.Method.GET +import org.http4k.core.Method.PUT +import org.http4k.core.Request +import org.http4k.core.Uri val jsonMapper = Json { prettyPrint = true @@ -15,6 +20,42 @@ val jsonMapper = Json { val j5 = Json5Module() +fun main() { + + val connectorHostUrl = Uri.of("http://localhost:8083") + + val configFile = Resources.loadResource("/configs/websocket-sink-lua-topics.json5") + + val jsonObject: JsonObject = j5.decodeObject(configFile.readText()) + println("json5 connector config: $jsonObject") + val configJson = jsonMapper.encodeToString(jsonObject) + println("json connector config: $configJson") + + val connectorName = jsonObject["name"]!!.jsonPrimitive.content + + val request: Request = + Request(PUT, connectorHostUrl.path("/connectors/$connectorName/config")) + .header("Content-Type", "application/json") + .header("Accept", "application/json") +// .header("Accept-Encoding", "identity") + .body(configJson) + + val client = OkHttp() + + // check the status of KC + val statusResponse = client(Request(GET, connectorHostUrl)) + println(statusResponse.toMessage()) + + val response = client(request) + + println(response.toMessage()) + println( + jsonMapper.encodeToString( + jsonMapper.decodeFromString(response.body.toString()) + ) + ) +} + //val updateConnectors by tasks.registering { // group = project.name // diff --git a/modules/infra-kafka-connect/src/main/resources/configs/websocket-sink-lua-topics.json5 b/modules/infra-kafka-connect/src/main/resources/configs/websocket-sink-lua-topics.json5 index 8fc8556..09963ad 100644 --- a/modules/infra-kafka-connect/src/main/resources/configs/websocket-sink-lua-topics.json5 +++ b/modules/infra-kafka-connect/src/main/resources/configs/websocket-sink-lua-topics.json5 @@ -14,7 +14,7 @@ // mandatory properties (for a complete properties list see the connector documentation): // The URI to use such as http://hostname:port/path - "camel.sink.path.httpUri": "http://localhost:/ws", + "camel.sink.path.httpUri": "http://localhost:8095/ws", // If the option is true, then the Exchange.HTTP_URI header is ignored, and use the endpoint's URI for request. You may also set the throwExceptionOnFailure to be false to let the AhcProducer send all the fault response back. // camel.sink.endpoint.bridgeEndpoint= // The initial in-memory buffer size used when transferring data between Camel and AHC Client.