[kafka] add system tests to consumer and producer data streams (#15334)

* [kafka] add system tests to consumer and producer data streams * update PR link * producer docker healthcheck * consumer docker healthcheck * update producer sample_event.json * moved docker tests setup from individual data streams to top level kafka _dev dir * revert raft data stream changes * fix file name

Author: stefans-elastic

packages/kafka/_dev/deploy/docker/Dockerfile

@@ -0,0 +1,15 @@
+FROM apache/kafka:4.0.0
+
+USER root
+
+# Copy jolokia agent jar from host into container
+COPY jolokia/jolokia-jvm-agent.jar /opt/jolokia/jolokia-jvm.jar
+
+# Copy startup script
+COPY startup.sh /opt/startup.sh
+COPY healthcheck.sh /opt/healthcheck.sh
+RUN chmod +x /opt/startup.sh /opt/healthcheck.sh && chown -R 1001:0 /opt
+
+USER 1001
+
+ENTRYPOINT ["/opt/startup.sh"]

packages/kafka/data_stream/controller/_dev/deploy/docker/docker-compose.yml renamed to packages/kafka/_dev/deploy/docker/docker-compose.yml

@@ -4,12 +4,20 @@ services:
  default:
  aliases:
  - svc-kafka
- image: docker.elastic.co/observability-ci/kafka-with-jolokia:4.0.0
+ build: .
  ports:
  - "9092:9092"
  - "9093:9093"
  - "9999:9999"
  - "8780:8780"
+ - "8774:8774"
+ - "8775:8775"
+ healthcheck:
+ test: [ "CMD-SHELL", "/opt/healthcheck.sh" ]
+ interval: 15s
+ timeout: 10s
+ retries: 5
+ start_period: 60s
  environment:
  KAFKA_NODE_ID: 1
  KAFKA_PROCESS_ROLES: broker,controller
@@ -29,4 +37,4 @@ services:
  -Dcom.sun.management.jmxremote.port=9999
  -Dcom.sun.management.jmxremote.rmi.port=9999
  volumes:
- - ./server.properties:/opt/kafka/config/kraft/server.properties
+ - ./server.properties:/opt/kafka/config/kraft/server.properties

packages/kafka/_dev/deploy/docker/healthcheck.sh

@@ -0,0 +1,15 @@
+#!/bin/bash
+# Exit immediately if a command exits with a non-zero status.
+set -e
+
+# 1. Check for the Kafka Broker process
+pgrep -f 'kafka.Kafka' > /dev/null || { echo "Healthcheck failed: Kafka broker not running."; exit 1; }
+
+# 2. Check for the Console Producer process
+pgrep -f 'kafka.tools.ConsoleProducer' > /dev/null || { echo "Healthcheck failed: Console producer not running."; exit 1; }
+
+# 3. Check for the Console Consumer process
+pgrep -f 'kafka.tools.consumer.ConsoleConsumer' > /dev/null || { echo "Healthcheck failed: Console consumer not running."; exit 1; }
+
+# If all checks pass, exit with 0 to indicate the container is healthy.
+exit 0

packages/kafka/_dev/deploy/docker/jolokia/jolokia-jvm-agent.jar

@@ -0,0 +1,76 @@
+#!/bin/bash
+set -e
+
+# --- 1. CONFIGURE AND START KAFKA BROKER ---
+
+# Add Jolokia agent to the broker's JMX options
+export KAFKA_JMX_OPTS="
+ -Dcom.sun.management.jmxremote
+ -Dcom.sun.management.jmxremote.local.only=false
+ -Dcom.sun.management.jmxremote.authenticate=false
+ -Dcom.sun.management.jmxremote.ssl=false
+ -Djava.rmi.server.hostname=0.0.0.0
+ -Dcom.sun.management.jmxremote.port=${KAFKA_JMX_PORT:-9999}
+ -Dcom.sun.management.jmxremote.rmi.port=${KAFKA_JMX_PORT:-9999}
+ -javaagent:/opt/jolokia/jolokia-jvm.jar=port=8780,host=0.0.0.0
+"
+
+LOG_DIR="/tmp/kraft-combined-logs"
+META_FILE="$LOG_DIR/meta.properties"
+CLUSTER_ID=${KAFKA_CLUSTER_ID:-"_ABcDEf1gHiJkLmNoPqRsT"} # Using a default Base64 ID
+
+# Ensure log directory exists and has correct permissions
+mkdir -p "$LOG_DIR"
+chown -R 1001:0 "$LOG_DIR"
+
+# Initialize storage if not formatted yet
+if [ ! -f "$META_FILE" ]; then
+ echo "Formatting KRaft storage with Cluster ID: $CLUSTER_ID"
+ /opt/kafka/bin/kafka-storage.sh format -t "$CLUSTER_ID" -c /opt/kafka/config/kraft/server.properties
+else
+ echo "KRaft storage already formatted."
+fi
+
+# Start the Kafka server in the background
+echo "Starting Kafka server..."
+/opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/kraft/server.properties &
+# Store the Process ID (PID) of the Kafka server
+kafka_pid=$!
+
+# --- 2. WAIT FOR BROKER AND CREATE TOPIC ---
+
+# Wait for the broker to become available on port 9092
+echo "Waiting for Kafka broker to be ready..."
+until KAFKA_JMX_OPTS= /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list > /dev/null 2>&1; do
+ echo "Broker not ready yet, sleeping..."
+ sleep 2
+done
+echo "Broker is ready!"
+
+# Create a topic for the producer and consumer
+echo "Creating topic 'my-topic'..."
+KAFKA_JMX_OPTS= /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --if-not-exists --topic my-topic --partitions 1 --replication-factor 1
+
+# --- 3. START PRODUCER AND CONSUMER ---
+
+# Start the console producer in the background
+echo "Starting producer with Jolokia on port 8775..."
+export KAFKA_OPTS="-javaagent:/opt/jolokia/jolokia-jvm.jar=port=8775,host=0.0.0.0"
+(while true; do echo "Test msg"; sleep 5; done) | \
+KAFKA_JMX_OPTS= /opt/kafka/bin/kafka-console-producer.sh \
+ --bootstrap-server localhost:9092 \
+ --topic my-topic &
+
+# Start the console consumer in the background
+echo "Starting consumer with Jolokia on port 8774..."
+export KAFKA_OPTS="-javaagent:/opt/jolokia/jolokia-jvm.jar=port=8774,host=0.0.0.0"
+KAFKA_JMX_OPTS= /opt/kafka/bin/kafka-console-consumer.sh \
+ --bootstrap-server localhost:9092 \
+ --topic my-topic \
+ --from-beginning &
+
+# --- 4. WAIT FOR KAFKA SERVER TO EXIT ---
+echo "Startup complete. All processes are running."
+# The 'wait' command blocks the script until the Kafka server process (kafka_pid) exits.
+# This keeps the container running.
+wait $kafka_pid

packages/kafka/data_stream/controller/_dev/deploy/docker/server.properties renamed to packages/kafka/_dev/deploy/docker/server.properties

@@ -1,4 +1,9 @@
 # newer versions go on top
+- version: "1.23.1"
+ changes:
+ - description: Add system tests for consumer and producer data streams.
+ type: enhancement
+ link: https://github.com/elastic/integrations/pull/15334
 - version: "1.23.0"
  changes:
  - description: Add support for producer and consumer data streams.

packages/kafka/_dev/deploy/docker/startup.sh

@@ -0,0 +1,10 @@
+vars:
+ hosts:
+ - http://{{Hostname}}:8780
+ metrics_path:
+ - /jolokia
+data_stream:
+ vars:
+ period: 10s
+ jolokia_hosts:
+ - http://{{Hostname}}:8774