first commit

.gitignore

@@ -0,0 +1,78 @@
+# Ignore node files
+node_modules
+dist
+.cache
+
+# Ignore editor workspace files
+.vscode
+.sublime-*
+.atom
+.editorconfig
+.nyc_output
+coverage
+*.lcov
+*.swp
+*.swo
+
+# Ignore all logfiles and tempfiles.
+/log/*
+/tmp/*
+!/log/.keep
+!/tmp/.keep
+
+# Ignore version control
+.tool-versions
+.*-version
+
+# Ignore env
+/.env
+.env.local
+compose.override.yml
+
+# Ignore brew
+Brewfile.lock.json
+
+# Compiled class files
+*.class
+
+# SBT specific
+target/
+lib_managed/
+src_managed/
+project/project/
+project/target/
+
+# Package Files #
+*.jar
+*.war
+*.ear
+*.zip
+*.tar.gz
+*.rar
+
+# IntelliJ IDEA specific
+.idea/
+*.iml
+*.iws
+*.ipr
+
+# Eclipse specific
+.classpath
+.project
+.settings/
+
+# Logs and databases
+*.log
+
+# OS generated files #
+.DS_Store
+.DS_Store?
+._*
+.Spotlight-V100
+.Trashes
+ehthumbs.db
+Thumbs.db
+
+# Metals - Scala language server
+.metals/
+.bloop/

Dockerfile

@@ -0,0 +1,45 @@
+FROM eclipse-temurin:17-jre AS builder
+
+ARG SCALA_VERSION=2.13.12
+ARG SBT_VERSION=1.9.8
+ARG NEO4J_CONNECTOR_VERSION=5.0.3
+ARG GUAVA_VERSION=32.1.2-jre
+
+RUN apt-get update && apt-get install -y wget unzip && \
+    apt-get clean && \
+    rm -rf /var/lib/apt/lists/*
+
+# Install Scala
+RUN wget -nv https://downloads.lightbend.com/scala/$SCALA_VERSION/scala-$SCALA_VERSION.tgz && \
+    tar xvf scala-$SCALA_VERSION.tgz && \
+    mv scala-$SCALA_VERSION /usr/share/scala && \
+    ln -s /usr/share/scala/bin/* /usr/bin/ && \
+    rm scala-$SCALA_VERSION.tgz
+
+# Install sbt
+RUN wget -nv https://github.com/sbt/sbt/releases/download/v$SBT_VERSION/sbt-$SBT_VERSION.tgz && \
+    tar -xzf sbt-$SBT_VERSION.tgz -C /usr/local --strip-components=1 && \
+    rm sbt-$SBT_VERSION.tgz
+
+# Be sure this aligns with CONNECT_PLUGIN_PATH
+WORKDIR /opt/kafka/connect-plugins
+
+# Download and extract the Kafka Connect Neo4j Connector
+RUN wget -nv -O /tmp/neo4j-connector.zip https://github.com/neo4j-contrib/neo4j-streams/releases/download/$NEO4J_CONNECTOR_VERSION/neo4j-kafka-connect-neo4j-$NEO4J_CONNECTOR_VERSION.zip && \
+    unzip /tmp/neo4j-connector.zip && \
+    rm /tmp/neo4j-connector.zip
+
+# Download Guava, as the dependency is oddly missing from the release.
+RUN wget -P ./neo4j-kafka-connect-neo4j-$NEO4J_CONNECTOR_VERSION/lib https://repo1.maven.org/maven2/com/google/guava/guava/$GUAVA_VERSION/guava-$GUAVA_VERSION.jar
+
+WORKDIR /usr/src/app
+
+COPY . .
+
+RUN sbt clean assembly
+
+FROM redpandadata/connectors:v1.0.13
+
+# Import the compiled binaries from the first stage.
+COPY --from=builder /usr/src/app/target/scala-*/*.jar /opt/kafka/connect-plugins/example-connector/
+COPY --from=builder /opt/kafka/connect-plugins/ /opt/kafka/connect-plugins/

build.sbt

@@ -0,0 +1,24 @@
+name := "ExampleConnector"
+
+version := "0.1"
+
+scalaVersion := "2.13.12"
+
+// Resolvers for additional repositories if needed
+resolvers ++= Seq(
+  "confluent" at "https://packages.confluent.io/maven/",
+)
+
+// Library dependencies
+libraryDependencies ++= Seq(
+  "org.apache.kafka" % "connect-api" % "3.6.1",
+  "org.apache.kafka" %% "kafka" % "3.6.1",
+  "org.playframework" %% "play-json" % "3.0.1",
+  // Add any additional dependencies here
+)
+
+assembly / assemblyMergeStrategy := {
+  case PathList("META-INF", "services", xs @ _*) => MergeStrategy.concat
+  case PathList("META-INF", xs @ _*) => MergeStrategy.discard
+  case x => MergeStrategy.first
+}

compose.yml

@@ -0,0 +1,126 @@
+services:
+  sbt:
+    tty: true
+    stdin_open: true
+    build:
+      context: .
+      target: builder
+    entrypoint: sbt
+    command: help
+    volumes:
+      - .:/usr/src/app:delegated
+      - sbt_ivy_cache:/root/.ivy2:cached
+      - sbt_cache:/root/.sbt:cached
+      - coursier_cache:/root/.cache/coursier:cached
+      - coursier_v1_cache:/root/.coursier/cache/v1:cached
+
+  redpanda:
+    image: redpandadata/redpanda:v23.3.1
+    entrypoint: rpk
+    command:
+      - redpanda
+      - start
+      - --mode
+      - dev-container
+      - --kafka-addr
+      - internal://0.0.0.0:9092,external://0.0.0.0:19092
+      # Address the broker advertises to clients that connect to the Kafka API.
+      # Use the internal addresses to connect to the Redpanda brokers
+      # from inside the same Docker network.
+      # Use the external addresses to connect to the Redpanda brokers
+      # from outside the Docker network.
+      - --advertise-kafka-addr
+      - internal://redpanda:9092,external://localhost:19092
+      - --pandaproxy-addr
+      - internal://0.0.0.0:8082,external://0.0.0.0:18082
+      # Address the broker advertises to clients that connect to the HTTP Proxy.
+      - --advertise-pandaproxy-addr
+      - internal://redpanda:8082,external://localhost:18082
+      - --schema-registry-addr
+      - internal://0.0.0.0:8081,external://0.0.0.0:18081
+      # Redpanda brokers use the RPC API to communicate with each other internally.
+      - --rpc-addr
+      - redpanda:33145
+      - --advertise-rpc-addr
+      - redpanda:33145
+    ports:
+      - 18081:18081
+      - 18082:18082
+      - 19092:19092
+      - 19644:9644
+    volumes:
+      - redpanda:/var/lib/redpanda/data
+      - ./data/:/opt/kafka/data:ro
+
+  console:
+    image: redpandadata/console:v2.3.8
+    entrypoint: /bin/sh
+    command: -c "echo \"$$CONSOLE_CONFIG_FILE\" > /tmp/config.yml; /app/console"
+    environment:
+      CONFIG_FILEPATH: /tmp/config.yml
+      CONSOLE_CONFIG_FILE: |
+        kafka:
+          brokers: ["redpanda:9092"]
+          schemaRegistry:
+            enabled: true
+            urls: ["http://redpanda:8081"]
+        redpanda:
+          adminApi:
+            enabled: true
+            urls: ["http://redpanda:9644"]
+        connect:
+          enabled: true
+          clusters:
+            - name: local-connect-cluster
+              url: "http://connect:8083"
+    ports:
+      - 8080:8080
+    depends_on:
+      - redpanda
+
+  connect:
+    build:
+      context: .
+    ports:
+        - 8083:8083
+    depends_on:
+      - redpanda
+    environment:
+      CONNECT_BOOTSTRAP_SERVERS: redpanda:9092
+      CONNECT_LOG_LEVEL: debug
+      CONNECT_METRICS_ENABLED: false
+      CONNECT_PLUGIN_PATH: /opt/kafka/connect-plugins
+      CONNECT_CONFIGURATION: |
+        key.converter=org.apache.kafka.connect.storage.StringConverter
+        value.converter=org.apache.kafka.connect.json.JsonConverter
+        value.converter.schemas.enable=false
+        group.id=connectors-group
+        offset.storage.topic=_connectors_offsets
+        config.storage.topic=_connectors_configs
+        status.storage.topic=_connectors_status
+        config.storage.replication.factor=-1
+        offset.storage.replication.factor=-1
+        status.storage.replication.factor=-1
+        config.providers=file
+        config.providers.file.class=org.apache.kafka.common.config.provider.FileConfigProvider
+    volumes:
+      - ./connect-password/:/opt/kafka/connect-password:ro
+      - ./connect-plugins/:/opt/kafka/connect-plugins:ro
+
+  neo4j:
+    image: neo4j:5
+    ports:
+      - 7474:7474 # HTTP
+      - 7687:7687 # Bolt
+    environment:
+      - NEO4J_AUTH=neo4j/password
+    volumes:
+      - neo4j:/data
+
+volumes:
+  sbt_ivy_cache: {}
+  sbt_cache: {}
+  coursier_cache: {}
+  coursier_v1_cache: {}
+  redpanda: {}
+  neo4j: {}

connect-config.json

@@ -0,0 +1,14 @@
+{
+  "name": "Neo4jSinkConnector",
+  "config": {
+    "topics": "events",
+    "connector.class": "streams.kafka.connect.sink.Neo4jSinkConnector",
+    "transforms": "cud",
+    "transforms.cud.type": "com.orbservability.connector.example.SampleTransformer",
+    "neo4j.server.uri": "bolt://neo4j:7687",
+    "neo4j.authentication.basic.username": "${file:/opt/kafka/connect-password/neo4j.properties:username}",
+    "neo4j.authentication.basic.password": "${file:/opt/kafka/connect-password/neo4j.properties:password}",
+    "neo4j.encryption.enabled": false,
+    "neo4j.topic.cud": "events"
+  }
+}

connect-password/neo4j.properties

@@ -0,0 +1,2 @@
+username=neo4j
+password=password

data/import.ndjson

@@ -0,0 +1,3 @@
+{"followerID":"1","followerName":"Sue","followeeID":"2","followeeName":"Bob"}
+{"followerID":"1","followerName":"Sue","followeeID":"3","followeeName":"Jon"}
+{"followerID":"3","followerName":"Jon","followeeID":"1","followeeName":"Sue"}

project/assembly.sbt

@@ -0,0 +1 @@
+addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "2.1.5")

project/build.properties

@@ -0,0 +1 @@
+sbt.version=1.9.8

project/metals.sbt

@@ -0,0 +1,6 @@
+// DO NOT EDIT! This file is auto-generated.
+
+// This file enables sbt-bloop to create bloop config files.
+
+addSbtPlugin("ch.epfl.scala" % "sbt-bloop" % "1.5.13")
+

src/main/resources/META-INF/services/org.apache.kafka.connect.transforms.Transformation

@@ -0,0 +1 @@
+com.orbservability.connector.example.SampleTransformer

src/main/scala/com/orbservability/connector/example/Event.scala

@@ -0,0 +1,20 @@
+package com.orbservability.connector.example
+
+case class Event(
+  followerID: String,
+  followerName: String,
+  followeeID: String,
+  followeeName: String,
+
+)
+
+object Event {
+  def fromMap(valueMap: scala.collection.mutable.Map[String,Any]): Event = {
+    Event(
+      followerID = valueMap("followerID").toString,
+      followerName = valueMap("followerName").toString,
+      followeeID = valueMap("followeeID").toString,
+      followeeName = valueMap("followeeName").toString,
+    )
+  }
+}