TelemetrySenderImpl extends AbstractDownstreamSender { @Override protected Future<ProtonDelivery> sendMessage(final Message message, final Span currentSpan) { Objects.requireNonNull(message); Objects.requireNonNull(currentSpan); final String messageId = String.format("%s-%d", getClass().getSimpleName(), MESSAGE_COUNTER.getAndIncrement()); message.setMessageId(messageId); logMessageIdAndSenderInfo(currentSpan, messageId); final SendMessageSampler.Sample sample = this.sampler.start(this.tenantId); final AtomicReference<ProtonDelivery> deliveryRef = new AtomicReference<>(); final ClientConfigProperties config = connection.getConfig(); final AtomicBoolean timeoutReached = new AtomicBoolean(false); final Long timerId = config.getSendMessageTimeout() > 0 ? connection.getVertx().setTimer(config.getSendMessageTimeout(), id -> { if (timeoutReached.compareAndSet(false, true)) { final ServerErrorException exception = new SendMessageTimeoutException( "waiting for delivery update timed out after " + config.getSendMessageTimeout() + "ms"); logMessageSendingError( "waiting for delivery update timed out for message [ID: {}, address: {}] after {}ms", messageId, getMessageAddress(message), connection.getConfig().getSendMessageTimeout()); Optional.ofNullable(deliveryRef.get()) .ifPresent(delivery -> ProtonHelper.released(delivery, true)); TracingHelper.logError(currentSpan, exception.getMessage()); Tags.HTTP_STATUS.set(currentSpan, HttpURLConnection.HTTP_UNAVAILABLE); currentSpan.finish(); sample.timeout(); } }) : null; final ProtonDelivery delivery = sender.send(message, deliveryUpdated -> { if (timerId != null) { connection.getVertx().cancelTimer(timerId); } final DeliveryState remoteState = deliveryUpdated.getRemoteState(); sample.completed(remoteState); if (timeoutReached.get()) { log.debug("ignoring received delivery update for message [ID: {}, address: {}]: waiting for the update has already timed out", messageId, getMessageAddress(message)); } else if (deliveryUpdated.remotelySettled()) { logUpdatedDeliveryState(currentSpan, message, deliveryUpdated); } else { logMessageSendingError("peer did not settle message [ID: {}, address: {}, remote state: {}], failing delivery", messageId, getMessageAddress(message), remoteState.getClass().getSimpleName()); TracingHelper.logError(currentSpan, new ServerErrorException( HttpURLConnection.HTTP_INTERNAL_ERROR, "peer did not settle message, failing delivery")); } currentSpan.finish(); }); deliveryRef.set(delivery); log.trace("sent message [ID: {}, address: {}], remaining credit: {}, queued messages: {}", messageId, getMessageAddress(message), sender.getCredit(), sender.getQueued()); return Future.succeededFuture(delivery); } protected  TelemetrySenderImpl(
            final HonoConnection con,
            final ProtonSender sender,
            final String tenantId,
            final String targetAddress,
            final SendMessageSampler sampler); @Override String getEndpoint(); static Future<DownstreamSender> create(
            final HonoConnection con,
            final String tenantId,
            final SendMessageSampler sampler,
            final Handler<String> remoteCloseHook); @Override Future<ProtonDelivery> sendAndWaitForOutcome(final Message rawMessage); @Override Future<ProtonDelivery> sendAndWaitForOutcome(final Message rawMessage, final SpanContext parent);  }
@Test public void testSendMessageFailsOnTimeout() { when(sender.send(any(Message.class), VertxMockSupport.anyHandler())).thenReturn(mock(ProtonDelivery.class)); when(vertx.setTimer(anyLong(), VertxMockSupport.anyHandler())).thenAnswer(invocation -> { final Handler<Long> handler = invocation.getArgument(1); final long timerId = 1; handler.handle(timerId); return timerId; }); final DownstreamSender messageSender = new TelemetrySenderImpl(connection, sender, "tenant", "telemetry/tenant", SendMessageSampler.noop()); final Message message = mock(Message.class); final Span span = mock(Span.class); ((TelemetrySenderImpl) messageSender).sendMessage(message, span); verify(span).finish(); }