rpm: add confluent (kafka) test case

.github/workflows/yum.yml

@@ -77,6 +77,16 @@ jobs:
           --env CENTOS_STREAM=${{ matrix.centos-stream }} \
           ${{ matrix.test-docker-image }} \
           /fluentd/fluent-package/yum/serverspec-test.sh
+      - name: Confluent Test
+        run: |
+          mkdir -p .bundle
+          docker run \
+          --rm \
+          --tty \
+          --env CI=true \
+          --volume ${PWD}:/fluentd:ro \
+          ${{ matrix.test-docker-image }} \
+          /fluentd/fluent-package/yum/confluent-test.sh
       - name: Binstubs Test
         run: |
           mkdir -p .bundle

fluent-package/apt/confluent-test.sh

@@ -34,41 +34,15 @@ chmod 644 /usr/share/keyrings/confluent-archive-keyring.gpg
 echo "deb [arch=${architecture} signed-by=/usr/share/keyrings/confluent-archive-keyring.gpg] https://packages.confluent.io/deb/7.6 stable main" > /etc/apt/sources.list.d/confluent.list
 apt update && apt install -y confluent-community-2.13 ${java_jdk} netcat-openbsd
 
-export KAFKA_OPTS=-Dzookeeper.4lw.commands.whitelist=ruok
-/usr/bin/zookeeper-server-start /etc/kafka/zookeeper.properties  &
-N_POLLING=30
-n=1
-while true ; do
-    sleep 1
-    status=$(echo ruok | nc localhost 2181)
-    if [ "$status" = "imok" ]; then
-	break
-    fi
-    n=$((n + 1))
-    if [ $n -ge $N_POLLING ]; then
-	echo "failed to get response from zookeeper-server"
-	exit 1
-    fi
-done
-/usr/bin/kafka-server-start /etc/kafka/server.properties &
-n=1
-while true ; do
-    sleep 1
-    status=$(/usr/bin/zookeeper-shell localhost:2181 ls /brokers/ids | sed -n 6p)
-    if [ "$status" = "[0]" ]; then
-	break
-    fi
-    n=$((n + 1))
-    if [ $n -ge $N_POLLING ]; then
-	echo "failed to get response from kafka-server"
-	exit 1
-    fi
-done
+CONFLUENT_SCRIPT=$(dirname $(realpath $0))/../run-confluent.sh
+echo ${CONFLUENT_SCRIPT}
+bash ${CONFLUENT_SCRIPT}
+
 if [ "$CI" = "true" ]; then
    echo "::endgroup::"
 fi
-/usr/bin/kafka-topics --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic test
+
 export PATH=/opt/fluent/bin:$PATH
 export INSTALLATION_TEST=true
-cd /fluentd && rake serverspec:kafka
 /usr/sbin/fluentd -c /fluentd/serverspec/test.conf &
+cd /fluentd && rake serverspec:kafka

fluent-package/run-confluent.sh

@@ -0,0 +1,33 @@
+#!/usr/bin/bash
+
+export KAFKA_OPTS=-Dzookeeper.4lw.commands.whitelist=ruok
+/usr/bin/zookeeper-server-start /etc/kafka/zookeeper.properties  &
+N_POLLING=30
+n=1
+while true ; do
+    sleep 1
+    status=$(echo ruok | nc localhost 2181)
+    if [ "$status" = "imok" ]; then
+	break
+    fi
+    n=$((n + 1))
+    if [ $n -ge $N_POLLING ]; then
+	echo "failed to get response from zookeeper-server"
+	exit 1
+    fi
+done
+/usr/bin/kafka-server-start /etc/kafka/server.properties &
+n=1
+while true ; do
+    sleep 1
+    status=$(/usr/bin/zookeeper-shell localhost:2181 ls /brokers/ids | sed -n 6p)
+    if [ "$status" = "[0]" ]; then
+	break
+    fi
+    n=$((n + 1))
+    if [ $n -ge $N_POLLING ]; then
+	echo "failed to get response from kafka-server"
+	exit 1
+    fi
+done
+/usr/bin/kafka-topics --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic test

fluent-package/yum/binstubs-test.sh

@@ -14,8 +14,8 @@ if [ "$CI" = "true" ]; then
    echo "::group::Setup binstubs test"
 fi
 
-distribution=$(cat /etc/system-release-cpe | awk '{print substr($0, index($1, "o"))}' | cut -d: -f2)
-version=$(cat /etc/system-release-cpe | awk '{print substr($0, index($1, "o"))}' | cut -d: -f4)
+distribution=$(cat /etc/system-release-cpe | awk '{print substr($0, index($0, "o"))}' | cut -d: -f2)
+version=$(cat /etc/system-release-cpe | awk '{print substr($0, index($0, "o"))}' | cut -d: -f4)
 
 ENABLE_UPGRADE_TEST=1
 case ${distribution} in

fluent-package/yum/confluent-test.sh

@@ -0,0 +1,72 @@
+#!/bin/bash
+
+set -exu
+
+if [ "$CI" = "true" ]; then
+   echo "::group::Setup confluent test"
+fi
+
+# Amazon Linux 2 system-release-cpe is:
+# cpe:2.3:o:amazon:amazon_linux:2
+# CentOS 7 system-release-cpe is:
+# cpe:/o:centos:centos:7
+# This means that column glitch exists.
+# So, we should remove before "o" character.
+
+distribution=$(cat /etc/system-release-cpe | awk '{print substr($0, index($0, "o"))}' | cut -d: -f2)
+version=$(cat /etc/system-release-cpe | awk '{print substr($0, index($0, "o"))}' | cut -d: -f4)
+
+case ${distribution} in
+    amazon)
+	case ${version} in
+	    2)
+		DNF=yum
+		DISTRIBUTION_VERSION=${version}
+		${DNF} install -y java-17-amazon-corretto-headless
+		;;
+	    2023)
+		DNF=dnf
+		DISTRIBUTION_VERSION=${version}
+		${DNF} install -y java-21-amazon-corretto-headless
+	    ;;
+	esac
+	;;
+    rocky|almalinux)
+	DNF=dnf
+	DISTRIBUTION_VERSION=$(echo ${version} | cut -d. -f1)
+	${DNF} install -y java-21-openjdk-headless
+	;;
+esac
+
+repositories_dir=/fluentd/fluent-package/yum/repositories
+ARCH=$(rpm --eval "%{_arch}")
+${DNF} install -y \
+  ${repositories_dir}/${distribution}/${DISTRIBUTION_VERSION}/${ARCH}/Packages/*.rpm
+
+fluentd --version
+
+/usr/sbin/fluent-gem install --no-document serverspec
+rpm --import https://packages.confluent.io/rpm/7.6/archive.key
+
+cat <<EOF > /etc/yum.repos.d/confluent.repo
+[Confluent]
+name=Confluent repository
+baseurl=https://packages.confluent.io/rpm/7.6
+gpgcheck=1
+gpgkey=https://packages.confluent.io/rpm/7.6/archive.key
+enabled=1
+EOF
+${DNF} update -y && ${DNF} install -y confluent-community-2.13 nmap-ncat
+
+CONFLUENT_SCRIPT=$(dirname $(realpath $0))/../run-confluent.sh
+echo ${CONFLUENT_SCRIPT}
+bash ${CONFLUENT_SCRIPT}
+
+if [ "$CI" = "true" ]; then
+   echo "::endgroup::"
+fi
+
+export PATH=/opt/fluent/bin:$PATH
+export INSTALLATION_TEST=true
+/usr/sbin/fluentd -c /fluentd/serverspec/test.conf &
+cd /fluentd && rake -T && rake serverspec:kafka

fluent-package/yum/install-test.sh

@@ -11,8 +11,8 @@ set -exu
 # This means that column glitch exists.
 # So, we should remove before "o" character.
 
-distribution=$(cat /etc/system-release-cpe | awk '{print substr($0, index($1, "o"))}' | cut -d: -f2)
-version=$(cat /etc/system-release-cpe | awk '{print substr($0, index($1, "o"))}' | cut -d: -f4)
+distribution=$(cat /etc/system-release-cpe | awk '{print substr($0, index($0, "o"))}' | cut -d: -f2)
+version=$(cat /etc/system-release-cpe | awk '{print substr($0, index($0, "o"))}' | cut -d: -f4)
 
 ENABLE_UPGRADE_TEST=1
 case ${distribution} in

fluent-package/yum/serverspec-test.sh

@@ -15,23 +15,17 @@ if [ "$CI" = "true" ]; then
    echo "::group::Setup serverspec test"
 fi
 
-distribution=$(cat /etc/system-release-cpe | awk '{print substr($0, index($1, "o"))}' | cut -d: -f2)
-version=$(cat /etc/system-release-cpe | awk '{print substr($0, index($1, "o"))}' | cut -d: -f4)
+distribution=$(cat /etc/system-release-cpe | awk '{print substr($0, index($0, "o"))}' | cut -d: -f2)
+version=$(cat /etc/system-release-cpe | awk '{print substr($0, index($0, "o"))}' | cut -d: -f4)
 
-ENABLE_SERVERSPEC_TEST=1
-ENABLE_KAFKA_TEST=1
-JAVA_JRE=java-11-openjdk
-N_POLLING=30
 case ${distribution} in
   amazon)
     case ${version} in
       2)
         DNF=yum
-        ENABLE_SERVERSPEC_TEST=0
         DISTRIBUTION_VERSION=${version}
         ;;
       2023)
-        ENABLE_SERVERSPEC_TEST=0
         DNF=dnf
         DISTRIBUTION_VERSION=${version}
         ;;
@@ -61,8 +55,6 @@ case ${distribution} in
     DNF=dnf
     DISTRIBUTION_VERSION=$(echo ${version} | cut -d. -f1)
     version=$DISTRIBUTION_VERSION
-    # Use newer JDK to avoid FileNotFoundException about tzdb.dat
-    JAVA_JRE=java-17-openjdk
     case ${version} in
       9)
         # FIXME: Accept SHA-1 signed confluent packages.
@@ -79,67 +71,7 @@ ${DNF} install -y \
 
 fluentd --version
 
-if [ $ENABLE_SERVERSPEC_TEST -eq 1 ]; then
-    curl -V > /dev/null 2>&1 || ${DNF} install -y curl
-    ${DNF} install -y which ${repositories_dir}/${distribution}/${DISTRIBUTION_VERSION}/x86_64/Packages/*.rpm
-
-    /usr/sbin/fluent-gem install --no-document serverspec
-    if [ $ENABLE_KAFKA_TEST -eq 1 ]; then
-        rpm --import https://packages.confluent.io/rpm/7.4/archive.key
-        cat >/etc/yum.repos.d/confluent.repo <<EOF;
-[Confluent]
-name=Confluent repository
-baseurl=https://packages.confluent.io/rpm/7.4
-gpgcheck=1
-gpgkey=https://packages.confluent.io/rpm/7.4/archive.key
-enabled=1
-
-[Confluent-Clients]
-name=Confluent Clients repository
-baseurl=https://packages.confluent.io/clients/rpm/centos/\$releasever/\$basearch
-gpgcheck=1
-gpgkey=https://packages.confluent.io/clients/rpm/archive.key
-enabled=1
-EOF
-	yum update -y && yum install -y confluent-community-2.13 ${JAVA_JRE} nc
-	export KAFKA_OPTS=-Dzookeeper.4lw.commands.whitelist=ruok
-	/usr/bin/zookeeper-server-start /etc/kafka/zookeeper.properties  &
-	n=1
-        while true ; do
-	    sleep 1
-	    status=$(echo ruok | nc localhost 2181)
-	    if [ "$status" = "imok" ]; then
-		break
-	    fi
-            n=$((n + 1))
-	    if [ $n -ge $N_POLLING ]; then
-		echo "failed to get response from zookeeper-server"
-		exit 1
-	    fi
-	done
-	# Allow connection to kafka server
-        echo "listeners=PLAINTEXT://localhost:9092" | tee -a /etc/kafka/server.properties
-	/usr/bin/kafka-server-start /etc/kafka/server.properties &
-	n=1
-	while true ; do
-	    sleep 1
-	    status=$(/usr/bin/kafka-topics --bootstrap-server localhost:9092 --list)
-	    if [ "$status" = "" ]; then
-		break
-	    fi
-            n=$((n + 1))
-	    if [ $n -ge $N_POLLING ]; then
-		echo "failed to get list of topics from kafka-server"
-		exit 1
-	    fi
-	done
-	/usr/bin/kafka-topics --bootstrap-server localhost:9092 --topic test --create --replication-factor 1 --partitions 1
-	/usr/sbin/fluentd -c /fluentd/serverspec/test.conf &
-    fi
-    if [ "$CI" = "true" ]; then
-	echo "::endgroup::"
-    fi
-    export PATH=/opt/fluent/bin:$PATH
-    export INSTALLATION_TEST=true
-    cd /fluentd && rake serverspec:linux
-fi
+/usr/sbin/fluent-gem install --no-document serverspec
+export PATH=/opt/fluent/bin:$PATH
+export INSTALLATION_TEST=true
+cd /fluentd && rake serverspec:linux

fluent-package/yum/systemd-test/commonvar.sh

@@ -1,7 +1,7 @@
 #!/bin/bash
 
-distribution=$(cat /etc/system-release-cpe | awk '{print substr($0, index($1, "o"))}' | cut -d: -f2)
-version=$(cat /etc/system-release-cpe | awk '{print substr($0, index($1, "o"))}' | cut -d: -f4)
+distribution=$(cat /etc/system-release-cpe | awk '{print substr($1, index($1, "o"))}' | cut -d: -f2)
+version=$(cat /etc/system-release-cpe | awk '{print substr($1, index($1, "o"))}' | cut -d: -f4)
 td_agent_version=4.5.2
 
 case $distribution in

serverspec/kafka/confluent.rb

@@ -1,30 +1,32 @@
 require_relative "../spec_helper"
 require "rdkafka"
 
-describe "rdkafka", :if => !centos8?(os) do
-  it "can receive message via Rdkafka client" do
-    config = {
-      "bootstrap.servers": "localhost:9092",
-      "group.id": "test"
-    }
-    consumer = Rdkafka::Config.new(config).consumer
-    consumer.subscribe("test")
+if ["redhat", "amazon", "debian"].include?(os[:family])
+  describe "rdkafka" do
+    it "can receive message via Rdkafka client" do
+      config = {
+        "bootstrap.servers": "localhost:9092",
+        "group.id": "test"
+      }
+      consumer = Rdkafka::Config.new(config).consumer
+      consumer.subscribe("test")
 
-    wait_for_consumer_assignment(consumer)
+      wait_for_consumer_assignment(consumer)
 
-    `echo "Hello, rdkafka" | /usr/bin/kafka-console-producer --broker-list localhost:9092 --topic test`
+      `echo "Hello, rdkafka" | /usr/bin/kafka-console-producer --broker-list localhost:9092 --topic test`
 
-    message = consumer.each { |message| break message }
-    expect(message.payload).to eq "Hello, rdkafka"
+      message = consumer.each { |message| break message }
+      expect(message.payload).to eq "Hello, rdkafka"
+    end
   end
-end
 
-describe "fluent-plugin-kafka", :if => !centos8?(os) do
-  it "can receive message via fluent-plugin-kafka" do
-    `echo "Hello, fluent-plugin-kafka" | /usr/bin/kafka-console-producer --broker-list localhost:9092 --topic test`
-    Dir.glob("/tmp/log/td-agent/*.log") do |path|
-      File.open(path) do |file|
-        expect(JSON.parse(file.readlines.last)["message"]).to eq "Hello, fluent-plugin-kafka"
+  describe "fluent-plugin-kafka" do
+    it "can receive message via fluent-plugin-kafka" do
+      `echo "Hello, fluent-plugin-kafka" | /usr/bin/kafka-console-producer --broker-list localhost:9092 --topic test`
+      Dir.glob("/tmp/log/td-agent/*.log") do |path|
+        File.open(path) do |file|
+          expect(JSON.parse(file.readlines.last)["message"]).to eq "Hello, fluent-plugin-kafka"
+        end
       end
     end
   end

serverspec/linux/td-agent.rb

@@ -6,7 +6,7 @@
   it { should be_installed }
 end
 
-if os[:family] == 'redhat'
+if ["redhat", "amazon"].include?(os[:family])
   describe user("fluentd") do
     it { should exist }
     it { should belong_to_group "fluentd" }

serverspec/spec_helper.rb

@@ -8,7 +8,3 @@ def wait_for_consumer_assignment(consumer)
     sleep 1
   end
 end
-
-def centos8?(os)
-  os[:family] == "redhat" and os[:release].split(".", 2)[0].to_i >= 8
-end