From 26626263d3bea127f8df66f45650e3a8382e0f8f Mon Sep 17 00:00:00 2001 From: Adam <897017+asemy@users.noreply.github.com> Date: Sat, 23 Oct 2021 22:44:20 +0200 Subject: [PATCH] kafka pipe working!!!!!!!!!!!!!!!!! lots of experiments too - to be deleted --- .../archetype/kotlin-jvm.gradle.kts | 10 +- .../build.gradle.kts | 155 ++++++++++++++++++ .../docker-compose.yml | 54 ++++++ .../kafka-pipe-jar.Dockerfile | 12 ++ .../kafka-pipe.Dockerfile | 31 ++++ .../factorio-events-kafka-pipe/kafka-pipe.sh | 21 +++ .../pipe-logs-to-kafka.sh | 50 ++++++ .../src/main/kotlin/fekp.kt | 17 ++ .../build.gradle.kts | 1 - modules/mod/infra/docker-compose.yml | 22 ++- modules/mod/infra/kafka-server.properties | 135 +++++++++++++++ modules/mod/infra/kafka.Dockerfile | 4 +- settings.gradle.kts | 1 + 13 files changed, 506 insertions(+), 7 deletions(-) create mode 100644 modules/factorio-events-kafka-pipe/build.gradle.kts create mode 100644 modules/factorio-events-kafka-pipe/docker-compose.yml create mode 100644 modules/factorio-events-kafka-pipe/kafka-pipe-jar.Dockerfile create mode 100644 modules/factorio-events-kafka-pipe/kafka-pipe.Dockerfile create mode 100644 modules/factorio-events-kafka-pipe/kafka-pipe.sh create mode 100644 modules/factorio-events-kafka-pipe/pipe-logs-to-kafka.sh create mode 100644 modules/factorio-events-kafka-pipe/src/main/kotlin/fekp.kt create mode 100644 modules/mod/infra/kafka-server.properties diff --git a/buildSrc/src/main/kotlin/dev/adamko/factoriowebmap/archetype/kotlin-jvm.gradle.kts b/buildSrc/src/main/kotlin/dev/adamko/factoriowebmap/archetype/kotlin-jvm.gradle.kts index 5f33e8a..ef894b0 100644 --- a/buildSrc/src/main/kotlin/dev/adamko/factoriowebmap/archetype/kotlin-jvm.gradle.kts +++ b/buildSrc/src/main/kotlin/dev/adamko/factoriowebmap/archetype/kotlin-jvm.gradle.kts @@ -8,6 +8,9 @@ plugins { } kotlin { + jvmToolchain { + (this as JavaToolchainSpec).languageVersion.set(JavaLanguageVersion.of(11)) // "8" + } } dependencies { @@ -16,8 +19,9 @@ dependencies { tasks.withType().configureEach { kotlinOptions.freeCompilerArgs += listOf( - "-Xopt-in=kotlin.OptIn", - "-Xopt-in=kotlin.ExperimentalStdlibApi", - "-Xopt-in=kotlin.time.ExperimentalTime", + "-Xopt-in=kotlin.OptIn", + "-Xopt-in=kotlin.ExperimentalStdlibApi", + "-Xopt-in=kotlin.time.ExperimentalTime", + "-Xopt-in=kotlinx.coroutines.ExperimentalCoroutinesApi", ) } \ No newline at end of file diff --git a/modules/factorio-events-kafka-pipe/build.gradle.kts b/modules/factorio-events-kafka-pipe/build.gradle.kts new file mode 100644 index 0000000..fc935c8 --- /dev/null +++ b/modules/factorio-events-kafka-pipe/build.gradle.kts @@ -0,0 +1,155 @@ +import org.jetbrains.kotlin.util.parseSpaceSeparatedArgs + +plugins { + id("dev.adamko.factoriowebmap.archetype.kotlin-jvm") + + application + +// id("com.github.johnrengelman.shadow") version "7.1.0" +} + +description = + """ + Send events from a Factorio server to a Kafka topic + """.trimIndent() + +val projectId: String by project.extra + +dependencies { + implementation(libs.bundles.logging) + implementation("eu.jrie.jetbrains:kotlin-shell-core:0.2.1") +} + +application { + mainClass.set("factorioEventsKafkaPipeKt") +} + +java { + withSourcesJar() +} + +tasks.jar { + manifest { + attributes( + "Implementation-Title" to project.name, + "Implementation-Version" to project.version, + "Main-Class" to application.mainClass, + ) + } + archiveBaseName.set(project.name) + duplicatesStrategy = DuplicatesStrategy.EXCLUDE + + from( + configurations.runtimeClasspath.map { config -> + config.map { + logger.trace("Adding runtimeClasspath file to jar - $it") + when (it.isDirectory) { + true -> it + false -> zipTree(it) + } + } + } + ) + +} + +tasks.create("dockerBuildPdl2k") { + group = projectId + + dependsOn(tasks.jar) + + val jar = tasks.jar.get().archiveFile.get() + logger.lifecycle("building docker with jar $jar") + inputs.file(jar) + + commandLine( + parseSpaceSeparatedArgs( + "docker-compose build pdl2k --build-arg JAR_LOCATION=${jar.asFile.canonicalPath}" + ) + ) + +// inputs(tasks.jar.get().archiveFile.get()) + +// commandLine( +// "docker", +// "build", +// "--build-arg", +// "JAR_FILE=", +// "-t dev-adamko/pdl2k" +// ) +} + +//tasks.jar.get().archiveFile.get().asFile.canonicalPath + +tasks.create("runInDocker") { + group = projectId +// docker run -v `pwd`:/mnt java:8 java -jar /mnt/myapp.jar + + dependsOn(tasks.jar) + val jarFile = tasks.jar.get().archiveFile.get().asFile + inputs.file(jarFile) + + commandLine( + listOf( + "docker", + "run", + "--rm", + "--name", project.name, +// """--workdir="/app" """, + "-v", ".//${jarFile.toRelativeString(project.projectDir).replace("\\", "//")}:/mnt/", + "openjdk:11.0.6-jre-buster", + "ls", "-la", "/mnt/*", +// "java", "-jar", "/mnt/${jarFile.name}", + ) + ) + + doFirst { + logger.lifecycle("Executing [$commandLine]") + } +} + +//afterEvaluate { +// tasks { +// create("jar2", Jar::class).apply { +// dependsOn("jvmMainClasses", "jsJar") +// group = "jar" +// manifest { +// attributes( +// mapOf( +// "Implementation-Title" to rootProject.name, +// "Implementation-Version" to rootProject.version, +// "Timestamp" to System.currentTimeMillis(), +// "Main-Class" to application.mainClass +// ) +// ) +// } +// +// from( +// configurations +// ) +// +// val dependencies = +// configurations["jvmRuntimeClasspath"] +// .filter { it.name.endsWith(".jar") } + +// project.tasks["jvmJar"].outputs.files +// dependencies.forEach { from(zipTree(it)) } +// into("/lib") +// } +// } +//} +//val fatJar = task("fatJar", type = Jar::class) { +// baseName = "safescape-lib-${project.name}" +// // manifest Main-Class attribute is optional. +// // (Used only to provide default main class for executable jar) +// from(configurations.runtimeClasspath.map({ if (it.isDirectory) it else zipTree(it) })) +// with(tasks["jar"] as CopySpec) +//} +// +//tasks { +// "build" { +// dependsOn(fatJar) +// } +//} +//from(configurations.runtimeClasspath) { +// into("lib") +//} \ No newline at end of file diff --git a/modules/factorio-events-kafka-pipe/docker-compose.yml b/modules/factorio-events-kafka-pipe/docker-compose.yml new file mode 100644 index 0000000..4cc28cf --- /dev/null +++ b/modules/factorio-events-kafka-pipe/docker-compose.yml @@ -0,0 +1,54 @@ +version: "3.9" + +services: + kp2: + image: openjdk:11-jre-buster + container_name: "factorio-events-kafka-pipe" + build: + dockerfile: "kafka-pipe.Dockerfile" + networks: + - "factorio-events-nw" + environment: + FACTORIO_SERVER_CONTAINER_NAME: "factorio-server" + KAFKA_HOST: "kafka:29092" + volumes: + # Unix and Windows? + # - "//./pipe/docker_engine:/var/run/docker.sock" + # - "//./pipe/docker_engine:/pipe/docker_engine" + - "//var/run/docker.sock:/var/run/docker.sock" + # - "/var/run/docker.sock:/var/run/docker.sock" + # Windows? + # - "\\.\\pipe\\docker_engine:\\.\\pipe\\docker_engine" + # Windows? + # - "/usr/local/bin/docker:/usr/bin/docker" + + +# kp3: +# image: edenhill/kcat:1.7.0 +# container_name: kafka-pipe3 +# networks: +# - "factorio-events-nw" +# volumes: +# - "./kafka-pipe.sh:/kafka-pipe.sh" +# entrypoint: +# # - "ls" +# # - "-la" +# # - "&&" +# - "/bin/sh" +# - "./kafka-pipe.sh" +# - "factorio-server" +# - "kafka-kraft" + +### + +# volumes: +# - "./build/libs/:/app" +# command: +# - "java" +# - "-jar" +# - "/app/pipe-docker-logs-to-kafka-0.0.2.jar" + +networks: + factorio-events-nw: + name: "factorio-events-nw" + external: true diff --git a/modules/factorio-events-kafka-pipe/kafka-pipe-jar.Dockerfile b/modules/factorio-events-kafka-pipe/kafka-pipe-jar.Dockerfile new file mode 100644 index 0000000..16ded19 --- /dev/null +++ b/modules/factorio-events-kafka-pipe/kafka-pipe-jar.Dockerfile @@ -0,0 +1,12 @@ +ARG JAR_LOCATION + +FROM openjdk:11.0.6-jre-buster +ARG JAR_LOCATION + +RUN apt-get update && apt-get install -y \ + curl \ + && rm -rf /var/lib/apt/lists/* + +COPY JAR_LOCATION /app.jar + +ENTRYPOINT ["java","-jar","/app.jar"] diff --git a/modules/factorio-events-kafka-pipe/kafka-pipe.Dockerfile b/modules/factorio-events-kafka-pipe/kafka-pipe.Dockerfile new file mode 100644 index 0000000..9fc1c79 --- /dev/null +++ b/modules/factorio-events-kafka-pipe/kafka-pipe.Dockerfile @@ -0,0 +1,31 @@ + +#FROM debian:buster-slim +FROM docker:dind-rootless + +ENV FACTORIO_SERVER_CONTAINER_NAME="factorio-server" +ENV KAFKA_HOST="kafka-kraft" + +#RUN apt-get update && apt-get install -y --no-install-recommends \ +# curl \ +# docker \ +# kafkacat + +# root is required for following Docker logs +USER root + +RUN apk add --update \ + curl \ + kafkacat +#USER rootless + +# \ +# && rm -rf /var/lib/apt/lists/* + +COPY kafka-pipe.sh kafka-pipe.sh + +#RUN useradd --user-group --system --create-home --no-log-init the_engineer +#USER the_engineer + +ENTRYPOINT [ "/bin/sh", "/kafka-pipe.sh" ] +#ENTRYPOINT /bin/bash +#CMD exec /kafka-pipe.sh diff --git a/modules/factorio-events-kafka-pipe/kafka-pipe.sh b/modules/factorio-events-kafka-pipe/kafka-pipe.sh new file mode 100644 index 0000000..0c8fa68 --- /dev/null +++ b/modules/factorio-events-kafka-pipe/kafka-pipe.sh @@ -0,0 +1,21 @@ +#!/usr/bin/bash + +# Pipe Docker logs from a Factorio server to a Kafka topic + +echo "Piping logs from $FACTORIO_SERVER_CONTAINER_NAME to $KAFKA_HOST" + +trap endProcess SIGINT +endProcess() { + echo "Exiting Factorio Events Kafka Pipe" + exit +} + +while :; do + docker logs --tail 0 -f "$FACTORIO_SERVER_CONTAINER_NAME" | + sed -n -e 's/^FactorioEvent: //p' | + kafkacat -P -b "$KAFKA_HOST" -t factorio-server-log + + echo "Error - retrying in 10 seconds" + + sleep 10 +done diff --git a/modules/factorio-events-kafka-pipe/pipe-logs-to-kafka.sh b/modules/factorio-events-kafka-pipe/pipe-logs-to-kafka.sh new file mode 100644 index 0000000..d89e815 --- /dev/null +++ b/modules/factorio-events-kafka-pipe/pipe-logs-to-kafka.sh @@ -0,0 +1,50 @@ +#!/usr/bin/zsh + +# stream logs from 'factorio' container +# follow the logs, and +# combine stdout and stderr +# only match lines that start with 'FactorioEvent: ', +# and only return the stuff *after* 'FactorioEvent: ' +docker logs --tail 0 -f factorio-server 2>&1 | + sed -n -e 's/^FactorioEvent: //p' | + kafkacat -P -b localhost -t factorio-server-log + +# listen to events +# kafkacat -C -b localhost -t factorio-server-log | jq + +# curl -sS --unix-socket /var/run/docker.sock "http://localhost/v1.41/events?filters={"service":"factorio-server"}" --output - | jq + +curl --no-buffer -sS \ + --unix-socket /var/run/docker.sock \ + "http://localhost/v1.41/containers/factorio-server/logs?stdout=true&follow=true&tail=0" | + tail -c+9 +# --output - + +curl \ + -sS \ + --unix-socket /var/run/docker.sock \ + -X POST \ + --insecure "http://localhost/v1.41/containers/factorio-server/attach?logs=true&stream=true&stdout=true&follow=true" | + tail -c+9 + +# sed 's/^........//' + + +# https://stackoverflow.com/questions/42873285/curl-retry-mechanism +# --retry-all-errors + +# --header "Sec-WebSocket-Key: SGVsbG8sIHdvcmxkIQ==" \ +# --header "Sec-WebSocket-Version: 13" \ +# --header "Host: localhost:80" \ +# --header "Origin: http://localhost:80" \ + +curl --include \ + --unix-socket /var/run/docker.sock \ + --no-buffer \ + --header "Connection: Upgrade" \ + --header "Upgrade: websocket" \ + --header "Host: localhost:80" \ + --header "Origin: http://localhost:80" \ + --header "Sec-WebSocket-Version: 13" \ + --header "Sec-WebSocket-Key: SGVsbG8sIHdvcmxkIQ==" \ + "http://localhost/v1.41/containers/factorio-server/attach/ws?logs=true&stdout=true" diff --git a/modules/factorio-events-kafka-pipe/src/main/kotlin/fekp.kt b/modules/factorio-events-kafka-pipe/src/main/kotlin/fekp.kt new file mode 100644 index 0000000..be87043 --- /dev/null +++ b/modules/factorio-events-kafka-pipe/src/main/kotlin/fekp.kt @@ -0,0 +1,17 @@ +import eu.jrie.jetbrains.kotlinshell.shell.shell + +fun main() { + shell { + + "echo hello world"() + +// """ +// curl \ +// --no-buffer \ +// -sS \ +// --unix-socket /var/run/docker.sock \ +// "http://localhost/v1.41/containers/factorio-server/logs?stdout=true&follow=true&tail=0" +// """.trimIndent() + + } +} \ No newline at end of file diff --git a/modules/factorio-events-processor/build.gradle.kts b/modules/factorio-events-processor/build.gradle.kts index 924baaa..e5eff9b 100644 --- a/modules/factorio-events-processor/build.gradle.kts +++ b/modules/factorio-events-processor/build.gradle.kts @@ -12,4 +12,3 @@ description = val projectId: String by project.extra val buildDir: Directory = layout.buildDirectory.dir(projectId).get() - diff --git a/modules/mod/infra/docker-compose.yml b/modules/mod/infra/docker-compose.yml index 7ea4ccc..466e4cf 100644 --- a/modules/mod/infra/docker-compose.yml +++ b/modules/mod/infra/docker-compose.yml @@ -8,6 +8,8 @@ services: # driver: "json-file" container_name: "factorio-server" restart: "unless-stopped" + networks: + - "factorio-events-nw" ports: - "34197:34197/udp" # factorio - "27015:27015/tcp" # rcon @@ -29,18 +31,28 @@ services: # - USERNAME=FactorioUsername # - TOKEN=FactorioToken - kafka-kraft: + kafka: image: dev-adamko/kafka-kraft - container_name: "kafka-kraft" + container_name: "kafka" + restart: "unless-stopped" build: dockerfile: "kafka.Dockerfile" target: "kafka-kraft" args: KAFKA_DL_URL: "https://dlcdn.apache.org/kafka/3.0.0/kafka_2.13-3.0.0.tgz" + networks: + - "factorio-events-nw" ports: - "9092:9092" - "9093:9093" +# environment: +# KAFKA_LISTENERS: "PLAINTEXT://kafka-kraft:29092,PLAINTEXT_HOST://localhost:9092,CONTROLLER_HOST://localhost:9093" # ,CONTROLLER://kafka-kraft:29093 +# KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: "PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT,CONTROLLER:PLAINTEXT" #,CONTROLLER_HOST:PLAINTEXT +# KAFKA_INTER_BROKER_LISTENER_NAME: "PLAINTEXT" +# KAFKA_ADVERTISED_LISTENERS: "PLAINTEXT://kafka-kraft:29092,PLAINTEXT_HOST://localhost:9092" +# KAKFA_CONTROLLER_LISTENER_NAMES: "CONTROLLER" #,CONTROLLER_HOST + # kafka-connect: # image: dev-adamko/kafka-connect # container_name: "kafka-connect" @@ -54,3 +66,9 @@ services: # ports: # - "8083:8083" # + + +networks: + factorio-events-nw: + name: "factorio-events-nw" + diff --git a/modules/mod/infra/kafka-server.properties b/modules/mod/infra/kafka-server.properties new file mode 100644 index 0000000..70ebc99 --- /dev/null +++ b/modules/mod/infra/kafka-server.properties @@ -0,0 +1,135 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You 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. + +# +# This configuration file is intended for use in KRaft mode, where +# Apache ZooKeeper is not present. See config/kraft/README.md for details. +# + +############################# Server Basics ############################# + +# The role of this server. Setting this puts us in KRaft mode +process.roles=broker,controller + +# The node id associated with this instance's roles +node.id=1 + +# The connect string for the controller quorum +controller.quorum.voters=1@localhost:9093 + +############################# Socket Server Settings ############################# + +# The address the socket server listens on. It will get the value returned from +# java.net.InetAddress.getCanonicalHostName() if not configured. +# FORMAT: +# listeners = listener_name://host_name:port +# EXAMPLE: +# listeners = PLAINTEXT://your.host.name:9092 +#listeners=PLAINTEXT://:9092,CONTROLLER://:9093 +#listeners=PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092,CONTROLLER://localhost:9093 +listeners=PLAINTEXT://:29092,PLAINTEXT_HOST://:9092,CONTROLLER://:9093 + +inter.broker.listener.name=PLAINTEXT +#KAFKA_INTER_BROKER_LISTENER_NAME: "PLAINTEXT" + +# Hostname and port the broker will advertise to producers and consumers. If not set, +# it uses the value for "listeners" if configured. Otherwise, it will use the value +# returned from java.net.InetAddress.getCanonicalHostName(). +#advertised.listeners=PLAINTEXT://localhost:9092 +advertised.listeners=PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092 + +# Listener, host name, and port for the controller to advertise to the brokers. If +# this server is a controller, this listener must be configured. +controller.listener.names=CONTROLLER + +# Maps listener names to security protocols, the default is for them to be the same. See the config documentation for more details +#listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL +listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT +# ,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL + +# The number of threads that the server uses for receiving requests from the network and sending responses to the network +num.network.threads=3 + +# The number of threads that the server uses for processing requests, which may include disk I/O +num.io.threads=8 + +# The send buffer (SO_SNDBUF) used by the socket server +socket.send.buffer.bytes=102400 + +# The receive buffer (SO_RCVBUF) used by the socket server +socket.receive.buffer.bytes=102400 + +# The maximum size of a request that the socket server will accept (protection against OOM) +socket.request.max.bytes=104857600 + + +############################# Log Basics ############################# + +# A comma separated list of directories under which to store log files +log.dirs=/tmp/kraft-combined-logs + +# The default number of log partitions per topic. More partitions allow greater +# parallelism for consumption, but this will also result in more files across +# the brokers. +num.partitions=1 + +# The number of threads per data directory to be used for log recovery at startup and flushing at shutdown. +# This value is recommended to be increased for installations with data dirs located in RAID array. +num.recovery.threads.per.data.dir=1 + +############################# Internal Topic Settings ############################# +# The replication factor for the group metadata internal topics "__consumer_offsets" and "__transaction_state" +# For anything other than development testing, a value greater than 1 is recommended to ensure availability such as 3. +offsets.topic.replication.factor=1 +transaction.state.log.replication.factor=1 +transaction.state.log.min.isr=1 + +############################# Log Flush Policy ############################# + +# Messages are immediately written to the filesystem but by default we only fsync() to sync +# the OS cache lazily. The following configurations control the flush of data to disk. +# There are a few important trade-offs here: +# 1. Durability: Unflushed data may be lost if you are not using replication. +# 2. Latency: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush. +# 3. Throughput: The flush is generally the most expensive operation, and a small flush interval may lead to excessive seeks. +# The settings below allow one to configure the flush policy to flush data after a period of time or +# every N messages (or both). This can be done globally and overridden on a per-topic basis. + +# The number of messages to accept before forcing a flush of data to disk +#log.flush.interval.messages=10000 + +# The maximum amount of time a message can sit in a log before we force a flush +#log.flush.interval.ms=1000 + +############################# Log Retention Policy ############################# + +# The following configurations control the disposal of log segments. The policy can +# be set to delete segments after a period of time, or after a given size has accumulated. +# A segment will be deleted whenever *either* of these criteria are met. Deletion always happens +# from the end of the log. + +# The minimum age of a log file to be eligible for deletion due to age +log.retention.hours=168 + +# A size-based retention policy for logs. Segments are pruned from the log unless the remaining +# segments drop below log.retention.bytes. Functions independently of log.retention.hours. +#log.retention.bytes=1073741824 + +# The maximum size of a log segment file. When this size is reached a new log segment will be created. +log.segment.bytes=1073741824 + +# The interval at which log segments are checked to see if they can be deleted according +# to the retention policies +log.retention.check.interval.ms=300000 diff --git a/modules/mod/infra/kafka.Dockerfile b/modules/mod/infra/kafka.Dockerfile index 48e4127..dca9f09 100644 --- a/modules/mod/infra/kafka.Dockerfile +++ b/modules/mod/infra/kafka.Dockerfile @@ -41,8 +41,10 @@ RUN ./bin/kafka-storage.sh format \ --config ./config/kraft/server.properties \ --cluster-id "$(cat cluster_id)" +COPY ./kafka-server.properties ./server.properties + # launch the broker in KRaft mode, which means that it runs without ZooKeeper -ENTRYPOINT ["./bin/kafka-server-start.sh", "./config/kraft/server.properties"] +ENTRYPOINT ["./bin/kafka-server-start.sh", "./server.properties"] ### Kafka Connect ## diff --git a/settings.gradle.kts b/settings.gradle.kts index c46dc6e..c15467c 100644 --- a/settings.gradle.kts +++ b/settings.gradle.kts @@ -7,6 +7,7 @@ include( ":modules:kt-rcon", ":modules:factorio-events-data-model", ":modules:factorio-events-processor", + ":modules:factorio-events-kafka-pipe", )