update kafka instructions for dealing with messages etc

derrickaw · derrickaw · commit b14971f73225 · 2025-09-15T14:21:12.000Z
diff --git a/dataflow/flex-templates/kafka_to_bigquery/README.md b/dataflow/flex-templates/kafka_to_bigquery/README.md
@@ -130,9 +130,9 @@ For this, we need two parts running:
 > </details>
 
 The Kafka server must be accessible to *external* applications.
-For this we need a
-[static IP address](https://cloud.google.com/compute/docs/ip-addresses/reserve-static-external-ip-address)
-for the Kafka server to live.
+For this we need an
+[external static IP address](https://cloud.google.com/compute/docs/ip-addresses/reserve-static-external-ip-address)
+for the Kafka server to live. Not an internal IP address.
 
 > ℹ️ If you already have a Kafka server running you can skip this section.
 > Just make sure to store its IP address into an environment variable.
@@ -186,6 +186,14 @@ export KAFKA_IMAGE="gcr.io/$PROJECT/samples/dataflow/kafka:latest"
 # Build the Kafka server image into Container Registry.
 gcloud builds submit --tag $KAFKA_IMAGE kafka/
 
+# If a different topic, address, kafka port, or zookeeper port is desired,
+# update the following environment variables before starting the server.
+# Otherwise, the default values will be used in the Dockerfile:
+export KAFKA_TOPIC=<topic-name>
+export KAFKA_ADDRESS=<kafka-address>
+export KAFKA_PORT=<kafka-port>
+export ZOOKEEPER_PORT=<zookeeper-port>
+
 # Create and start a new instance.
 # The --address flag binds the VM's address to the static address we created.
 # The --container-env KAFKA_ADDRESS is an environment variable passed to the
@@ -200,6 +208,41 @@ gcloud compute instances create-with-container kafka-vm \
   --tags "kafka-server"
 ```
 
+### Sending messages to Kafka server
+
+The Kafka server should be running at this point, but in its current state no
+messages are being sent to a topic which will cause the KafkaToBigQuery
+template to fail. So ssh into the `kafka-vm` that was created earlier and issue
+the below commands that are required based on your timing. Messages sent before
+the template is started will be present when the template is started. If the 
+desire is to send messages after the template has started, then the messages
+will be processed as they are sent.
+
+```sh
+# 1. If the existing topic is not sufficient, please create a new one:
+docker run --rm --network host bitnami/kafka:3.4.0 \
+/opt/bitnami/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 \
+--create --topic <topic-name> --partitions 1 --replication-factor 1
+
+# 2. If the existing topic needs deleting, please delete it:
+docker run --rm --network host bitnami/kafka:3.4.0 \
+/opt/bitnami/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 \
+--delete --topic <topic-name>
+
+# 3. If messages need to be sent, send them to the Kafka topic via the following
+# command and then hit enter after each message. End via ctrl+c:
+docker run -i --rm --network host bitnami/kafka:3.4.0 \
+/opt/bitnami/kafka/bin/kafka-console-producer.sh \
+--bootstrap-server localhost:9092 --topic messages
+
+# 4. If the messages need to be verified that they exist, issue this command
+# and end via ctrl+c:
+docker run -it --rm --network host bitnami/kafka:3.4.0 \
+/opt/bitnami/kafka/bin/kafka-console-consumer.sh \
+--bootstrap-server localhost:9092 --topic messages --from-beginning
+```
+
+
 ### Creating and running a Flex Template
 
 > <details><summary>
diff --git a/dataflow/flex-templates/kafka_to_bigquery/kafka/Dockerfile b/dataflow/flex-templates/kafka_to_bigquery/kafka/Dockerfile
@@ -33,7 +33,7 @@ ENV ZOOKEEPER_PORT="${ZOOKEEPER_PORT:-2181}"
 
 # Download and install Apache Kafka.
 RUN apk add --no-cache bash \
-  && wget http://apache.mirrors.spacedump.net/kafka/${KAFKA_VERSION}/kafka_${SCALA_VERSION}-${KAFKA_VERSION}.tgz \
+  && wget http://archive.apache.org/dist/kafka/${KAFKA_VERSION}/kafka_${SCALA_VERSION}-${KAFKA_VERSION}.tgz \
     -O /tmp/kafka.tgz \
   && tar xzf /tmp/kafka.tgz -C /opt && rm /tmp/kafka.tgz \
   && ln -s /opt/kafka_${SCALA_VERSION}-${KAFKA_VERSION} /opt/kafka
