feat: add kafka migration (#330)

* Add Migration tooling

* Update push-image.sh

* Rename PSL_job.json to pubsub_lite_job.json

* Update Dockerfile

* Rename start-psl-connector.sh to start-pubsub-lite-connector.sh

migration/.gcp/gmk_bootstrap_servers

@@ -0,0 +1 @@
+bootstrap.<google-managed-kafka-cluster-name>.<google-managed-kafka-cluster-region name>.managedkafka.<google-managed-cluster-host-project-name>.cloud.goog:9092

migration/.gcp/gmk_sasl_service_account

@@ -0,0 +1 @@
+<service-account-name>@<gcp-project>.iam.gserviceaccount.com

migration/.gcp/gmk_sasl_service_account_key

@@ -0,0 +1 @@
+<base64 encoded sasl service account key>

migration/.gcp/kafka_config_storage_topic

@@ -0,0 +1 @@
+<kafka topic name used by kafka connect for tracking the config>

migration/.gcp/kafka_connect_group_id

@@ -0,0 +1 @@
+<kafka connect group id(unique per worker group) for the kafka connect workers in distributed mode>

migration/.gcp/kafka_offset_storage_topic

@@ -0,0 +1 @@
+<kafka topic name used by kafka connect for tracking the offsets>

migration/.gcp/kafka_sink_topic

@@ -0,0 +1 @@
+<target sink kafka topic name used by kafka connect for migrating the data from pubsub-lite topic>

migration/.gcp/kafka_ssl_truststore_location

@@ -0,0 +1 @@
+<full path of the ssl truststore jks file location>

migration/.gcp/kafka_ssl_truststore_password

@@ -0,0 +1 @@
+<password for the ssl truststore jks>

migration/.gcp/kafka_status_storage_topic

@@ -0,0 +1 @@
+<kafka topic name used by kafka connect for tracking the status>

migration/.gcp/pubsub_lite_gcp_location

@@ -0,0 +1 @@
+<GCP location for the pubsub lite source subscription to be used for migrating the pubsub lite topic to sink kafka topic>

migration/.gcp/pubsub_lite_gcp_project

@@ -0,0 +1 @@
+<GCP project that hosts the pubsub lite source subscription to be used for migrating the pubsub lite topic to sink kafka topic> 

migration/.gcp/pubsub_lite_gcp_subscription

@@ -0,0 +1 @@
+<pubsub lite source subscription name to be used for migrating the pubsub lite topic to kafka topic> 

migration/.gcp/pubsub_lite_job_name

@@ -0,0 +1 @@
+PubSubLiteSourceConnector

migration/docker/Dockerfile

@@ -0,0 +1,84 @@
+FROM --platform=linux/amd64 eclipse-temurin:21
+ARG KAFKA_VERSION="3.4.0"
+ARG KAFKA_CONNECT_VERSION="${KAFKA_VERSION}"
+ARG KAFKA_SCALA_VERSION="2.13"
+ARG PUBSUB_GROUP_KAFKA_CONNECTOR_VERSION="1.2.0"
+ARG KAFKA_HOME_ROOT="/opt"
+ARG KAFKA_CONFIG_DIR="${KAFKA_HOME}/config"
+ARG KAFKA_RELEASE="kafka_${KAFKA_SCALA_VERSION}-${KAFKA_VERSION}"
+ARG KAFKA_TARBALL="${KAFKA_RELEASE}.tgz"
+ARG KAFKA_DOWNLOAD_URL="https://archive.apache.org/dist/kafka/${KAFKA_VERSION}/${KAFKA_TARBALL}"
+ENV KAFKA_HEAP_OPTS="-Xms2G -Xmx2G"
+ENV KAFKA_HOME="${KAFKA_HOME_ROOT}/kafka"
+ARG KAFKA_PLUGINS_DIR="${KAFKA_HOME}/plugins"
+# The pubsub-group-kafka-connector file needs to be pre-built/downloaded using maven or other similar tool.
+# References:
+# 1) https://github.com/googleapis/java-pubsub-group-kafka-connector/releases/
+# 2) https://central.sonatype.com/artifact/com.google.cloud/pubsub-group-kafka-connector
+ARG PUBSUB_GROUP_KAFKA_CONNECTOR_JAR="pubsub-group-kafka-connector-${PUBSUB_GROUP_KAFKA_CONNECTOR_VERSION}.jar"
+ARG KAFKA_CONNECT_CONFIGURE_SCRIPT="configure-kafka-connect.sh"
+ARG BUILD_KAFKA_CONNECT_STARTUP_SCRIPT="start-kafka-connect.sh"
+ARG BUILD_PUBSUB_LITE_JOB_STARTUP_SCRIPT="start-pubsub-lite-connector.sh"
+ARG BUILD_KAFKA_CONNECT_CONFIG_FILE="kafka-connect.properties"
+ARG BUILD_PUBSUB_LITE_JOB_FILE="pubsub_lite_job.json"
+ENV JAVA_HOME="/opt/java/openjdk"
+ENV PATH="${KAFKA_HOME}/bin:${JAVA_HOME}/bin:${PATH}"
+ENV KAFKA_CONNECT_STARTUP_SCRIPT="${KAFKA_HOME}/bin/${BUILD_KAFKA_CONNECT_STARTUP_SCRIPT}"
+ENV PUBSUB_LITE_JOB_STARTUP_SCRIPT="${KAFKA_HOME}/bin/${BUILD_PUBSUB_LITE_JOB_STARTUP_SCRIPT}"
+ENV KAFKA_CONNECT_CONFIG_FILE="${KAFKA_CONFIG_DIR}/${BUILD_KAFKA_CONNECT_CONFIG_FILE}"
+ENV PUBSUB_LITE_JOB_FILE="${KAFKA_CONFIG_DIR}/${BUILD_PUBSUB_LITE_JOB_FILE}"
+
+RUN apt-get -y -qq update \
+    && apt-get -y -qq install iproute2 bind9-dnsutils
+
+RUN wget -q ${KAFKA_DOWNLOAD_URL} \
+    && tar -xzf ${KAFKA_TARBALL} -C ${KAFKA_HOME_ROOT} \
+    && ln -s ${KAFKA_HOME_ROOT}/${KAFKA_RELEASE} ${KAFKA_HOME} \
+    && rm -f ${KAFKA_TARBALL}
+
+RUN mkdir -p ${KAFKA_PLUGINS_DIR}
+COPY ${PUBSUB_GROUP_KAFKA_CONNECTOR_JAR} \
+${KAFKA_PLUGINS_DIR}/${PUBSUB_GROUP_KAFKA_CONNECTOR_JAR}
+COPY ${BUILD_KAFKA_CONNECT_CONFIG_FILE} ${KAFKA_CONNECT_CONFIG_FILE}
+COPY ${BUILD_PUBSUB_LITE_JOB_FILE} ${PUBSUB_LITE_JOB_FILE}
+COPY ${KAFKA_CONNECT_CONFIGURE_SCRIPT} .
+COPY ${BUILD_KAFKA_CONNECT_STARTUP_SCRIPT} ${KAFKA_CONNECT_STARTUP_SCRIPT}
+COPY ${BUILD_PUBSUB_LITE_JOB_STARTUP_SCRIPT} ${PUBSUB_LITE_JOB_STARTUP_SCRIPT}
+RUN chmod +x ${KAFKA_CONNECT_CONFIGURE_SCRIPT}
+RUN chmod +x ${KAFKA_CONNECT_STARTUP_SCRIPT}
+RUN chmod +x ${PUBSUB_LITE_JOB_STARTUP_SCRIPT}
+RUN --mount=type=secret,id=gmk_bootstrap_servers \
+    --mount=type=secret,id=gmk_sasl_service_account \
+    --mount=type=secret,id=gmk_sasl_service_account_key \
+    --mount=type=secret,id=kafka_sink_topic \
+    --mount=type=secret,id=kafka_connect_group_id \
+    --mount=type=secret,id=pubsub_lite_gcp_project \
+    --mount=type=secret,id=pubsub_lite_gcp_location \
+    --mount=type=secret,id=pubsub_lite_gcp_subscription \
+    --mount=type=secret,id=pubsub_lite_job_name \
+    --mount=type=secret,id=kafka_config_storage_topic \
+    --mount=type=secret,id=kafka_offset_storage_topic \
+    --mount=type=secret,id=kafka_status_storage_topic \
+    --mount=type=secret,id=kafka_ssl_truststore_location \
+    --mount=type=secret,id=kafka_ssl_truststore_password \
+    KAFKA_CONNECT_CONFIG_FILE="${KAFKA_CONNECT_CONFIG_FILE}" \
+    KAFKA_BOOTSTRAP_SERVERS="$(cat /run/secrets/gmk_bootstrap_servers)" \
+    KAFKA_SASL_SERVICE_ACCOUNT="$(cat /run/secrets/gmk_sasl_service_account)"\
+    KAFKA_SASL_SERVICE_ACCOUNT_KEY="$(cat /run/secrets/gmk_sasl_service_account_key)" \
+    KAFKA_SINK_TOPIC="$(cat /run/secrets/kafka_sink_topic)" \
+    KAFKA_CONNECT_GROUP_ID="$(cat /run/secrets/kafka_connect_group_id)" \
+    KAFKA_PLUGINS_DIR=${KAFKA_PLUGINS_DIR} \
+    PUBSUB_LITE_GCP_PROJECT="$(cat /run/secrets/pubsub_lite_gcp_project)" \
+    PUBSUB_LITE_GCP_LOCATION="$(cat /run/secrets/pubsub_lite_gcp_location)" \
+    PUBSUB_LITE_GCP_SUBSCRIPTION="$(cat /run/secrets/pubsub_lite_gcp_subscription)" \
+    PUBSUB_LITE_JOB_NAME="$(cat /run/secrets/pubsub_lite_job_name)" \
+    KAFKA_CONFIG_STORAGE_TOPIC="$(cat /run/secrets/kafka_config_storage_topic)" \
+    KAFKA_OFFSET_STORAGE_TOPIC="$(cat /run/secrets/kafka_offset_storage_topic)" \
+    KAFKA_STATUS_STORAGE_TOPIC="$(cat /run/secrets/kafka_status_storage_topic)" \
+    KAFKA_SSL_TRUSTSTORE_LOCATION="$(cat /run/secrets/kafka_ssl_truststore_location)" \
+    KAFKA_SSL_TRUSTSTORE_PASSWORD="$(cat /run/secrets/kafka_ssl_truststore_password)" \
+    ./${KAFKA_CONNECT_CONFIGURE_SCRIPT} \
+    && rm -f ./${KAFKA_CONNECT_CONFIGURE_SCRIPT}
+
+EXPOSE 8083
+CMD ${KAFKA_CONNECT_STARTUP_SCRIPT}

migration/docker/build-image.sh

@@ -0,0 +1,18 @@
+SELF_DIR="$(dirname $(readlink -f $0))"
+SECRETS_DIR="$(dirname ${SELF_DIR})/.gcp"
+docker build --platform=linux/amd64 --file Dockerfile --tag psl-to-gmk:latest \
+  --secret id=gmk_sasl_service_account,src="${SECRETS_DIR}/gmk_sasl_service_account" \
+  --secret id=gmk_sasl_service_account_key,src="${SECRETS_DIR}/gmk_sasl_service_account_key" \
+  --secret id=gmk_bootstrap_servers,src="${SECRETS_DIR}/gmk_bootstrap_servers" \
+  --secret id=kafka_sink_topic,src="${SECRETS_DIR}/kafka_sink_topic" \
+  --secret id=kafka_connect_group_id,src="${SECRETS_DIR}/kafka_connect_group_id" \
+  --secret id=pubsub_lite_gcp_project,src="${SECRETS_DIR}/pubsub_lite_gcp_project" \
+  --secret id=pubsub_lite_gcp_location,src="${SECRETS_DIR}/pubsub_lite_gcp_location" \
+  --secret id=pubsub_lite_gcp_subscription,src="${SECRETS_DIR}/pubsub_lite_gcp_subscription" \
+  --secret id=pubsub_lite_job_name,src="${SECRETS_DIR}/pubsub_lite_job_name" \
+  --secret id=kafka_config_storage_topic,src="${SECRETS_DIR}/kafka_config_storage_topic" \
+  --secret id=kafka_offset_storage_topic,src="${SECRETS_DIR}/kafka_offset_storage_topic" \
+  --secret id=kafka_status_storage_topic,src="${SECRETS_DIR}/kafka_status_storage_topic" \
+  --secret id=kafka_ssl_truststore_location,src="${SECRETS_DIR}/kafka_ssl_truststore_location" \
+  --secret id=kafka_ssl_truststore_password,src="${SECRETS_DIR}/kafka_ssl_truststore_password" \
+  --no-cache .

migration/docker/configure-kafka-connect.sh

@@ -0,0 +1,25 @@
+#!/usr/bin/env bash
+# All the variables must be supplied as environment variables for this script
+# Update Kafka Connect Sink config
+sed -i -e "s#__KAFKA_BOOTSTRAP_SERVERS__#${KAFKA_BOOTSTRAP_SERVERS}#g;" \
+    "${KAFKA_CONNECT_CONFIG_FILE}"
+# Update Kafka Connect internal topics config
+sed -i -e "s#__KAFKA_CONFIG_STORAGE_TOPIC__#${KAFKA_CONFIG_STORAGE_TOPIC}#g; s#__KAFKA_OFFSET_STORAGE_TOPIC__#${KAFKA_OFFSET_STORAGE_TOPIC}#g; s#__KAFKA_STATUS_STORAGE_TOPIC__#${KAFKA_STATUS_STORAGE_TOPIC}#g" \
+    "${KAFKA_CONNECT_CONFIG_FILE}"
+# Update Kafka Connect group id and Kafka Connect plugins directory. Kafka Connect group id needs to be unique and must not conflict with the consumer group ids
+sed -i -e "s#__KAFKA_CONNECT_GROUP_ID__#${KAFKA_CONNECT_GROUP_ID}#g; s#__KAFKA_PLUGINS_DIR__#${KAFKA_PLUGINS_DIR}#g" \
+    "${KAFKA_CONNECT_CONFIG_FILE}"
+# Update Kafka Connect SASL config
+sed -i -e "s#__KAFKA_SASL_SERVICE_ACCOUNT__#${KAFKA_SASL_SERVICE_ACCOUNT}#g; s#__KAFKA_SASL_SERVICE_ACCOUNT_KEY__#${KAFKA_SASL_SERVICE_ACCOUNT_KEY}#g" \
+    "${KAFKA_CONNECT_CONFIG_FILE}"
+# Update Kafka Connect SSL truststore config
+sed -i -e "s#__KAFKA_SSL_TRUSTSTORE_LOCATION__#${KAFKA_SSL_TRUSTSTORE_LOCATION}#g; s#__KAFKA_SSL_TRUSTSTORE_PASSWORD__#${KAFKA_SSL_TRUSTSTORE_PASSWORD}#g" \
+    "${KAFKA_CONNECT_CONFIG_FILE}"
+
+#Update PubSub Lite Job File
+sed -i -e "s#__PUBSUB_LITE_JOB_NAME__#${PUBSUB_LITE_JOB_NAME}#g; s#__KAFKA_SINK_TOPIC__#${KAFKA_SINK_TOPIC}#g; s#__PUBSUB_LITE_GCP_PROJECT__#${PUBSUB_LITE_GCP_PROJECT}#g; s#__PUBSUB_LITE_GCP_LOCATION__#${PUBSUB_LITE_GCP_LOCATION}#g; s#__PUBSUB_LITE_GCP_SUBSCRIPTION__#${PUBSUB_LITE_GCP_SUBSCRIPTION}#g;" \
+    "${PUBSUB_LITE_JOB_FILE}"
+
+#Update PSL Job Start Script
+sed -i -e "s#__PUBSUB_LITE_JOB_NAME__#${PUBSUB_LITE_JOB_NAME}#g;" \
+    "${PSL_JOB_STARTUP_SCRIPT}"

migration/docker/kafka-connect.properties

@@ -0,0 +1,85 @@
+# Copyright 2024 Google LLC
+#
+# Licensed 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.
+# Kafka Connect Distributed Mode Configuration
+
+# Bootstrap servers for Kafka brokers
+bootstrap.servers=__KAFKA_BOOTSTRAP_SERVERS__
+
+# Group ID for Kafka Connect worker
+group.id=__KAFKA_CONNECT_GROUP_ID__
+
+# REST API endpoint for Kafka Connect
+rest.port=8083
+
+# Hostname for REST API endpoint
+rest.host.name=__KAFKA_REST_ADVERTISED_HOST_NAME__
+
+# Client ID for the worker. This will appear in server logs for tracking
+client.id=__KAFKA_CONNECT_WORKER_CLIENT_ID__
+
+# Classpath for plugins (including connectors)
+plugin.path=__KAFKA_PLUGINS_DIR__
+
+# Offset commit interval in milliseconds
+offset.flush.interval.ms=10000
+
+# Enable or disable the internal converter used for offset storage
+config.storage.topic=__KAFKA_CONFIG_STORAGE_TOPIC__
+offset.storage.topic=__KAFKA_OFFSET_STORAGE_TOPIC__
+status.storage.topic=__KAFKA_STATUS_STORAGE_TOPIC__
+
+# Number of worker threads for handling HTTP requests
+rest.advertised.host.name=__KAFKA_REST_ADVERTISED_HOST_NAME__
+rest.advertised.port=8083
+
+# Number of worker threads for handling HTTP requests
+rest.threads.max=50
+
+# Default partition assignment strategy
+partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor
+
+# Kafka Connect-specific settings
+offset.storage.replication.factor=3
+config.storage.replication.factor=3
+status.storage.replication.factor=3
+offset.storage.partitions=25
+status.storage.partitions=5
+
+# SASL auth related configuration
+sasl.mechanism=PLAIN
+security.protocol=SASL_SSL
+sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
+  username="__KAFKA_SASL_SERVICE_ACCOUNT__" \
+  password="__KAFKA_SASL_SERVICE_ACCOUNT_KEY__";
+
+producer.sasl.mechanism=PLAIN
+producer.security.protocol=SASL_SSL
+producer.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
+  username="__KAFKA_SASL_SERVICE_ACCOUNT__" \
+  password="__KAFKA_SASL_SERVICE_ACCOUNT_KEY__";
+
+consumer.sasl.mechanism=PLAIN
+consumer.security.protocol=SASL_SSL
+consumer.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
+  username="__KAFKA_SASL_SERVICE_ACCOUNT__" \
+  password="__KAFKA_SASL_SERVICE_ACCOUNT_KEY__";
+
+# SSL Truststore related configuration
+ssl.truststore.location=__KAFKA_SSL_TRUSTSTORE_LOCATION__
+ssl.truststore.password=__KAFKA_SSL_TRUSTSTORE_PASSWORD__
+
+# Set the key converter for the Pub/Sub Lite source connector.
+key.converter=org.apache.kafka.connect.converters.ByteArrayConverter
+# Set the value converter for the Pub/Sub Lite source connector.
+value.converter=org.apache.kafka.connect.converters.ByteArrayConverter

migration/docker/pubsub_lite_job.json

@@ -0,0 +1,11 @@
+{
+    "name": "__PUBSUB_LITE_JOB_NAME__",
+    "config": {
+        "connector.class": "com.google.pubsublite.kafka.source.PubSubLiteSourceConnector",
+        "tasks.max": "10",
+        "kafka.topic": "__KAFKA_SINK_TOPIC__",
+        "pubsublite.project": "__PUBSUB_LITE_GCP_PROJECT__",
+        "pubsublite.location": "__PUBSUB_LITE_GCP_LOCATION__",
+        "pubsublite.subscription": "__PUBSUB_LITE_GCP_SUBSCRIPTION__"
+        }
+}

migration/docker/push-image.sh

@@ -0,0 +1,21 @@
+# Copyright 2024 Google LLC
+#
+# Licensed 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.
+# Dockerfile for building Kafka Connect image
+DOCKER_IMAGE_NAME="psl-to-gmk"
+DOCKER_IMAGE_TAG=latest
+GCP_PROJECT="<GCP Project>"
+DOCKER_REPOSTORY=gcr.io/${GCP_PROJECT}
+docker tag ${DOCKER_IMAGE_NAME} \
+${DOCKER_REPOSTORY}/${DOCKER_IMAGE_NAME}:${DOCKER_IMAGE_TAG}
+docker push ${DOCKER_REPOSTORY}/${DOCKER_IMAGE_NAME}:${DOCKER_IMAGE_TAG}

migration/docker/start-kafka-connect.sh

@@ -0,0 +1,9 @@
+#!/usr/bin/env bash
+${PSL_JOB_STARTUP_SCRIPT} &
+
+START_SCRIPT="${KAFKA_HOME}/bin/connect-distributed.sh"
+KAFKA_REST_ADVERTISED_HOST_NAME="$(/sbin/ip -o -4 addr list eth0 | awk '{print $4}' | cut -d/ -f1)"
+KAFKA_CONNECT_WORKER_CLIENT_ID="$(hostname --fqdn)"
+sed -i -e "s#__KAFKA_REST_ADVERTISED_HOST_NAME__#${KAFKA_REST_ADVERTISED_HOST_NAME}#g; s#__KAFKA_CONNECT_WORKER_CLIENT_ID__#${KAFKA_CONNECT_WORKER_CLIENT_ID}#g" \
+    "${KAFKA_CONNECT_CONFIG_FILE}"
+${START_SCRIPT} ${KAFKA_CONNECT_CONFIG_FILE}

migration/docker/start-pubsub-lite-connector.sh

@@ -0,0 +1,19 @@
+#!/usr/bin/env bash
+
+#Poll Kafka Connect until it is up
+while true
+do
+    echo "Pinging Connect Rest Endpoint"
+    CONNECT_PING=$(curl localhost:8083 | grep "version")
+    if [[ $CONNECT_PING != "" ]]; then
+        break
+    fi
+    sleep 30
+done
+#Once Kafka Connect is up, if the PubSub Lite migration job 
+#does not yet exist, submit the Job
+CONNECT_JOBS=$(curl localhost:8083/connectors | grep "__PUBSUB_LITE_JOB_NAME__")
+if [[ $CONNECT_JOBS == "" ]]; then
+    echo "No Connect Job found, posting Job"
+    curl -H "Content-Type: application/json" -H "Accept: application/json" --data "@/opt/kafka/config/PSL_job.json" localhost:8083/connectors
+fi

migration/k8s.yaml

@@ -0,0 +1,44 @@
+---
+apiVersion: "apps/v1"
+kind: "Deployment"
+metadata:
+  name: "<workflow_name>"
+  namespace: "default"
+  labels:
+    app: "<workflow_name>"
+spec:
+  replicas: 3
+  selector:
+    matchLabels:
+      app: "<workflow_name>"
+  template:
+    metadata:
+      labels:
+        app: "<workflow_name>"
+    spec:
+      serviceAccountName: <gke_service_account>
+      containers:
+      - name: "psl-to-gmk-1"
+        image: "gcr.io/<gcp_project>/psl-to-gmk:latest"
+---
+apiVersion: "autoscaling/v2"
+kind: "HorizontalPodAutoscaler"
+metadata:
+  name: "<workflow_name>-hpa-iwbr"
+  namespace: "default"
+  labels:
+    app: "<workflow_name>"
+spec:
+  scaleTargetRef:
+    kind: "Deployment"
+    name: "<workflow_name>"
+    apiVersion: "apps/v1"
+  minReplicas: 1
+  maxReplicas: 5
+  metrics:
+  - type: "Resource"
+    resource:
+      name: "cpu"
+      target:
+        type: "Utilization"
+        averageUtilization: 80