@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 consumerSubmitsExchangeToProcessor() 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();

            processMessageCaptor.getValue().accept(messageContext);

            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.getBody()).isInstanceOf(BinaryData.class);
            assertThat(inMessage.getBody(BinaryData.class).toString()).isEqualTo(MESSAGE_BODY);
            assertThat(inMessage.getHeaders()).isEqualTo(createExpectedMessageHeaders());
        }
    }