fix: use startObservation for parent observation to preserve custom conventions

When a parent observation was present, the observation was created using
Observation.createNotStarted() with just the convention name, bypassing
the customObservationConvention and ObservationDocumentation setup.

Refactor startObservation to accept an optional parent observation and
extract the observe-and-send logic into a private method, ensuring both
code paths use the same convention/documentation pipeline.

Author: Igor Macedo Quintanilha

spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/operations/AbstractMessagingTemplate.java

@@ -305,32 +305,22 @@ public <T> CompletableFuture<SendResult<T>> sendAsync(@Nullable String endpointN
 		// Capture parent observation on the calling thread to propagate trace context across async boundary
 		var parentObservation = this.observationRegistry.getCurrentObservation();
 
-		return preProcessMessageForSendAsync(endpointToUse, message).thenCompose(preprocessedMessage -> {
-			// Create observation context with preprocessed message (now includes FIFO headers if applicable)
-			var context = this.observationSpecifics.createContext(preprocessedMessage, endpointToUse);
-
-			// Start observation and link to parent to maintain trace continuity
-			Observation observation;
-			if (parentObservation != null) {
-				observation = Observation.createNotStarted(this.observationSpecifics.getDefaultConvention().getName(),
-						() -> context, this.observationRegistry).parentObservation(parentObservation).start();
-			}
-			else {
-				observation = startObservation(context);
-			}
-
-			// Add trace headers to the message
-			var carrier = Objects.requireNonNull(context.getCarrier(), "No carrier found in context.");
-			var messageWithObservationHeaders = MessageHeaderUtils.addHeadersIfAbsent(preprocessedMessage, carrier);
+		return preProcessMessageForSendAsync(endpointToUse, message).thenCompose(
+				preprocessedMessage -> observeAndSend(preprocessedMessage, message, endpointToUse, parentObservation));
+	}
 
-			return doSendAndCompleteObservation(messageWithObservationHeaders, endpointToUse, context, observation)
-					.exceptionallyCompose(
-							t -> CompletableFuture.failedFuture(new MessagingOperationFailedException(
-									"Message send operation failed for message %s to endpoint %s"
-											.formatted(MessageHeaderUtils.getId(message), endpointToUse),
-									endpointToUse, message, t)))
-					.whenComplete((v, t) -> logSendMessageResult(endpointToUse, message, t));
-		});
+	private <T> CompletableFuture<SendResult<T>> observeAndSend(Message<T> preprocessedMessage,
+			Message<T> originalMessage, String endpointToUse, @Nullable Observation parentObservation) {
+		var context = this.observationSpecifics.createContext(preprocessedMessage, endpointToUse);
+		Observation observation = startObservation(context, parentObservation);
+		var carrier = Objects.requireNonNull(context.getCarrier(), "No carrier found in context.");
+		var messageWithObservationHeaders = MessageHeaderUtils.addHeadersIfAbsent(preprocessedMessage, carrier);
+		return doSendAndCompleteObservation(messageWithObservationHeaders, endpointToUse, context, observation)
+				.exceptionallyCompose(t -> CompletableFuture.failedFuture(new MessagingOperationFailedException(
+						"Message send operation failed for message %s to endpoint %s"
+								.formatted(MessageHeaderUtils.getId(originalMessage), endpointToUse),
+						endpointToUse, originalMessage, t)))
+				.whenComplete((v, t) -> logSendMessageResult(endpointToUse, originalMessage, t));
 	}
 
 	private <T> CompletableFuture<SendResult<T>> doSendAndCompleteObservation(Message<T> message, String endpointToUse,
@@ -353,13 +343,18 @@ private void completeObservation(@Nullable SendResult<?> sendResult, AbstractTem
 	}
 
 	@SuppressWarnings("unchecked")
-	private <Context extends Observation.Context> Observation startObservation(Context observationContext) {
+	private <Context extends Observation.Context> Observation startObservation(Context observationContext,
+			@Nullable Observation parentObservation) {
 		ObservationConvention<Context> defaultConvention = (ObservationConvention<Context>) observationSpecifics
 				.getDefaultConvention();
 		ObservationConvention<Context> customConvention = (ObservationConvention<Context>) this.customObservationConvention;
 		ObservationDocumentation documentation = observationSpecifics.getDocumentation();
-		return documentation.start(customConvention, defaultConvention, () -> observationContext,
-				this.observationRegistry);
+		Observation observation = documentation.observation(customConvention, defaultConvention,
+				() -> observationContext, this.observationRegistry);
+		if (parentObservation != null) {
+			observation.parentObservation(parentObservation);
+		}
+		return observation.start();
 	}
 
 	protected abstract <T> Message<T> preProcessMessageForSend(String endpointToUse, Message<T> message);

spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/operations/SqsTemplateObservationTest.java

@@ -22,6 +22,7 @@
 import io.awspring.cloud.sqs.listener.SqsHeaders;
 import io.awspring.cloud.sqs.support.observation.SqsTemplateObservation;
 import io.micrometer.common.KeyValues;
+import io.micrometer.observation.Observation;
 import io.micrometer.observation.tck.TestObservationRegistry;
 import io.micrometer.observation.tck.TestObservationRegistryAssert;
 import java.util.UUID;
@@ -163,6 +164,44 @@ void shouldCaptureErrorsInObservation() {
 				.hasSingleObservationThat().hasError().assertThatError().isInstanceOf(RuntimeException.class);
 	}
 
+	@Test
+	void shouldApplyCustomConventionWhenParentObservationIsPresent() {
+		// given
+		SqsTemplateObservation.Convention customConvention = mock(SqsTemplateObservation.Convention.class);
+		given(customConvention.supportsContext(any())).willReturn(true);
+		given(customConvention.getName()).willReturn("spring.aws.sqs.template");
+
+		String lowCardinalityCustomKeyName = "custom.lowCardinality.key";
+		String lowCardinalityCustomValue = "custom-lowCardinality-value";
+		String highCardinalityCustomKeyName = "custom.highCardinality.key";
+		String highCardinalityCustomValue = "custom-highCardinality-value";
+		given(customConvention.getLowCardinalityKeyValues(any()))
+				.willReturn(KeyValues.of(lowCardinalityCustomKeyName, lowCardinalityCustomValue));
+		given(customConvention.getHighCardinalityKeyValues(any()))
+				.willReturn(KeyValues.of(highCardinalityCustomKeyName, highCardinalityCustomValue));
+
+		TestObservationRegistry customRegistry = TestObservationRegistry.create();
+
+		SqsTemplate templateWithCustomConvention = SqsTemplate.builder().sqsAsyncClient(mockSqsAsyncClient)
+				.configure(
+						options -> options.observationRegistry(customRegistry).observationConvention(customConvention))
+				.build();
+
+		// when - send within a parent observation scope
+		Observation parentObservation = Observation.start("parent-observation", customRegistry);
+		try (var ignored = parentObservation.openScope()) {
+			templateWithCustomConvention.send(queueName, "test-payload");
+		}
+		finally {
+			parentObservation.stop();
+		}
+
+		// then - custom convention should be applied even with parent observation
+		TestObservationRegistryAssert.then(customRegistry).hasNumberOfObservationsEqualTo(2)
+				.hasAnObservationWithAKeyValue(lowCardinalityCustomKeyName, lowCardinalityCustomValue)
+				.hasAnObservationWithAKeyValue(highCardinalityCustomKeyName, highCardinalityCustomValue);
+	}
+
 	@Test
 	void shouldSupportCustomKeyValuesInActiveSending() {
 		// given