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 d0b27857772e196edd5c7cb2762952e484b8fb3b Author: Keerthan <[email protected]> AuthorDate: Wed Sep 23 21:38:08 2026 +0530 Fix #5935: Fallback to existing namespace broker when applyTo is omitted --- pkg/util/bindings/arkmq.go | 38 +++++++++++- pkg/util/bindings/arkmq_test.go | 130 +++++++++++++++++++++++++++++++++++++++- 2 files changed, 164 insertions(+), 4 deletions(-) diff --git a/pkg/util/bindings/arkmq.go b/pkg/util/bindings/arkmq.go index 6c15f481c..21599d138 100644 --- a/pkg/util/bindings/arkmq.go +++ b/pkg/util/bindings/arkmq.go @@ -175,18 +175,50 @@ func (a ArkMQBindingProvider) lookupBrokerURL(ctx BindingContext, address *arkmq if clusterName == "" && address.Labels != nil { clusterName = address.Labels[arkmqv1beta1.ArkMQBrokerLabel] } - if clusterName == "" { - return "", fmt.Errorf("no %q label or applyTo defined on address %s", arkmqv1beta1.ArkMQBrokerLabel, endpoint.Ref.Name) - } namespace := endpoint.Ref.Namespace if namespace == "" { namespace = ctx.Namespace } + if clusterName == "" { + // By default, the resource is reconciled by the existing broker in the same namespace. + // Fallback to looking for an ActiveMQArtemis broker in the namespace. + fallbackName, err := a.fallbackBrokerName(ctx, namespace, endpoint.Ref.Name) + if err != nil { + return "", err + } + clusterName = fallbackName + } + return a.getBrokerURL(ctx, clusterName, namespace) } +func (a ArkMQBindingProvider) fallbackBrokerName(ctx BindingContext, namespace, addressName string) (string, error) { + client := a.Client + if client == nil { + arkmqClient, err := internalclientset.NewForConfig(ctx.Client.GetConfig()) + if err != nil { + return "", err + } + client = arkmqClient + } + + brokers, err := client.BrokerV1beta1().ActiveMQArtemises(namespace).List(ctx.Ctx, v1.ListOptions{}) + if err != nil { + return "", err + } + if len(brokers.Items) == 0 { + return "", fmt.Errorf("no ActiveMQArtemis broker found in namespace %s for address %s", namespace, addressName) + } + if len(brokers.Items) > 1 { + return "", fmt.Errorf("multiple ActiveMQArtemis brokers found in namespace %s (%d found); specify %q label or applyTo on address %s", + namespace, len(brokers.Items), arkmqv1beta1.ArkMQBrokerLabel, addressName) + } + + return brokers.Items[0].Name, nil +} + func (a ArkMQBindingProvider) getBrokerURL(ctx BindingContext, clusterName, namespace string) (string, error) { client := a.Client if client == nil { diff --git a/pkg/util/bindings/arkmq_test.go b/pkg/util/bindings/arkmq_test.go index 529cd8cf5..3a3be32e8 100644 --- a/pkg/util/bindings/arkmq_test.go +++ b/pkg/util/bindings/arkmq_test.go @@ -464,7 +464,135 @@ func TestArkMQMissingBrokerLabel(t *testing.T) { Type: camelv1.EndpointTypeSink, }, endpoint) require.Error(t, err) - assert.Contains(t, err.Error(), "no \"ActiveMQArtemis\" label or applyTo defined on address") + assert.Contains(t, err.Error(), "no ActiveMQArtemis broker found in namespace test") +} + +func TestArkMQBrokerFallback(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + broker := &arkmqv1beta1.ActiveMQArtemis{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "test", + Name: "default-broker", + }, + Status: arkmqv1beta1.ActiveMQArtemisStatus{ + PortStatus: []arkmqv1beta1.ActiveMQArtemisPortStatus{ + { + Name: "core", + Port: 61616, + }, + }, + }, + } + + address := &arkmqv1beta1.ActiveMQArtemisAddress{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "test", + Name: "unattached-address", + }, + Spec: arkmqv1beta1.ActiveMQArtemisAddressSpec{ + QueueName: "my-fallback-queue", + // No applyTo and no labels! + }, + } + + svc := &corev1.Service{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "test", + Name: "default-broker-hdls-svc", + }, + Spec: corev1.ServiceSpec{ + Ports: []corev1.ServicePort{ + { + Name: "core", + Port: 61616, + }, + }, + }, + } + + client, err := internal.NewFakeClient(svc) + require.NoError(t, err) + + bindingContext := BindingContext{ + Ctx: ctx, + Client: client, + Namespace: "test", + Profile: camelv1.TraitProfileKubernetes, + } + + endpoint := camelv1.Endpoint{ + Ref: &corev1.ObjectReference{ + Kind: "ActiveMQArtemisAddress", + Name: "unattached-address", + APIVersion: "broker.amq.io/v1beta1", + }, + } + + provider := ArkMQBindingProvider{ + Client: fake.NewSimpleClientset(broker, address), + } + + binding, err := provider.Translate(bindingContext, EndpointContext{ + Type: camelv1.EndpointTypeSink, + }, endpoint) + require.NoError(t, err) + assert.NotNil(t, binding) + assert.Equal(t, "jms:queue:my-fallback-queue?brokerURL=tcp%3A%2F%2Fdefault-broker-hdls-svc.test.svc%3A61616", binding.URI) +} + +func TestArkMQMultipleBrokersFallback(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + broker1 := &arkmqv1beta1.ActiveMQArtemis{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "test", + Name: "broker-1", + }, + } + broker2 := &arkmqv1beta1.ActiveMQArtemis{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "test", + Name: "broker-2", + }, + } + + address := &arkmqv1beta1.ActiveMQArtemisAddress{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "test", + Name: "unattached-address", + }, + } + + client, err := internal.NewFakeClient() + require.NoError(t, err) + + bindingContext := BindingContext{ + Ctx: ctx, + Client: client, + Namespace: "test", + Profile: camelv1.TraitProfileKubernetes, + } + + endpoint := camelv1.Endpoint{ + Ref: &corev1.ObjectReference{ + Kind: "ActiveMQArtemisAddress", + Name: "unattached-address", + APIVersion: "broker.amq.io/v1beta1", + }, + } + + provider := ArkMQBindingProvider{ + Client: fake.NewSimpleClientset(broker1, broker2, address), + } + + _, err = provider.Translate(bindingContext, EndpointContext{ + Type: camelv1.EndpointTypeSink, + }, endpoint) + require.Error(t, err) + assert.Contains(t, err.Error(), "multiple ActiveMQArtemis brokers found in namespace test (2 found)") } func TestArkMQPassThrough(t *testing.T) {
