Add more definitive wait for kafka clusters
@@ -5,6 +5,7 @@ package e2e
import (
"context"
"fmt"
+ "strings"
"testing"
framework "github.com/operator-framework/operator-sdk/pkg/test"
@@ -14,8 +15,11 @@ import (
corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+ "k8s.io/apimachinery/pkg/types"
+ "k8s.io/apimachinery/pkg/util/wait"
v1 "github.com/jaegertracing/jaeger-operator/pkg/apis/jaegertracing/v1"
+ kafkav1beta1 "github.com/jaegertracing/jaeger-operator/pkg/apis/kafka/v1beta1"
)
type StreamingTestSuite struct {
@@ -275,8 +279,21 @@ func getTLSVolumes(kafkaUserName string) []corev1.Volume {
}
func waitForKafkaInstance() {
- err := WaitForStatefulset(t, fw.KubeClient, kafkaNamespace, "my-cluster-kafka", retryInterval, timeout)
- require.NoError(t, err, "Error waiting for my-cluster-kafka")
+ kafkaInstance := &kafkav1beta1.Kafka{}
+
+ err := wait.Poll(retryInterval, timeout, func() (done bool, err error) {
+ err = fw.Client.Get(context.Background(), types.NamespacedName{Name: "my-cluster", Namespace: kafkaNamespace}, kafkaInstance)
+ require.NoError(t, err)
+
+ for _, condition := range kafkaInstance.Status.Conditions {
+ if strings.EqualFold(condition.Type, "ready") && strings.EqualFold(condition.Status, "true") {
+ return true, nil
+ }
+ }
+
+ return false, nil
+ })
+ require.NoError(t, err)
}
func waitForElasticSearch() {
