The Bulletproof Backbone: Implementing Azure Service Bus for Zero-Loss Banking Transfers

In my previous article, we discussed why Azure Service Bus is the best option for our banking funds transfer functionality. We also discussed the product definitions and architectural perspectives for our use case. In this article we cover how to implement the Azure Messaging layer in our application (You can find the code and architecture diagrams (activity diagram, state diagram, sequence diagram) in my Github.

Understanding our Application Use Case.

Remember our API incorporates two main services : Funds Transfer and Notification Service. When User A sends funds to User B, the TransactionService performs respective credit and debit operations on users’ accounts. Thereafter, we send messages (sms, push, email) to both users notifying them of the respective transaction. We cover this in 5 steps: configuring Azure Service Bus, processing the funds, sending notifications to Azure Service Bus queue, processing the messages in Azure Service Bus and confirming the operations.

1. Configuring Azure Service Bus

This ServiceBusConfig class is a central configuration hub within your Spring Boot application, dedicated to setting up the necessary clients for interacting with Azure Service Bus. The serviceBusSenderAsyncClient() bean creates and configures an instance of ServiceBusSenderAsyncClient. This client is specifically designed for asynchronously sending messages to your primary Azure Service Bus queue transaction-notifications-queue.

 @Bean
    public ServiceBusSenderAsyncClient serviceBusSenderAsyncClient() {
        return new ServiceBusClientBuilder()
                .connectionString(connectionString)
                .sender()
                .queueName(queueName)
                .buildAsyncClient();
    }

The serviceBusReceiverAsyncClient() bean provides an ServiceBusReceiverAsyncClient for your primary queue transaction-notifications-queue for manual message reception, where your application explicitly controls when a message is "settled" (completed, abandoned, or dead-lettered).

@Bean
    public ServiceBusReceiverAsyncClient serviceBusReceiverAsyncClient() {
        return new ServiceBusClientBuilder()
                .connectionString(connectionString)
                .receiver()
                .queueName(queueName)
                .disableAutoComplete() // Disables automatic message completion/abandonment
                .buildAsyncClient();
    }

Then, serviceBusProcessorClientBuilder() bean provides a builder for ServiceBusProcessorClient. A ServiceBusProcessorClient is a higher-level abstraction for push-based message consumption. It handles the complexities of concurrent message processing, automatic message settlement (unless configured otherwise), and error handling through predefined handlers.

@Bean
    public ServiceBusProcessorClientBuilder serviceBusProcessorClientBuilder() {
        return new ServiceBusClientBuilder()
                .connectionString(connectionString)
                .processor()
                .queueName(queueName);
    }

The failedNotificationSenderAsyncClient() bean provides a separate ServiceBusSenderAsyncClient specifically configured to send messages to a dead-letter queue (failedTransactionQueueName). When a message fails to be processed or sent successfully to its primary destination (even after retries), it's crucial not to simply discard it. Instead, it's redirected to this dead-letter queue. This ensures:

• No Silent Data Loss: Every message is accounted for.

• Forensic Analysis: Messages in the dead-letter queue can be inspected to understand why they failed, debug issues, and potentially reprocess them. This is indispensable for financial reconciliation and compliance.

• Isolation of Concerns: Using a dedicated client and queue for failed messages prevents issues in the dead-lettering mechanism from impacting the primary message flow.

2. Processing Funds Transfer : Transaction Service

When a user enters the transaction details his then calls the TransactionService first validates user existence in the database and whether the action is a self-transfer operation. We then validate the transaction operation; whether the sender has enough funds in their account and whether they are within their daily transaction limit. If successful, we debit and credit the sender’s and receiver’s account respectively with the custom transaction created by the createTransaction() method which include transaction details such as TransactionId, TransactionType, Description, Amount, Timestamp and respective balances after the transaction and persist this to a database.

@Transactional
    public void transferFunds(TransactionRequestDTO request) {
        log.info("Attempting fund transfer from user {} to user {} for amount {}",
                request.getFromUserId(), request.getToUserId(), request.getAmount());
        //Validate whether the users exists
        Account fromAccount = accountRepository.findByCustomerId(request.getFromUserId())
                .orElseThrow(() -> new IllegalArgumentException("Sender account not found for user ID: " + request.getFromUserId()));
        Account toAccount = accountRepository.findByCustomerId(request.getToUserId())
                .orElseThrow(() -> new IllegalArgumentException("Recipient account not found for user ID: " + request.getToUserId()));

        // Self-transfer check
        if (fromAccount.getCustomerId().equals(toAccount.getCustomerId())) {
            throw new IllegalArgumentException("Cannot transfer funds to the same account.");
        }
        // Validate if the sender has sufficient balance and has not exceeded transfer limits
        try {
            validateTransaction(fromAccount, request.getAmount());
        } catch (InsufficientBalanceException | LimitExceededException e) {
            log.warn("Transaction validation failed for user {}: {}", fromAccount.getCustomerId(), e.getMessage());
            throw e;
        }
        // Generate a custom Transaction ID using a custom TransactionIdGenerator 
        String transactionId = TransactionIdGenerator.generate();
        log.debug("Generated transaction ID: {}", transactionId);

        // Debit sender account
        fromAccount.setBalance(fromAccount.getBalance().subtract(request.getAmount()));
        fromAccount.setDailyTransactionAmount(
                fromAccount.getDailyTransactionAmount().add(request.getAmount())
        );
        fromAccount.setDailyTransactionLimit(fromAccount.getDailyTransactionLimit().subtract(request.getAmount()));

        Transaction debitTransaction = createTransaction(transactionId,
                TransactionType.TRANSFER_OUT, request.getAmount().negate(),
                String.format("Transfer to %s (%s)", toAccount.getCustomerName(), toAccount.getCustomerId()),
                LocalDateTime.now(), fromAccount.getBalance());

        fromAccount.addTransaction(debitTransaction);
        accountRepository.save(fromAccount);
        //Credit the receiver's account
        toAccount.setBalance(toAccount.getBalance().add(request.getAmount()));
        Transaction creditTransaction = createTransaction(
                transactionId, TransactionType.TRANSFER_IN, request.getAmount(),
                String.format("Transfer from %s (%s)", fromAccount.getCustomerName(), fromAccount.getCustomerId()),
                LocalDateTime.now(),
                toAccount.getBalance()
        );
        toAccount.addTransaction(creditTransaction);
        accountRepository.save(toAccount);

        log.info("Funds transferred successfully for transaction ID: {}", transactionId);
        try {
            // we initiate the notification functionality to send notifications to both users
            notificationService.sendTransferNotifications(transactionId, fromAccount, toAccount, request.getAmount());
            log.debug("Notification enqueued for transaction ID: {}", transactionId);
        } catch (Exception e) {
            log.error("Failed to enqueue notification for transaction ID: {}. This will NOT rollback the financial transaction.", transactionId, e);
            // We could consider alternative notification methods or a monitoring alert here.
        }
    }

3. Creating and Sending Notifications to Azure Service Bus : Notification Service

The Notification Service’s functionality is to create the two notifications for the sender and receiver and send these notifications to the Azure Service Queue for delivery via the sendTransferNotifications() method. This method then calls the sendNotificationToQueue() which asynchronously sends messages to Azure Service Bus queue via ServiceBusSenderAsyncClient.

@Override
    @Async
    public void sendTransferNotifications(String transactionId, Account sender, Account recipient, BigDecimal amount) {
        log.info("Asynchronously preparing and sending transfer notifications for transaction ID: {}", transactionId);
        // Format message with custom message formatter
        String senderMessage = messageFormatter.formatSenderMessage(
                transactionId, amount, recipient.getCustomerName(), sender.getBalance(), LocalDateTime.now()
        );
        // Create a notification for the sender
        TransactionNotification senderNotification = createNotification(
                transactionId, sender.getCustomerId(), senderMessage, LocalDateTime.now(),
                TransactionType.TRANSFER_OUT.name(),
                amount, recipient.getCustomerName(), sender.getCustomerName()
        );
        //Send notification to Azure Service Bus queue
        sendNotificationToQueue(senderNotification);
        // Format message for the recipient
        String recipientMessage = messageFormatter.formatRecipientMessage(
                transactionId, amount, sender.getCustomerName(), recipient.getBalance(), LocalDateTime.now()
        );
        // Create recipient's notification
        TransactionNotification recipientNotification = createNotification(
                transactionId, recipient.getCustomerId(), recipientMessage, LocalDateTime.now(),
                TransactionType.TRANSFER_IN.name(),
                amount, recipient.getCustomerName(), sender.getCustomerName()
        );
        // Send notification to Azure Service Bus queue
        sendNotificationToQueue(recipientNotification);
    }

The ServiceBusSenderAsyncClient.sendMessage() is very critical for our use case. It asynchronously sends notification messages to an Azure Service Bus queue and, crucially, handles potential failures gracefully by redirecting unsent messages to a dedicated dead-letter queue. This pattern is vital for ensuring message delivery guarantees and auditability.

/**
Asynchronously sends a TransactionNotification to the primary Azure Service Bus queue. 
This method leverages the non-blocking ServiceBusSenderAsyncClient and includes a robust
error handling mechanism with a fallback to a dedicated dead-letter queue
*/
private void sendNotificationToQueue(TransactionNotification notification) {
        if (serviceBusSenderAsyncClient == null) {
            log.error("CRITICAL ERROR: ServiceBusSenderClient is not initialized. Cannot send notification for transaction ID: {}", notification.getTransactionId());
            throw new IllegalStateException("ServiceBusSenderClient is not initialized. Application is misconfigured or in an invalid state.");
        }

        try {
            String jsonNotification = objectMapper.writeValueAsString(notification);
            ServiceBusMessage message = new ServiceBusMessage(jsonNotification);
            message.setCorrelationId(notification.getTransactionId());

            serviceBusSenderAsyncClient.sendMessage(message)
                            .doOnSuccess(aVoid -> {
                                log.info("Sent notification to Service Bus queue '{}' for transaction ID: {} message ID: {}", queueName, notification.getTransactionId(), message.getMessageId());
                            })
                             .onErrorResume(error -> {
                                 log.error("CRITICAL ERROR: Failed to send message to Azure Service Bus queue '{}' for transaction ID: {} AFTER ALL RETRIES. Notification data: {}",
                                         queueName, notification.getTransactionId(), notification, error);

                                 return Mono.defer(() ->{
                                     try{
                                         ServiceBusMessage deadLetterMessage = new ServiceBusMessage(jsonNotification);
                                         deadLetterMessage.setCorrelationId(notification.getTransactionId());
                                         deadLetterMessage.getApplicationProperties().put("failureReason", error.getMessage());

                                         return failedNotificationSenderAsyncClient.sendMessage(deadLetterMessage)
                                                 .doOnSuccess(inform -> log.info("Moved failed notification for transaction ID: {} to failed-notifications-queue", notification.getTransactionId()))
                                                 .doOnError(deadLetterQueueError -> log.error("CRITICAL ERROR:  Failed to send message to DEAD LETTER QUEUE for transaction ID: {}. Data lost: {}", notification.getTransactionId(), notification, deadLetterQueueError))
                                                 .then();


                                     } catch (IllegalStateException deadLetterQueueError){
                                         log.error("CRITICAL ERROR: Could not send failed notification to DEAD LETTER QUEUE for transaction ID: {}. Data lost: {}", notification.getTransactionId(), notification, deadLetterQueueError);
                                         return Mono.empty();
                                     }
                                 });
                             })
                             .subscribe();
        } catch (JsonProcessingException e) {
            log.error("ERROR: Failed to serialize TransactionNotification to JSON for transaction ID: {}. Notification data: {}",
                    notification.getTransactionId(), notification, e);
            throw new NotificationSerializationException("Failed to serialize notification for transaction ID: " + notification.getTransactionId(), e);
        }
    }

The Async suffix is key. It implies that this client is designed for non-blocking I/O operations. It leverages underlying asynchronous mechanisms (like Netty or other NIO frameworks) to send messages without consuming a thread per request. This prevents thread exhaustion and improves scalability. The ServiceBusMessage object, encapsulates the payload and metadata of the message to be sent. The SDK handles the serialization, network communication, and interaction with the Azure Service Bus endpoint. The Mono<Void> it returns will complete when the message is successfully acknowledged by the Service Bus, or it will emit an error if the sending fails for example due to network issues, Service Bus unavailability, transient errors. This reliably gets the transaction notification into the messaging system while its asynchronous nature fundamentally maintains the performance of your core banking transaction processing.

Dead Letter Pattern

Note that we implement Dead Letter logic with Mono.defer() where the failedNotificationSenderAsyncClient instance is configured as the core of the Dead Letter Pattern to send to the dead-letter queue (failed-transactions-queue). Here, using a separate client is good practice to isolate concerns. Thus, instead of losing the message, it's moved to a holding area for manual inspection, reprocessing, or further automated handling which reduces data loss and provides a recovery mechanism. We then chain a fallback doOnError which ensures that if sending to the dead-letter queue fails, we gain transparency. That is, If data is lost, the system explicitly logs it, allowing for investigation and reconciliation ensuring financial data integrity.

Including onErrorResume embodies the principle of resilience where we ensure that transient or even persistent failures in the primary message sending path do not lead to lost data or application crashes. It implements a robust retry and dead-lettering strategy, which is non-negotiable for financial systems. Therefore, with this integration we benefit from a few responsible approaches required for a financial system:

1. High Throughput & Responsiveness: By using ServiceBusSenderAsyncClient and Reactor's Mono, the application thread that initiates the notification send is not blocked. This allows the core banking system to process the next transaction immediately, leading to higher throughput and a more responsive user experience (e.g., instant transaction confirmation).

2. Guaranteed Delivery (or Audited Failure): The onErrorResume with the dead-letter queue is a critical pattern for "at-least-once" delivery. While it doesn't guarantee the message will reach its final destination on the first try, it guarantees that if it fails, it will be moved to a safe, auditable location (the dead-letter queue). This means:

• No Silent Data Loss: Every notification attempt is either successful or explicitly recorded as a failure in the dead-letter queue. This is paramount for financial compliance and reconciliation.

• Recovery Mechanism: Messages in the dead-letter queue can be manually inspected, debugged, and potentially reprocessed once the underlying issue is resolved.

3. Robust Error Handling: The layered error handling (primary onErrorResume for send failures, and doOnError for dead-letter send failures, plus try-catch for synchronous preparation errors) ensures that almost all failure scenarios are anticipated and logged, preventing application crashes and providing clear diagnostics.

4. Observability & Auditability: Extensive log.info and log.error statements, combined with correlation IDs and failure reasons in the dead-letter message properties, provide a rich set of data for monitoring, auditing, and troubleshooting. You can easily see successful sends, and quickly identify and investigate failures.

5. Decoupling: The use of Azure Service Bus decouples the core banking transaction processing from the notification sending logic. If the notification service is temporarily down or slow, it doesn't impact the ability to process transactions.

6. Handling Incoming Messages (NotificationProcessor)

In order to effectively handle messages, we must first listen to all incoming messages in the Azure Service Bus queue. The NotificationProcessor class handles these operations. First, we configure and initialize error handlers and listen to incoming messages with the startListening() method which initializes the AzureServiceBusProcessorClient for this purpose. With handleMessage() initialized by the AzureServiceBusProcessorClient for each incoming message, we deserialize the incoming message to a proper TransactionNotificationwhich is then processed and send to the users. The processNotification() is initiated to purposefully handle these messages for example, send the message and SMS, push notification, email or any other handling. For the purposes of simplicity of showing how we integrate Azure Service Bus as our message service, we have logged the message. However, we can expand this functionality in actual applications as we wish.

Finally, after processing the message, the stopListening() method is invoked by Spring when the application context is shut down to ensure that we release resources and stop listening for the messages which guarantees message loss prevention and/or incomplete message handling during shutdown.

/**
Handles a received message from Azure Service Bus. This method is invoked by the
Service Bus Processor Client for each message received. It deserializes the message body 
into a TransactionNotification object, processes it, and then completes the message on 
the Service Bus. If deserialization fails, the message is dead-lettered. If any other error 
occurs during processing, the message is abandoned, making it available for re-delivery.
*/
public void handleMessage(ServiceBusReceivedMessageContext context) {
        ServiceBusReceivedMessage message = context.getMessage();
        try {
            String messageBody = message.getBody().toString();
            log.debug("Received message from Service Bus (Sequence #{}): {}", message.getSequenceNumber(), messageBody);

            TransactionNotification notification = objectMapper.readValue(messageBody, TransactionNotification.class);
            processNotification(notification);

            context.complete();
            log.info("Successfully processed and completed message for transaction ID: {}", notification.getTransactionId());

        } catch (JsonProcessingException e) {
            log.error("Failed to deserialize message body to TransactionNotification. Message will be dead-lettered. Message body: {}", message.getBody().toString(), e);
            context.deadLetter();
        } catch (Exception e) {
            log.error("Error processing message from Service Bus. Message will be abandoned. Message body: {}. Sequence #{}", message.getBody().toString(), message.getSequenceNumber(), e);
            context.abandon();
        }
    }

/**
Processes the received TransactionNotification.
This method contains the core business logic for handling a transaction notification.
In a real application, this would involve more complex operations like updating a database,
sending emails, or triggering other downstream services.
*/

private void processNotification(TransactionNotification notification) {
        log.info("Starting processing for notification: {}, Notification details: Transaction ID: {}, User ID: {}, Message: {}",
                notification.getTransactionId(), notification.getTransactionId(), notification.getUserId(), notification.getMessage());

        // System.out.println for immediate console output, typically replaced with a more robust notification system
        System.out.println("----**NEW NOTIFICATION**----");
        System.out.println("Transaction ID: " + notification.getTransactionId());
        System.out.println("User ID: " + notification.getUserId());
        System.out.println("Message: " + notification.getMessage());
        System.out.println("Timestamp: " + notification.getTimestamp());
        System.out.println("Transaction Type: " + notification.getTransactionType());
        System.out.println("Amount: " + notification.getAmount());
        if (notification.getRecipientName() != null) {
            System.out.println("Recipient Name: " + notification.getRecipientName());
        }
        if (notification.getSenderName() != null) {
            System.out.println("Sender Name: " + notification.getSenderName());
        }
        System.out.println("-------------------------");
    }

5. The result …

We have successfully sent the funds and both users receive the notifications.

and we can confirm the operations on Azure portal.

Happy coding 😊.