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 6e95ab0ff35e99651ec9111dffbcb08d5d75f3b1 Author: Keerthan <[email protected]> AuthorDate: Wed Sep 23 21:32:11 2026 +0530 Fix #5935: Verify Kubernetes Service in ArkMQ binding and wait for queue in e2e setup --- e2e/arkmq/setup/setup.sh | 1 + pkg/util/bindings/arkmq.go | 34 ++++++++++++++- pkg/util/bindings/arkmq_test.go | 94 +++++++++++++++++++++++++++++++++++++++-- 3 files changed, 124 insertions(+), 5 deletions(-) diff --git a/e2e/arkmq/setup/setup.sh b/e2e/arkmq/setup/setup.sh index 59afa39f7..c6b0030f5 100644 --- a/e2e/arkmq/setup/setup.sh +++ b/e2e/arkmq/setup/setup.sh @@ -39,3 +39,4 @@ kubectl wait activemqartemis/my-broker --for=condition=Ready --timeout=300s -n a #### Setup an ActiveMQ Artemis queue address kubectl apply -f $SCRIPT_DIR/queue.yaml -n arkmq +kubectl wait activemqartemisaddress/my-queue --for=condition=Ready --timeout=120s -n arkmq diff --git a/pkg/util/bindings/arkmq.go b/pkg/util/bindings/arkmq.go index 1e1797a8e..6c15f481c 100644 --- a/pkg/util/bindings/arkmq.go +++ b/pkg/util/bindings/arkmq.go @@ -23,6 +23,7 @@ import ( camelv1 "github.com/apache/camel-k/v2/pkg/apis/camel/v1" arkmqv1beta1 "github.com/apache/camel-k/v2/pkg/apis/duck/arkmq/v1beta1" "github.com/apache/camel-k/v2/pkg/client/arkmq/clientset/internalclientset" + "github.com/apache/camel-k/v2/pkg/util/kubernetes" "github.com/apache/camel-k/v2/pkg/util/uri" k8serrors "k8s.io/apimachinery/pkg/api/errors" v1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -211,10 +212,39 @@ func (a ArkMQBindingProvider) getBrokerURL(ctx BindingContext, clusterName, name } } + svcName := fmt.Sprintf("%s-hdls-svc", clusterName) + if ctx.Client != nil { + svc, err := kubernetes.LookupService(ctx.Ctx, ctx.Client, namespace, svcName) + if err != nil { + return "", err + } + if svc == nil { + svc, err = kubernetes.LookupService(ctx.Ctx, ctx.Client, namespace, clusterName) + if err != nil { + return "", err + } + } + if svc == nil { + return "", fmt.Errorf("could not find service %s in namespace %s for ArkMQ broker %s", svcName, namespace, clusterName) + } + + for _, p := range svc.Spec.Ports { + if p.Name == "core" || p.Name == "all" || p.Name == "openwire" || p.Port == 61616 { + port = p.Port + break + } + } + + if svc.Namespace != "" { + return fmt.Sprintf("tcp://%s.%s.svc:%d", svc.Name, svc.Namespace, port), nil + } + return fmt.Sprintf("tcp://%s:%d", svc.Name, port), nil + } + if namespace != "" { - return fmt.Sprintf("tcp://%s-hdls-svc.%s.svc:%d", clusterName, namespace, port), nil + return fmt.Sprintf("tcp://%s.%s.svc:%d", svcName, namespace, port), nil } - return fmt.Sprintf("tcp://%s-hdls-svc:%d", clusterName, port), nil + return fmt.Sprintf("tcp://%s:%d", svcName, port), nil } func (a ArkMQBindingProvider) lookupAddress(ctx BindingContext, endpoint camelv1.Endpoint) (*arkmqv1beta1.ActiveMQArtemisAddress, error) { diff --git a/pkg/util/bindings/arkmq_test.go b/pkg/util/bindings/arkmq_test.go index 5994f5214..529cd8cf5 100644 --- a/pkg/util/bindings/arkmq_test.go +++ b/pkg/util/bindings/arkmq_test.go @@ -95,7 +95,22 @@ func TestArkMQLookupAddress(t *testing.T) { }, } - client, err := internal.NewFakeClient() + svc := &corev1.Service{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "test", + Name: "mybroker-hdls-svc", + }, + Spec: corev1.ServiceSpec{ + Ports: []corev1.ServicePort{ + { + Name: "core", + Port: 61616, + }, + }, + }, + } + + client, err := internal.NewFakeClient(svc) require.NoError(t, err) bindingContext := BindingContext{ @@ -193,7 +208,22 @@ func TestArkMQLookupAddressByName(t *testing.T) { }, } - client, err := internal.NewFakeClient() + svc := &corev1.Service{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "test", + Name: "mybroker-hdls-svc", + }, + Spec: corev1.ServiceSpec{ + Ports: []corev1.ServicePort{ + { + Name: "core", + Port: 61616, + }, + }, + }, + } + + client, err := internal.NewFakeClient(svc) require.NoError(t, err) bindingContext := BindingContext{ @@ -242,7 +272,22 @@ func TestArkMQBrokerDirect(t *testing.T) { }, } - client, err := internal.NewFakeClient() + svc := &corev1.Service{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "test", + Name: "mybroker-hdls-svc", + }, + Spec: corev1.ServiceSpec{ + Ports: []corev1.ServicePort{ + { + Name: "core", + Port: 61616, + }, + }, + }, + } + + client, err := internal.NewFakeClient(svc) require.NoError(t, err) bindingContext := BindingContext{ @@ -308,6 +353,49 @@ func TestArkMQBrokerDirect(t *testing.T) { assert.Contains(t, err.Error(), "missing destination or queue property") } +func TestArkMQMissingService(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + broker := &arkmqv1beta1.ActiveMQArtemis{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "test", + Name: "mybroker", + }, + } + + client, err := internal.NewFakeClient() // no service in client! + require.NoError(t, err) + + bindingContext := BindingContext{ + Ctx: ctx, + Client: client, + Namespace: "test", + Profile: camelv1.TraitProfileKubernetes, + } + + endpoint := camelv1.Endpoint{ + Ref: &corev1.ObjectReference{ + Kind: "ActiveMQArtemis", + Name: "mybroker", + APIVersion: "broker.amq.io/v1beta1", + }, + Properties: asEndpointProperties(map[string]string{ + "destination": "orders", + }), + } + + provider := ArkMQBindingProvider{ + Client: fake.NewSimpleClientset(broker), + } + + _, err = provider.Translate(bindingContext, EndpointContext{ + Type: camelv1.EndpointTypeSink, + }, endpoint) + require.Error(t, err) + assert.Contains(t, err.Error(), "could not find service mybroker-hdls-svc in namespace test") +} + func TestArkMQUnsupportedKind(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel()
