This is an automated email from the ASF dual-hosted git repository. squakez pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/camel-k.git
commit a25f1257f1b04e6b6a0ba14ba860e233bf7a8229 Author: Anshul Jagota <[email protected]> AuthorDate: Sat Oct 31 09:02:16 2026 +0530 fix(e2e/keda): use YAML flows in integration test and pre-create Kafka topic --- e2e/keda/files/kafka-ephemeral.yaml | 6 +++++- e2e/keda/files/keda-it-kafkatopic-to-log-auto.yaml | 13 +++++++------ e2e/keda/keda_kafka_test.go | 10 +++++++--- 3 files changed, 19 insertions(+), 10 deletions(-) diff --git a/e2e/keda/files/kafka-ephemeral.yaml b/e2e/keda/files/kafka-ephemeral.yaml index d1e6512db..794101634 100644 --- a/e2e/keda/files/kafka-ephemeral.yaml +++ b/e2e/keda/files/kafka-ephemeral.yaml @@ -45,6 +45,8 @@ spec: value: "PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093" - name: KAFKA_ADVERTISED_LISTENERS value: "PLAINTEXT://my-cluster-kafka-bootstrap.keda.svc:9092" + - name: KAFKA_INTER_BROKER_LISTENER_NAME + value: "PLAINTEXT" - name: KAFKA_CONTROLLER_LISTENER_NAMES value: "CONTROLLER" - name: KAFKA_LISTENER_SECURITY_PROTOCOL_MAP @@ -53,6 +55,8 @@ spec: value: "1@localhost:9093" - name: KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR value: "1" + - name: KAFKA_OFFSETS_TOPIC_NUM_PARTITIONS + value: "1" - name: KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR value: "1" - name: KAFKA_TRANSACTION_STATE_LOG_MIN_ISR @@ -83,4 +87,4 @@ spec: ports: - name: plaintext port: 9092 - targetPort: 9092 + targetPort: 9092 \ No newline at end of file diff --git a/e2e/keda/files/keda-it-kafkatopic-to-log-auto.yaml b/e2e/keda/files/keda-it-kafkatopic-to-log-auto.yaml index 75b5bfc85..34d16ec9b 100644 --- a/e2e/keda/files/keda-it-kafkatopic-to-log-auto.yaml +++ b/e2e/keda/files/keda-it-kafkatopic-to-log-auto.yaml @@ -20,11 +20,12 @@ kind: Integration metadata: name: keda-kafka-auto-discovery spec: - sources: - - content: | - from("kafka:my-topic?brokers=my-cluster-kafka-bootstrap.keda.svc:9092&groupId=auto-group") - .to("log:info"); - name: routes.java + flows: + - route: + from: + steps: + - to: log:info + uri: kafka:my-topic?brokers=my-cluster-kafka-bootstrap.keda.svc:9092&groupId=auto-group traits: keda: enabled: true @@ -32,4 +33,4 @@ spec: maxReplicaCount: 1 cooldownPeriod: 20 pollingInterval: 10 - # No triggers specified - auto-discovery should create kafka scaler + # No triggers specified - auto-discovery should create kafka scaler \ No newline at end of file diff --git a/e2e/keda/keda_kafka_test.go b/e2e/keda/keda_kafka_test.go index db8245adc..1eb7b34eb 100644 --- a/e2e/keda/keda_kafka_test.go +++ b/e2e/keda/keda_kafka_test.go @@ -36,6 +36,10 @@ func TestKafkaKedaAutoscale(t *testing.T) { WithExistingNamedTestNamespace(t, func(ctx context.Context, g *WithT, ns string) { ExpectExecSucceed(t, g, Kubectl("apply", "-f", "files/kafka-ephemeral.yaml", "-n", ns)) ExpectExecSucceed(t, g, Kubectl("rollout", "status", "deployment", "my-cluster-kafka", "-n", ns, "--timeout=180s")) + ExpectExecSucceed(t, g, Kubectl("exec", "deployment/my-cluster-kafka", "-n", ns, "--", + "/opt/kafka/bin/kafka-topics.sh", "--create", "--if-not-exists", + "--topic", "my-topic", "--bootstrap-server", "localhost:9092", + "--partitions", "1", "--replication-factor", "1")) t.Run("Serverless Kafka Autoscale", func(t *testing.T) { consumerName := "keda-kafka-auto-discovery" @@ -44,7 +48,7 @@ func TestKafkaKedaAutoscale(t *testing.T) { ExpectExecSucceed(t, g, Kubectl("apply", "-f", "files/keda-it-kafkatopic-to-log-auto.yaml", "-n", ns)) // Start consumer - g.Eventually(IntegrationConditionStatus(t, ctx, ns, consumerName, v1.IntegrationConditionReady), TestTimeoutMedium). + g.Eventually(IntegrationConditionStatus(t, ctx, ns, consumerName, v1.IntegrationConditionReady), TestTimeoutLong). Should(Equal(corev1.ConditionTrue)) g.Eventually(IntegrationStatusReplicas(t, ctx, ns, consumerName), TestTimeoutShort). Should(PointTo(BeNumerically("==", 1))) @@ -55,7 +59,7 @@ func TestKafkaKedaAutoscale(t *testing.T) { // Start producer to generate traffic ExpectExecSucceed(t, g, Kubectl("apply", "-f", "files/timer-to-kafka.yaml", "-n", ns)) - g.Eventually(PipeConditionStatus(t, ctx, ns, producerName, v1.PipeConditionReady), TestTimeoutMedium). + g.Eventually(PipeConditionStatus(t, ctx, ns, producerName, v1.PipeConditionReady), TestTimeoutLong). Should(Equal(corev1.ConditionTrue)) // Consumer scales up to 1 @@ -160,4 +164,4 @@ func TestKafkaKedaAutoMetadata(t *testing.T) { ExpectExecSucceed(t, g, Kubectl("delete", "it", integrationName, "-n", ns)) }) }, "keda") -} +} \ No newline at end of file
