From 19d351f1d95cb11d1ae0da851b108613fa1e9e58 Mon Sep 17 00:00:00 2001
From: Casper Thygesen <cth@trifork.com>
Date: Mon, 27 May 2024 18:01:29 +0000
Subject: [PATCH] docker compose up kafka-to-http http-to-kafka

Signed-off-by: Casper Thygesen <cth@trifork.com>
---
 resources/docker/kafka/docker-compose.yaml | 22 ++++++++++
 resources/docker/kafka/http.yaml           | 48 ++++++++++++++++++++++
 resources/docker/kafka/stream_in.yaml      | 33 +++++++++++++++
 3 files changed, 103 insertions(+)
 create mode 100644 resources/docker/kafka/http.yaml
 create mode 100644 resources/docker/kafka/stream_in.yaml

diff --git a/resources/docker/kafka/docker-compose.yaml b/resources/docker/kafka/docker-compose.yaml
index 086eb34e44..9d65a7dfdf 100644
--- a/resources/docker/kafka/docker-compose.yaml
+++ b/resources/docker/kafka/docker-compose.yaml
@@ -18,4 +18,26 @@ services:
     command: [ '-w', '-c', '/benthos.yaml' ]
     volumes:
       - ./out.yaml:/benthos.yaml
+    network_mode: host
+  
+  kafka-to-http:
+    image: jeffail/benthos:local
+    build:
+      context: ../../../
+      dockerfile: ./resources/docker/Dockerfile
+    command: [ '-w', '-c', '/benthos.yaml' ]
+    ports:
+      - "4195:4195"
+    volumes:
+      - ./http.yaml:/benthos.yaml
+    network_mode: host
+
+  http-to-kafka:
+    image: jeffail/benthos:local
+    build:
+      context: ../../../
+      dockerfile: ./resources/docker/Dockerfile
+    command: [ '-w', '-c', '/benthos.yaml' ]
+    volumes:
+      - ./stream_in.yaml:/benthos.yaml
     network_mode: host
\ No newline at end of file
diff --git a/resources/docker/kafka/http.yaml b/resources/docker/kafka/http.yaml
new file mode 100644
index 0000000000..d8b3cb8d00
--- /dev/null
+++ b/resources/docker/kafka/http.yaml
@@ -0,0 +1,48 @@
+# todo: read until <https://www.benthos.dev/docs/components/inputs/read_until> ?
+http:
+  address: 0.0.0.0:4195
+  enabled: true
+  root_path: /kafka
+  debug_endpoints: false
+
+logger:
+  level: DEBUG
+  format: logfmt
+  add_timestamp: true
+  static_fields:
+    "@service": benthos
+
+input:
+  kafka_franz:
+    seed_brokers: [localhost:9092] 
+    topics: [JobNameInputTopic] 
+    consumer_group: benthos2
+    sasl:
+      - mechanism: OAUTHBEARER 
+        tokenEndpoint: http://localhost:1852/realms/local-development/protocol/openid-connect/token
+        scope: "kafka" 
+        clientId: "default-access" 
+        clientSecret: "default-access-secret" 
+
+pipeline:
+  processors:
+    - bloblang: |
+        root = this
+        meta "content-type" = "application/json"
+
+    - catch:
+        - log:
+            level: ERROR
+            message: ${! error() }
+        - bloblang: root = deleted()
+
+output:
+  label: "http_out"
+  http_server:
+    address: "" # global
+    timeout: 5s
+    path: /get
+    stream_path: /get/stream
+    ws_path: /get/ws
+    allowed_verbs:
+      - GET
\ No newline at end of file
diff --git a/resources/docker/kafka/stream_in.yaml b/resources/docker/kafka/stream_in.yaml
new file mode 100644
index 0000000000..eafb020365
--- /dev/null
+++ b/resources/docker/kafka/stream_in.yaml
@@ -0,0 +1,33 @@
+http:
+  enabled: false
+
+input:
+  label: ""
+  http_client:
+    url: "http://localhost:4195/kafka/get/stream"
+    verb: GET
+    headers: {}
+    rate_limit: ""
+    timeout: 5s
+    payload: ""
+    drop_on: 
+      - 408 # timeout e.g no more data.
+    stream:
+      enabled: true
+      reconnect: true
+      scanner:
+        lines: {} # todo
+    auto_replay_nacks: true
+
+output:
+  label: sink
+  kafka_franz:
+    seed_brokers: [localhost:9092] # No default (required)
+    topic: localstream_in # No default (required)
+    key: "" # No default (optional)
+    sasl:
+      - mechanism: OAUTHBEARER
+        tokenEndpoint: http://localhost:1852/realms/local-development/protocol/openid-connect/token
+        scope: "kafka"
+        clientId: "default-access"
+        clientSecret: "default-access-secret"