ci: fix kafka e2e test (numaproj#16)

* ci: fix kafka e2e test Signed-off-by: Derek Wang <[email protected]>
ashwinidulams · May 27, 2022 · be6b441 · be6b441
1 parent 5c9433f
commit be6b441
Show file tree

Hide file tree

Showing 7 changed files with 41 additions and 10 deletions.
diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml
@@ -121,7 +121,7 @@ jobs:
       max-parallel: 4
       matrix:
         driver: [jetstream]
-        case: [e2e]
+        case: [e2e, kafka-e2e]
     steps:
       - name: Checkout code
         uses: actions/checkout@v3

diff --git a/Makefile b/Makefile
@@ -97,11 +97,22 @@ test-e2e:
 test-kafka-e2e:
 test-%: 
 	$(MAKE) image e2eapi-image
-	kubectl -n numaflow-system delete po -lapp=controller-manager,app.kubernetes.io/part-of=numaflow
+	kubectl -n numaflow-system delete po -lapp.kubernetes.io/component=controller-manager,app.kubernetes.io/part-of=numaflow
 	kubectl -n numaflow-system delete po e2e-api-pod  --ignore-not-found=true
 	cat test/manifests/e2e-api-pod.yaml |  sed '[email protected]/numaproj/@$(IMAGE_NAMESPACE)/@' | sed 's/:$(BASE_VERSION)/:$(VERSION)/' | kubectl -n numaflow-system apply -f -
 	go generate $(shell find ./test/$* -name '*.go')
-	go test -v -timeout 10m -count 1 --tags test -p 1 ./test/$*
+	-go test -v -timeout 10m -count 1 --tags test -p 1 ./test/$*
+	$(MAKE) cleanup-e2e
+
+
+.PHONY: cleanup-e2e
+cleanup-e2e:
+	kubectl -n numaflow-system delete svc -lnumaflow-e2e=true --ignore-not-found=true
+	kubectl -n numaflow-system delete sts -lnumaflow-e2e=true --ignore-not-found=true
+	kubectl -n numaflow-system delete deploy -lnumaflow-e2e=true --ignore-not-found=true
+	kubectl -n numaflow-system delete cm -lnumaflow-e2e=true --ignore-not-found=true
+	kubectl -n numaflow-system delete secret -lnumaflow-e2e=true --ignore-not-found=true
+	kubectl -n numaflow-system delete po -lnumaflow-e2e=true --ignore-not-found=true
 
 .PHONY: image
 image: clean dist/$(BINARY_NAME)-linux-amd64

diff --git a/config/apps/kafka/kafka-minimal.yaml b/config/apps/kafka/kafka-minimal.yaml
@@ -67,7 +67,7 @@ spec:
     spec:
       containers:
         - name: main
-          image: wurstmeister/kafka
+          image: wurstmeister/kafka:2.13-2.8.1
           imagePullPolicy: IfNotPresent
           ports:
             - containerPort: 9092
@@ -81,7 +81,7 @@ spec:
             - name: KAFKA_BROKER_ID
               value: "0"
             - name: KAFKA_CREATE_TOPICS
-              value: "input-topic:1:2,middle-topic:1:2,output-topic:1:2"
+              value: "input-topic:1:1,middle-topic:1:1,output-topic:1:1"
           readinessProbe:
             tcpSocket:
               port: 9092

diff --git a/config/apps/kafka/kustomization.yaml b/config/apps/kafka/kustomization.yaml
@@ -1,2 +1,6 @@
 resources:
   - kafka-minimal.yaml
+
+commonLabels:
+  "numaflow-e2e": "true"
+
diff --git a/test/e2e-api/kafka.go b/test/e2e-api/kafka.go
@@ -2,12 +2,13 @@ package main
 
 import (
 	"fmt"
-	"github.com/Shopify/sarama"
 	"io/ioutil"
 	"log"
 	"net/http"
 	"strconv"
 	"time"
+
+	"github.com/Shopify/sarama"
 )
 
 func init() {
@@ -17,12 +18,14 @@ func init() {
 		topic := r.URL.Query().Get("topic")
 		admin, err := sarama.NewClusterAdmin(brokers, sarama.NewConfig())
 		if err != nil {
+			log.Println(err)
 			w.WriteHeader(500)
 			_, _ = w.Write([]byte(err.Error()))
 			return
 		}
 		defer admin.Close()
 		if err = admin.CreateTopic(topic, &sarama.TopicDetail{NumPartitions: 1, ReplicationFactor: 1}, true); err != nil {
+			log.Println(err)
 			w.WriteHeader(500)
 			_, _ = w.Write([]byte(err.Error()))
 			return
@@ -34,6 +37,7 @@ func init() {
 		topic := r.URL.Query().Get("topic")
 		admin, err := sarama.NewClusterAdmin(brokers, sarama.NewConfig())
 		if err != nil {
+			log.Println(err)
 			w.WriteHeader(500)
 			_, _ = w.Write([]byte(err.Error()))
 			return
@@ -49,13 +53,15 @@ func init() {
 	http.HandleFunc("/kafka/list-topics", func(w http.ResponseWriter, r *http.Request) {
 		consumer, err := sarama.NewConsumer(brokers, sarama.NewConfig())
 		if err != nil {
+			log.Println(err)
 			w.WriteHeader(500)
 			_, _ = w.Write([]byte(err.Error()))
 			return
 		}
 		defer consumer.Close()
 		topics, err := consumer.Topics()
 		if err != nil {
+			log.Println(err)
 			w.WriteHeader(500)
 			_, _ = w.Write([]byte(err.Error()))
 			return
@@ -68,6 +74,7 @@ func init() {
 		topic := r.URL.Query().Get("topic")
 		count, err := strconv.Atoi(r.URL.Query().Get("count"))
 		if err != nil {
+			log.Println(err)
 			w.WriteHeader(500)
 			_, _ = w.Write([]byte(err.Error()))
 			return
@@ -81,6 +88,7 @@ func init() {
 		defer consumer.Close()
 		pConsumer, err := consumer.ConsumePartition(topic, 0, sarama.OffsetOldest)
 		if err != nil {
+			log.Println(err)
 			w.WriteHeader(500)
 			_, _ = w.Write([]byte(err.Error()))
 			return
@@ -104,6 +112,7 @@ func init() {
 		key := r.URL.Query().Get("key")
 		buf, err := ioutil.ReadAll(r.Body)
 		if err != nil {
+			log.Println(err)
 			w.WriteHeader(500)
 			_, _ = w.Write([]byte(err.Error()))
 			return
@@ -135,6 +144,7 @@ func init() {
 		mf := newMessageFactory(r.URL.Query())
 		duration, err := time.ParseDuration(r.URL.Query().Get("sleep"))
 		if err != nil {
+			log.Println(err)
 			w.WriteHeader(400)
 			_, _ = w.Write([]byte(err.Error()))
 			return

diff --git a/test/kafka-e2e/kafka_test.go b/test/kafka-e2e/kafka_test.go
@@ -12,8 +12,10 @@ import (
 	"time"
 )
 
-//go:generate kubectl -n numaflow-system delete statefulset zookeeper kafka-broker  --ignore-not-found=true
-//go:generate kubectl apply  -k ../../config/apps/kafka -n numaflow-system
+//go:generate kubectl -n numaflow-system delete statefulset zookeeper kafka-broker --ignore-not-found=true
+//go:generate kubectl apply -k ../../config/apps/kafka -n numaflow-system
+// Wait for zookeeper to come up
+//go:generate sleep 60
 
 type KafkaSuite struct {
 	fixtures.E2ESuite
@@ -73,6 +75,7 @@ func (ks *KafkaSuite) TestKafkaSink() {
 	fixtures.ExpectKafkaTopicCount(outputTopic, 15, 3*time.Second)
 
 }
+
 func (ks *KafkaSuite) TestKafkaSourceSink() {
 	inputTopic := fixtures.CreateKafkaTopic()
 	outputTopic := fixtures.CreateKafkaTopic()
@@ -86,8 +89,9 @@ func (ks *KafkaSuite) TestKafkaSourceSink() {
 					Name: "input",
 					Source: &dfv1.Source{
 						Kafka: &dfv1.KafkaSource{
-							Brokers: []string{"kafka-broker:9092"},
-							Topic:   inputTopic,
+							Brokers:           []string{"kafka-broker:9092"},
+							Topic:             inputTopic,
+							ConsumerGroupName: "test-group",
 						},
 					},
 				},

diff --git a/test/manifests/e2e-api-pod.yaml b/test/manifests/e2e-api-pod.yaml
@@ -2,6 +2,8 @@ apiVersion: v1
 kind: Pod
 metadata:
   name: e2e-api-pod
+  labels:
+    numaflow-e2e: "true"
 spec:
   containers:
     - name: main