@Override
    protected void doStart() throws Exception {
        super.doStart();

        LOG.debug("Creating connection to Azure ServiceBus");
        client = getEndpoint().getServiceBusClientFactory().createServiceBusProcessorClient(getConfiguration(),
                this::processMessage, this::processError);
        client.start();
    }
@Test
    void consumerPropagatesApplicationPropertiesToMessageHeaders() throws Exception {
        try (ServiceBusConsumer consumer = new ServiceBusConsumer(endpoint, processor)) {
            consumer.doStart();
            verify(client).start();
            verify(clientFactory).createServiceBusProcessorClient(any(), any(), any());

            when(messageContext.getMessage()).thenReturn(message);
            configureMockMessage();
            message.getApplicationProperties().put(PROPAGATED_HEADER_KEY, PROPAGATED_HEADER_VALUE);
            when(headerFilterStrategy.applyFilterToExternalHeaders(anyString(), any(), any())).thenReturn(false);

            processMessageCaptor.getValue().accept(messageContext);

            verify(headerFilterStrategy, atLeastOnce()).applyFilterToExternalHeaders(anyString(), any(), any(Exchange.class));
            verifyNoMoreInteractions(headerFilterStrategy);

            verify(processor).process(any(Exchange.class), any(AsyncCallback.class));
            Exchange exchange = exchangeCaptor.getValue();
            assertThat(exchange).isNotNull();

            Message inMessage = exchange.getIn();
            assertThat(inMessage).isNotNull();
            assertThat(inMessage.getHeaders()).containsEntry(PROPAGATED_HEADER_KEY, PROPAGATED_HEADER_VALUE);
        }
    }