diff --git a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/FluxTrace.java b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/FluxTrace.java index 8649598d6f23..a6908ec0f5ec 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/FluxTrace.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/FluxTrace.java @@ -60,7 +60,10 @@ protected void hookOnNext(ServiceBusMessageContext message) { return; } - Context span = instrumentation.startProcessInstrumentation("ServiceBus.process", message.getMessage(), Context.NONE); + Context span = instrumentation.startProcessInstrumentation("ServiceBus.process", + message.getMessage().getApplicationProperties(), + message.getMessage().getEnqueuedTime(), + Context.NONE); message.getMessage().setContext(span); AutoCloseable scope = tracer.makeSpanCurrent(span); try { diff --git a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/FluxTraceV2.java b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/FluxTraceV2.java deleted file mode 100644 index 33c9b3021780..000000000000 --- a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/FluxTraceV2.java +++ /dev/null @@ -1,71 +0,0 @@ -// Copyright (c) Microsoft Corporation. All rights reserved. -// Licensed under the MIT License. - -package com.azure.messaging.servicebus; - -import com.azure.messaging.servicebus.implementation.instrumentation.ReceiverKind; -import com.azure.messaging.servicebus.implementation.instrumentation.ServiceBusReceiverInstrumentation; -import org.reactivestreams.Subscription; -import reactor.core.CoreSubscriber; -import reactor.core.publisher.BaseSubscriber; -import reactor.core.publisher.Flux; -import reactor.core.publisher.FluxOperator; - -import java.util.Objects; - -/** - * Flux operator that traces receive and process calls - */ -final class FluxTraceV2 extends FluxOperator { - private final ServiceBusReceiverInstrumentation instrumentation; - - FluxTraceV2(Flux upstream, ServiceBusReceiverInstrumentation instrumentation) { - super(upstream); - this.instrumentation = instrumentation; - } - - @Override - public void subscribe(CoreSubscriber coreSubscriber) { - Objects.requireNonNull(coreSubscriber, "'coreSubscriber' cannot be null."); - - source.subscribe(new TracingSubscriber(coreSubscriber, instrumentation)); - } - - private static class TracingSubscriber extends BaseSubscriber { - - private final CoreSubscriber downstream; - private final ServiceBusReceiverInstrumentation instrumentation; - TracingSubscriber(CoreSubscriber downstream, ServiceBusReceiverInstrumentation instrumentation) { - this.downstream = downstream; - this.instrumentation = instrumentation; - } - - @Override - public reactor.util.context.Context currentContext() { - return downstream.currentContext(); - } - - @Override - protected void hookOnSubscribe(Subscription subscription) { - downstream.onSubscribe(this); - } - - @Override - protected void hookOnNext(ServiceBusReceivedMessage message) { - instrumentation.instrumentProcess(message, ReceiverKind.ASYNC_RECEIVER, msg -> { - downstream.onNext(msg); - return null; - }); - } - - @Override - protected void hookOnError(Throwable throwable) { - downstream.onError(throwable); - } - - @Override - protected void hookOnComplete() { - downstream.onComplete(); - } - } -} diff --git a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusAsyncConsumer.java b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusAsyncConsumer.java index 280490f15070..fcddd200a436 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusAsyncConsumer.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusAsyncConsumer.java @@ -49,19 +49,10 @@ class ServiceBusAsyncConsumer implements AutoCloseable { this.linkProcessor = null; this.messageSerializer = messageSerializer; - final boolean useFluxTrace = instrumentation.isEnabled() && instrumentation.isAsyncReceiverInstrumentation(); - if (useFluxTrace) { - // This ServiceBusAsyncConsumer is backing ServiceBusReceiverAsyncClient instance (client has instrumentation is enabled). - final Flux deserialize = messageFlux - .map(message -> this.messageSerializer.deserialize(message, ServiceBusReceivedMessage.class)); - this.processor = new FluxTraceV2(deserialize, instrumentation); - } else { - // This ServiceBusAsyncConsumer is backing either - // 1. a ServiceBusReceiverAsyncClient instance (client has no instrumentation enabled) - // 2. Or a ServiceBusProcessorClient instance (processor client internally deal with instrumentation). - this.processor = messageFlux - .map(message -> this.messageSerializer.deserialize(message, ServiceBusReceivedMessage.class)); - } + // This ServiceBusAsyncConsumer is backing ServiceBusReceiverAsyncClient instance (client has instrumentation is enabled). + final Flux deserialize = messageFlux + .map(message -> this.messageSerializer.deserialize(message, ServiceBusReceivedMessage.class)); + this.processor = TracingFluxOperator.create(deserialize, instrumentation); } /** diff --git a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSessionReceiverAsyncClient.java b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSessionReceiverAsyncClient.java index 9cb63ce4a85f..0e7c2205227a 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSessionReceiverAsyncClient.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSessionReceiverAsyncClient.java @@ -274,7 +274,7 @@ private Mono acquireSpecificOrNextSession(String session, null, receiverOptions.getMaxLockRenewDuration()); final ServiceBusSingleSessionManager sessionManager = new ServiceBusSingleSessionManager(LOGGER, identifier, - sessionReceiver, receiverOptions.getPrefetchCount(), messageSerializer, connectionCacheWrapper.getRetryOptions()); + sessionReceiver, receiverOptions.getPrefetchCount(), messageSerializer, connectionCacheWrapper.getRetryOptions(), instrumentation); final ReceiverOptions newReceiverOptions = createNamedSessionOptions(receiverOptions.getReceiveMode(), receiverOptions.getPrefetchCount(), receiverOptions.getMaxLockRenewDuration(), diff --git a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSingleSessionManager.java b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSingleSessionManager.java index 702ac378cc5d..cccad6ed8d4d 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSingleSessionManager.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSingleSessionManager.java @@ -11,7 +11,9 @@ import com.azure.core.util.logging.LoggingEventBuilder; import com.azure.messaging.servicebus.implementation.DispositionStatus; import com.azure.messaging.servicebus.implementation.MessageUtils; +import com.azure.messaging.servicebus.implementation.instrumentation.ServiceBusReceiverInstrumentation; import org.apache.qpid.proton.amqp.transport.DeliveryState; +import org.apache.qpid.proton.message.Message; import reactor.core.publisher.Flux; import reactor.core.publisher.FluxSink; import reactor.core.publisher.Mono; @@ -34,11 +36,12 @@ final class ServiceBusSingleSessionManager implements IServiceBusSessionManager private final MessageSerializer serializer; private final Duration operationTimeout; private final ServiceBusSessionReactorReceiver sessionReceiver; - private final MessageFlux messageFlux; + private final Flux messageFlux; + private final ServiceBusReceiverInstrumentation instrumentation; ServiceBusSingleSessionManager(ClientLogger logger, String identifier, ServiceBusSessionReactorReceiver sessionReceiver, int prefetch, - MessageSerializer serializer, AmqpRetryOptions retryOptions) { + MessageSerializer serializer, AmqpRetryOptions retryOptions, ServiceBusReceiverInstrumentation instrumentation) { this.logger = Objects.requireNonNull(logger, "logger cannot be null."); this.identifier = identifier; this.sessionReceiver = Objects.requireNonNull(sessionReceiver, "sessionReceiver cannot be null."); @@ -46,7 +49,10 @@ final class ServiceBusSingleSessionManager implements IServiceBusSessionManager Objects.requireNonNull(retryOptions, "retryOptions cannot be null."); this.operationTimeout = retryOptions.getTryTimeout(); final Flux messageFluxUpstream = new SessionReceiverStream(sessionReceiver).flux(); - this.messageFlux = new MessageFlux(messageFluxUpstream, prefetch, CreditFlowMode.RequestDriven, NULL_RETRY_POLICY); + this.instrumentation = Objects.requireNonNull(instrumentation, "instrumentation cannot be null"); + MessageFlux messageFluxLocal = new MessageFlux(messageFluxUpstream, prefetch, CreditFlowMode.RequestDriven, + NULL_RETRY_POLICY); + this.messageFlux = TracingFluxOperator.create(messageFluxLocal, instrumentation); } @Override diff --git a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/SessionsMessagePump.java b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/SessionsMessagePump.java index 2afe6d63cedc..fb3bb5d6e403 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/SessionsMessagePump.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/SessionsMessagePump.java @@ -177,7 +177,7 @@ final class SessionsMessagePump { this.processMessage = Objects.requireNonNull(processMessage, "'processMessage' cannot be null."); this.processError = Objects.requireNonNull(processError, "'processError' cannot be null."); this.onTerminate = Objects.requireNonNull(onTerminate, "'onTerminate' cannot be null."); - this.receiversTracker = new SessionReceiversTracker(logger, maxConcurrentSessions, fullyQualifiedNamespace, entityPath, receiveMode); + this.receiversTracker = new SessionReceiversTracker(logger, maxConcurrentSessions, fullyQualifiedNamespace, entityPath, receiveMode, instrumentation); this.nextSession = new NextSession(pumpId, fullyQualifiedNamespace, entityPath, sessionAcquirer).mono(); } @@ -645,14 +645,16 @@ static final class SessionReceiversTracker { private final String entityPath; private final ServiceBusReceiveMode receiveMode; private final ConcurrentHashMap receivers; + private final ServiceBusReceiverInstrumentation instrumentation; private SessionReceiversTracker(ClientLogger logger, int size, String fullyQualifiedNamespace, String entityPath, - ServiceBusReceiveMode receiveMode) { + ServiceBusReceiveMode receiveMode, ServiceBusReceiverInstrumentation instrumentation) { this.logger = logger; this.fullyQualifiedNamespace = fullyQualifiedNamespace; this.entityPath = entityPath; this.receiveMode = receiveMode; this.receivers = new ConcurrentHashMap<>(size); + this.instrumentation = instrumentation; } private void track(ServiceBusSessionReactorReceiver receiver) { @@ -749,11 +751,14 @@ private Mono updateDisposition(ServiceBusReceivedMessage message, Disposit final ServiceBusSessionReactorReceiver receiver = receivers.get(sessionId); final DeliveryState deliveryState = MessageUtils.getDeliveryState(dispositionStatus, deadLetterReason, deadLetterDescription, propertiesToModify, transactionContext); + + Mono updateDispositionMono; if (receiver != null) { - return receiver.updateDisposition(message.getLockToken(), deliveryState); + updateDispositionMono = receiver.updateDisposition(message.getLockToken(), deliveryState); } else { - return Mono.error(DeliveryNotOnLinkException.noMatchingDelivery(message.getLockToken(), deliveryState)); + updateDispositionMono = Mono.error(DeliveryNotOnLinkException.noMatchingDelivery(message.getLockToken(), deliveryState)); } + return instrumentation.instrumentSettlement(updateDispositionMono, message, message.getContext(), dispositionStatus); } private Mono checkNull(Object options, ServiceBusTransactionContext transactionContext) { diff --git a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/TracingFluxOperator.java b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/TracingFluxOperator.java new file mode 100644 index 000000000000..830f52086389 --- /dev/null +++ b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/TracingFluxOperator.java @@ -0,0 +1,82 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.messaging.servicebus; + +import com.azure.messaging.servicebus.implementation.instrumentation.ReceiverKind; +import com.azure.messaging.servicebus.implementation.instrumentation.ServiceBusReceiverInstrumentation; +import org.apache.qpid.proton.message.Message; +import org.reactivestreams.Subscription; +import reactor.core.CoreSubscriber; +import reactor.core.publisher.BaseSubscriber; +import reactor.core.publisher.Flux; +import reactor.core.publisher.FluxOperator; + +import java.util.Objects; +import java.util.function.BiConsumer; +import java.util.function.Function; + +/** + * Flux operator that traces receive and process calls + */ +final class TracingFluxOperator extends BaseSubscriber { + + public static Flux create(Flux upstream, ServiceBusReceiverInstrumentation instrumentation) { + if (!instrumentation.isEnabled() && instrumentation.isAsyncReceiverInstrumentation()) { + return upstream; + } + + return new FluxOperator(upstream) { + @SuppressWarnings("unchecked") + @Override + public void subscribe(CoreSubscriber actual) { + Objects.requireNonNull(actual, "'actual' cannot be null."); + source.subscribe(new TracingFluxOperator(actual, (msg, handler) -> { + if (msg instanceof Message) { + instrumentation.instrumentProcess((Message) msg, ReceiverKind.ASYNC_RECEIVER, + (Function) handler); + } else if (msg instanceof ServiceBusReceivedMessage) { + instrumentation.instrumentProcess((ServiceBusReceivedMessage) msg, ReceiverKind.ASYNC_RECEIVER, + (Function) handler); + } + })); + } + }; + } + + private final CoreSubscriber downstream; + private final BiConsumer> instrumentation; + private TracingFluxOperator(CoreSubscriber downstream, BiConsumer> instrumentation) { + this.downstream = downstream; + this.instrumentation = instrumentation; + } + + @Override + public reactor.util.context.Context currentContext() { + return downstream.currentContext(); + } + + @Override + protected void hookOnSubscribe(Subscription subscription) { + downstream.onSubscribe(this); + } + + @Override + protected void hookOnNext(T message) { + instrumentation.accept(message, msg -> { + downstream.onNext(msg); + return null; + }); + } + + @Override + protected void hookOnError(Throwable throwable) { + downstream.onError(throwable); + } + + @Override + protected void hookOnComplete() { + downstream.onComplete(); + } +} + diff --git a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/instrumentation/ServiceBusReceiverInstrumentation.java b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/instrumentation/ServiceBusReceiverInstrumentation.java index 5cb4616c97bd..cb54adcb41a2 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/instrumentation/ServiceBusReceiverInstrumentation.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/instrumentation/ServiceBusReceiverInstrumentation.java @@ -8,16 +8,25 @@ import com.azure.core.util.tracing.Tracer; import com.azure.messaging.servicebus.ServiceBusReceivedMessage; import com.azure.messaging.servicebus.implementation.DispositionStatus; +import org.apache.qpid.proton.amqp.Symbol; +import org.apache.qpid.proton.message.Message; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import java.time.Instant; +import java.time.OffsetDateTime; +import java.time.ZoneOffset; +import java.util.Date; +import java.util.Map; import java.util.function.Function; +import static com.azure.core.amqp.AmqpMessageConstant.ENQUEUED_TIME_UTC_ANNOTATION_NAME; + /** * Contains convenience methods to instrument specific calls with traces and metrics. */ public final class ServiceBusReceiverInstrumentation { + private static final Symbol ENQUEUED_TIME_SYMBOL = Symbol.getSymbol(ENQUEUED_TIME_UTC_ANNOTATION_NAME.getValue()); private final ServiceBusMeter meter; private final ServiceBusTracer tracer; private final ReceiverKind receiverKind; @@ -67,37 +76,44 @@ public boolean isAsyncReceiverInstrumentation() { * for sync receiver. * Reports consumer lag metric. */ - public Context startProcessInstrumentation(String name, ServiceBusReceivedMessage message, Context parent) { - if (message == null || (!tracer.isEnabled() && !meter.isConsumerLagEnabled())) { + public Context startProcessInstrumentation(String name, Map applicationProperties, + OffsetDateTime enqueuedTime, Context parent) { + if (applicationProperties == null || (!tracer.isEnabled() && !meter.isConsumerLagEnabled())) { return parent; } - Context span = (tracer.isEnabled() && receiverKind != ReceiverKind.SYNC_RECEIVER) ? tracer.startProcessSpan(name, message, parent) : parent; + Context span = (tracer.isEnabled() && receiverKind != ReceiverKind.SYNC_RECEIVER) + ? tracer.startProcessSpan(name, applicationProperties, enqueuedTime, parent) + : parent; // important to record metric after span is started - meter.reportConsumerLag(message.getEnqueuedTime(), span); + meter.reportConsumerLag(enqueuedTime, span); return span; } public void instrumentProcess(ServiceBusReceivedMessage message, ReceiverKind caller, Function handleMessage) { - if (receiverKind != caller) { + if (receiverKind != caller || message == null) { handleMessage.apply(message); return; } - Context span = startProcessInstrumentation("ServiceBus.process", message, Context.NONE); + Context span = startProcessInstrumentation("ServiceBus.process", message.getApplicationProperties(), + message.getEnqueuedTime(), Context.NONE); ContextAccessor.setContext(message, span); - AutoCloseable scope = tracer.makeSpanCurrent(span); - Throwable error = null; - try { - error = handleMessage.apply(message); - } catch (Throwable t) { - error = t; - throw t; - } finally { - tracer.endSpan(error, span, scope); + wrap(span, message, handleMessage); + } + + public void instrumentProcess(Message message, ReceiverKind caller, Function handleMessage) { + if (receiverKind != caller || message == null || message.getApplicationProperties() == null) { + handleMessage.apply(message); + return; } + + Context span = startProcessInstrumentation("ServiceBus.process", message.getApplicationProperties().getValue(), + getEnqueuedTime(message), Context.NONE); + //ContextAccessor.setContext(message, span); + wrap(span, message, handleMessage); } /** @@ -145,4 +161,30 @@ private static String getSettlementSpanName(DispositionStatus status) { return "ServiceBus.unknown"; } } + + private void wrap(Context span, T message, Function handleMessage) { + AutoCloseable scope = tracer.makeSpanCurrent(span); + Throwable error = null; + try { + error = handleMessage.apply(message); + } catch (Throwable t) { + error = t; + throw t; + } finally { + tracer.endSpan(error, span, scope); + } + } + + private OffsetDateTime getEnqueuedTime(Message message) { + if (message.getMessageAnnotations() == null || message.getMessageAnnotations().getValue() == null) { + return null; + } + + Object date = message.getMessageAnnotations().getValue().get(ENQUEUED_TIME_SYMBOL); + if (date instanceof Date) { + return ((Date) date).toInstant().atOffset(ZoneOffset.UTC); + } + + return null; + } } diff --git a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/instrumentation/ServiceBusTracer.java b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/instrumentation/ServiceBusTracer.java index 4f24630ec75f..532a71a249f0 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/instrumentation/ServiceBusTracer.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/instrumentation/ServiceBusTracer.java @@ -285,13 +285,13 @@ Context startSpanWithLink(String spanName, OperationName operationName, ServiceB /** * Starts span. Used by ServiceBus*Instrumentations. */ - Context startProcessSpan(String spanName, ServiceBusReceivedMessage message, Context parent) { - if (isEnabled() && message != null) { + Context startProcessSpan(String spanName, Map applicationProperties, OffsetDateTime enqueuedTime, Context parent) { + if (isEnabled() && applicationProperties != null) { StartSpanOptions startOptions = createStartOption(SpanKind.CONSUMER, OperationName.PROCESS) - .setRemoteParent(extractContext(message.getApplicationProperties())); + .setRemoteParent(extractContext(applicationProperties)); - if (message.getEnqueuedTime() != null) { - startOptions.setAttribute(MESSAGE_ENQUEUED_TIME_ATTRIBUTE_NAME, message.getEnqueuedTime().toEpochSecond()); + if (enqueuedTime != null) { + startOptions.setAttribute(MESSAGE_ENQUEUED_TIME_ATTRIBUTE_NAME, enqueuedTime.toEpochSecond()); } return tracer.start(spanName, startOptions, parent); diff --git a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/TracingIntegrationTests.java b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/TracingIntegrationTests.java index 7a50929d25f9..6bc5730921d2 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/TracingIntegrationTests.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/TracingIntegrationTests.java @@ -592,6 +592,62 @@ public void sendAndProcessNoAutoComplete() throws InterruptedException { assertSettledVsProcessed(completed, processed, 1); } + @Test + public void sendAndProcessSessionNoAutoComplete() throws InterruptedException { + String sessionQueueName = getSessionQueueName(0); + String sessionId = UUID.randomUUID().toString(); + ServiceBusMessage message = new ServiceBusMessage(CONTENTS_BYTES).setSessionId(sessionId); + + OpenTelemetry otel = configureOTel(getFullyQualifiedDomainName(), sessionQueueName); + clientOptions = new ClientOptions().setTracingOptions(new OpenTelemetryTracingOptions().setOpenTelemetry(otel)); + sender = toClose(new ServiceBusClientBuilder() + .connectionString(getConnectionString()) + .clientOptions(clientOptions) + .sender() + .queueName(sessionQueueName) + .buildAsyncClient()); + + StepVerifier.create(sender.sendMessage(message)).expectComplete().verify(TIMEOUT); + + CountDownLatch processFound = new CountDownLatch(2); + spanProcessor.notifyIfCondition(processFound, span -> span.getName().equals("ServiceBus.process") || span.getName().equals("ServiceBus.complete")); + + AtomicReference currentInProcess = new AtomicReference<>(); + AtomicReference receivedMessage = new AtomicReference<>(); + processor = toClose(new ServiceBusClientBuilder() + .connectionString(getConnectionString()) + .clientOptions(clientOptions) + .sessionProcessor() + .maxConcurrentSessions(1) + .queueName(sessionQueueName) + .disableAutoComplete() + .processMessage(mc -> { + currentInProcess.compareAndSet(null, Span.current()); + receivedMessage.compareAndSet(null, mc.getMessage()); + mc.complete(); + }) + .processError(e -> fail("unexpected error", e.getException())) + .buildProcessorClient()); + + toClose((AutoCloseable) () -> processor.stop()); + processor.start(); + assertTrue(processFound.await(20, TimeUnit.SECONDS)); + processor.stop(); + + assertTrue(currentInProcess.get().getSpanContext().isValid()); + List spans = spanProcessor.getEndedSpans(); + + List processed = findSpans(spans, "ServiceBus.process") + .stream().filter(p -> p.equals(currentInProcess.get())).collect(Collectors.toList()); + assertEquals(1, processed.size()); + assertConsumerSpan(processed.get(0), receivedMessage.get(), "ServiceBus.process"); + + List completed = findSpans(spans, "ServiceBus.complete"); + assertEquals(1, completed.size()); + assertClientSpan(completed.get(0), Collections.singletonList(receivedMessage.get()), "ServiceBus.complete", "settle"); + assertSettledVsProcessed(completed, processed, 1); + } + @Test public void sendAndProcessParallel() throws InterruptedException { int messageCount = 10; diff --git a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/implementation/instrumentation/ServiceBusReceiverInstrumentationTests.java b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/implementation/instrumentation/ServiceBusReceiverInstrumentationTests.java index 46597b57f963..55766945d4e4 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/implementation/instrumentation/ServiceBusReceiverInstrumentationTests.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/implementation/instrumentation/ServiceBusReceiverInstrumentationTests.java @@ -35,7 +35,7 @@ public void testInstrumentNullMessageNoMeter() { ServiceBusReceiverInstrumentation instrumentation = new ServiceBusReceiverInstrumentation(tracer, null, "fqdn", "entityPath", null, ReceiverKind.ASYNC_RECEIVER); - instrumentation.startProcessInstrumentation("span name", null, Context.NONE); + instrumentation.startProcessInstrumentation("span name", null, null, Context.NONE); instrumentation.instrumentSettlement(Mono.just(1), null, Context.NONE, DispositionStatus.ABANDONED); verify(tracer, never()).start(anyString(), any(StartSpanOptions.class), any(Context.class)); } @@ -48,7 +48,7 @@ public void testInstrumentNullMessageNoTracer() { "fqdn", "entityPath", null, ReceiverKind.ASYNC_RECEIVER); // does not throw - instrumentation.startProcessInstrumentation("span name", null, Context.NONE); + instrumentation.startProcessInstrumentation("span name", null, null, Context.NONE); instrumentation.instrumentSettlement(Mono.just(1), null, Context.NONE, DispositionStatus.ABANDONED); } @@ -58,7 +58,7 @@ public void testInstrumentNullMessageDisabled() { "fqdn", "entityPath", null, ReceiverKind.ASYNC_RECEIVER); // does not throw - instrumentation.startProcessInstrumentation("span name", null, Context.NONE); + instrumentation.startProcessInstrumentation("span name", null, null, Context.NONE); instrumentation.instrumentSettlement(Mono.just(1), null, Context.NONE, DispositionStatus.ABANDONED); } }