feat: return kubeconfig for service monitor

# About this branch: `monitor_integration_headers_only`

**This is a debugging isolation branch, not a feature branch. Do not merge it as-is.**

## Why it exists

It was created off commit `35ceebf` ("feat: return kubeconfig for service monitor") to isolate

that commit's changes from `a46fb32` ("fix: harden kubeconfig retrieval and registry secrets

listeners against lost/silent failures"), which was suspected as a possible cause of a manual

Karate test failure (a service order stuck at `FEASIBILITY_CHECKED`, "Service order not

completed after 2 min").

**Included:** the kubeconfig-response header propagation and the `.onFailure()` ->

`Status.FAILURE` response added in `35ceebf` (`RegistryService.retrieveKubernetesConfigFromKafkaMessage`,

`RegistryKafkaProducer.sendKubeConfigResponseMessage(response, headers)`).

**Deliberately excluded:** everything from `a46fb32` - the ack/nack timing rework in

`RegistrySecretsKafkaListener` (`receiveRegistrySecretsMessage`, `retrieveKubernetesConfig`),

the retry-with-backoff logic, the blank-ID/not-found handling, and the global

`quarkus.rest-client.vault-api.connect-timeout`/`read-timeout` settings.

## What we learned from it

- The original Feasibility Check hang turned out to be unrelated to the service-monitor

  diagnosis flow (`registry-retrieve-kubernetes-config` / `registry-response-kubernetes-config-outgoing-channel`)

  entirely - that flow is a side-flow that doesn't gate the service order.

- Along the way we found a real, separate bug: `service.monitor`'s kubeconfig retrieval

  requests carry no `Authorization` header, traced back to sonata's Kogito BPMN process

  (`serviceDeploymentProcess.bpmn2`), not a `service.monitor` bug.

- The actual Feasibility Check / service-order hang is still unresolved - see

  `~/todo_revised.md`, VACATION section, item 2.

## Status / next steps

See `~/todo_revised.md`:

- TODO item 11 (a-d): revise the ack/nack design, decide on `failure-strategy=dead-letter-queue`,

  then reapply `a46fb32` onto `monitor_integration` accordingly, then fix that branch's

  integration tests against the final version.

- VACATION item 1: the `service.monitor`/sonata missing-`Authorization`-header fix.

- VACATION item 2: the still-open original Feasibility Check hang investigation.

Once those are resolved, this branch has served its purpose and can be deleted - its findings

belong in the real fix on `monitor_integration`, not in this branch itself.

        logger.debugf(ApplicationProperties.PAYLOAD_S, responseMessage.toStringAsJson());

    }

    public void sendKubeConfigResponseMessage(

            VaultSecretResponseMessage responseMessage, Headers headers) {

        OutgoingKafkaRecordMetadata<Object> kafkaMetadata =

                OutgoingKafkaRecordMetadata.builder().withHeaders(headers).build();

        vaultSecretResponseMessageEmitter.send(Message.of(responseMessage, Metadata.of(kafkaMetadata)));

        logger.debugf(

                "Kube Conf Response data message with headers sent to KAFKA topic: %s",

                REGISTRY_RESPONSE_KUBE_CONF_CHANNEL);

        logger.debugf(ApplicationProperties.PAYLOAD_S, responseMessage.toStringAsJson());

    }

    public void sendOrganizationCredentialsResponseMessage(

            CloudEventRegistryOrganizationCredentialsResult responseMessage) {

        organizationCredentialsResultEmitter.send(responseMessage);

    }

    public Uni<Void> retrieveKubernetesConfigFromKafkaMessage(Message<String> message) {

        String serviceId = message.getPayload();

        Headers headers = extractKafkaHeaders(message);

        return Uni.createFrom()

                .voidItem()

                .invoke(

                            VaultSecretResponseMessage response =

                                    getKubernetesConfigSecretResponseMessage(message);

                            logger.debugf(

                                    "Successfully retrieved KubeConfig for ID: %s. Sending response.",

                                    message.getPayload());

                            registryKafkaProducer.sendKubeConfigResponseMessage(response);

                                    "Successfully retrieved KubeConfig for ID: %s. Sending response.", serviceId);

                            registryKafkaProducer.sendKubeConfigResponseMessage(response, headers);

                        })

                .onFailure()

                .invoke(

                        failure -> {

                            VaultSecretResponseMessage failureResponse =

                                    buildKubernetesConfigSecretResponseMessage(

                                            null, serviceId, Status.FAILURE, failure.getMessage());

                            registryKafkaProducer.sendKubeConfigResponseMessage(failureResponse, headers);

                        });

    }

## Test Environment

%test.quarkus.oidc.enabled=false

%test.quarkus.otel.enabled=false

%test.quarkus.kafka.devservices.enabled=false

%test.quarkus.vault.devservices.enabled=false

%test.quarkus.rest-client.vault-api.url=http://localhost:8200

# %test.kubernetes.namespace-sa.vault.secret-kv-path=test/kube-sa/path

                        () -> assertThat(vaultSecretResponseMessageInMemorySink.received()).hasSize(1));

    }

    @Test

    void sendKubeConfigResponseMessage_withHeaders() {

        VaultSecretResponseMessage response = new VaultSecretResponseMessage();

        response.setId("vault-id");

        response.setStatus(Status.SUCCESS);

        String value = "value";

        RecordHeader recordHeader = new RecordHeader("correlationId", value.getBytes());

        RecordHeaders recordHeaders = new RecordHeaders();

        recordHeaders.add(recordHeader);

        producer.sendKubeConfigResponseMessage(response, recordHeaders);

        await()

                .atMost(Duration.ofSeconds(5))

                .untilAsserted(

                        () -> assertThat(vaultSecretResponseMessageInMemorySink.received()).hasSize(1));

        Message<VaultSecretResponseMessage> message =

                vaultSecretResponseMessageInMemorySink.received().get(0);

        assertThat(message.getPayload().getId()).isNotBlank().isEqualTo("vault-id");

        Headers headers =

                message.getMetadata(OutgoingKafkaRecordMetadata.class).orElseThrow().getHeaders();

        assertThat(headers.toArray()).isNotEmpty().hasSize(1);

        Header header = headers.lastHeader("correlationId");

        assertThat(header).isNotNull();

        String stringValue = new String(header.value(), StandardCharsets.UTF_8);

        assertThat(stringValue).isEqualTo("value");

    }

    @Test

    void sendOrganizationCredentialsResponseMessage_Success() {

        CloudEventRegistryOrganizationCredentialsResult response =