successfully create KC websocket connector (not sure if it works though...)
This commit is contained in:
Adam
parent
427b8951d3
commit
83c991842f
7 changed files with 94 additions and 13 deletions
|
|
@ -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]
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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/
|
||||
#
|
||||
|
|
|
|||
|
|
@ -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'"
|
||||
})
|
||||
}
|
||||
|
|
@ -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<JsonElement>(response.body.toString())
|
||||
)
|
||||
)
|
||||
}
|
||||
|
||||
//val updateConnectors by tasks.registering {
|
||||
// group = project.name
|
||||
//
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue