AbstractVertxBasedMqttProtocolAdapter extends AbstractProtocolAdapterBase<T> { void handlePublishedMessage(final MqttPublishMessage message, final MqttEndpoint endpoint, final Device authenticatedDevice, final OptionalInt traceSamplingPriority) { final SpanContext spanContext = message.topicName() != null ? Optional.ofNullable(PropertyBag.fromTopic(message.topicName())) .map(propertyBag -> TracingHelper.extractSpanContext(tracer, propertyBag::getPropertiesIterator)) .orElse(null) : null; final Span span = newChildSpan(spanContext, "PUBLISH", endpoint, authenticatedDevice); span.setTag(Tags.MESSAGE_BUS_DESTINATION.getKey(), message.topicName()); span.setTag(TracingHelper.TAG_QOS.getKey(), message.qosLevel().toString()); traceSamplingPriority.ifPresent(prio -> TracingHelper.setTraceSamplingPriority(span, prio)); final MqttContext context = MqttContext.fromPublishPacket(message, endpoint, span, authenticatedDevice); context.setTimer(getMetrics().startTimer()); final Future<Void> spanPreparationFuture = authenticatedDevice == null ? applyTraceSamplingPriorityForTopicTenant(context.topic(), span) : Future.succeededFuture(); spanPreparationFuture .compose(v -> checkTopic(context)) .compose(ok -> onPublishedMessage(context)) .onComplete(processing -> { if (processing.succeeded()) { Tags.HTTP_STATUS.set(span, HttpURLConnection.HTTP_ACCEPTED); onMessageSent(context); } else { if (processing.cause() instanceof ServiceInvocationException) { final ServiceInvocationException sie = (ServiceInvocationException) processing.cause(); Tags.HTTP_STATUS.set(span, sie.getErrorCode()); } else { Tags.HTTP_STATUS.set(span, HttpURLConnection.HTTP_INTERNAL_ERROR); } if (processing.cause() instanceof ClientErrorException) { } else { onMessageUndeliverable(context); } TracingHelper.logError(span, processing.cause()); if (context.deviceEndpoint().isConnected()) { span.log("closing connection to device"); context.deviceEndpoint().close(); } } span.finish(); }); }  final void setAuthHandler(final AuthHandler<MqttConnectContext> authHandler); @Override int getPortDefaultValue(); @Override int getInsecurePortDefaultValue(); @Autowired final void setMetrics(final MqttAdapterMetrics metrics); void setMqttSecureServer(final MqttServer server); void setMqttInsecureServer(final MqttServer server); final Future<Void> uploadMessage(
            final MqttContext ctx,
            final ResourceIdentifier resource,
            final MqttPublishMessage message); final Future<Void> uploadTelemetryMessage(
            final MqttContext ctx,
            final String tenant,
            final String deviceId,
            final Buffer payload); final Future<Void> uploadEventMessage(
            final MqttContext ctx,
            final String tenant,
            final String deviceId,
            final Buffer payload); final Future<Void> uploadCommandResponseMessage(
            final MqttContext ctx,
            final ResourceIdentifier targetAddress);  }
@Test public void testHandlePublishedMessageFailsForMalformedTopic() { final MqttServer server = getMqttServer(false); final AbstractVertxBasedMqttProtocolAdapter<MqttProtocolAdapterProperties> adapter = getAdapter(server); final MqttEndpoint endpoint = mockEndpoint(); final Buffer payload = Buffer.buffer("hello"); final MqttPublishMessage msg = mock(MqttPublishMessage.class); when(msg.qosLevel()).thenReturn(MqttQoS.AT_LEAST_ONCE); when(msg.topicName()).thenReturn(null); when(msg.payload()).thenReturn(payload); adapter.handlePublishedMessage(msg, endpoint, null, OptionalInt.empty()); verify(metrics, never()).reportTelemetry( any(MetricsTags.EndpointType.class), anyString(), any(), eq(MetricsTags.ProcessingOutcome.FORWARDED), any(MetricsTags.QoS.class), anyInt(), any()); verify(endpoint, never()).publishAcknowledge(anyInt()); }