@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 synchronizationDoesNotAbandonMessageWhenReceiveModeIsReceiveAndDelete() throws Exception {
        try (ServiceBusConsumer consumer = new ServiceBusConsumer(endpoint, processor)) {
            when(configuration.getServiceBusReceiveMode()).thenReturn(ServiceBusReceiveMode.RECEIVE_AND_DELETE);
            consumer.doStart();
            verify(client).start();
            verify(clientFactory).createServiceBusProcessorClient(any(), any(), any());

            when(messageContext.getMessage()).thenReturn(message);

            processMessageCaptor.getValue().accept(messageContext);

            verify(messageContext).getMessage();

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

            Synchronization synchronization = exchange.getExchangeExtension().handoverCompletions().get(0);
            synchronization.onFailure(exchange);

            verifyNoMoreInteractions(messageContext);
        }
    }