Current version is 2.13.0, click here for the index

Azure Service Bus Queue Spring Cloud Stream Binder client library for Java

The project provides Spring Cloud Stream Binder for Azure Service Bus Queue which allows you to build message-driven microservice using Spring Cloud Stream based on Azure Service Bus Queue.

Source code | Package (Maven) | API reference documentation | Product documentation | Samples

Getting started

Prerequisites

Environment checklist

Include the package

[Add azure-spring-cloud-dependencies].
Add dependency. <version> can be skipped because we already add azure-spring-cloud-dependencies.

<dependency>
  <groupId>com.azure.spring</groupId>
  <artifactId>azure-spring-cloud-stream-binder-servicebus-queue</artifactId>
</dependency>

Key concepts

The Spring Cloud Stream Binder for Azure Service Bus Queue provides the binding implementation for the Spring Cloud Stream.

This implementation uses Spring Integration Service Bus Queue Channel Adapters at its foundation.

Scheduled Message

This binder supports submitting messages to a queue for delayed processing. Users can send scheduled messages with header x-delay expressing in milliseconds a delay time for the message. The message will be delivered to the respective queues after x-delay milliseconds.

Consumer Group

This binder has no consumer group support since all consumers share one queue.

Partitioning Support

This binder has no partition support even service bus queue supports partition.

Configuration Options

The binder provides the following configuration options:

Spring Cloud Azure Properties

|Name | Description | Required | Default |:---|:---|:---|:--- spring.cloud.azure.auto-create-resources | If enable auto-creation for Azure resources | | false spring.cloud.azure.region | Region name of the Azure resource group, e.g. westus | Yes if spring.cloud.azure.auto-create-resources is enabled. | spring.cloud.azure.environment | Azure Cloud name for Azure resources, supported values are azure, azurechina, azure_germany and azureusgovernment which are case insensitive | |azure | spring.cloud.azure.client-id | Client (application) id of a service principal or Managed Service Identity (MSI) | Yes if service principal or MSI is used as credential configuration. | spring.cloud.azure.client-secret | Client secret of a service principal | Yes if service principal is used as credential configuration. | spring.cloud.azure.msi-enabled | If enable MSI as credential configuration | Yes if MSI is used as credential configuration. | false spring.cloud.azure.resource-group | Name of Azure resource group | Yes if service principal or MSI is used as credential configuration. | spring.cloud.azure.subscription-id | Subscription id of an MSI | Yes if MSI is used as credential configuration. | spring.cloud.azure.tenant-id | Tenant id of a service principal | Yes if service principal is used as credential configuration. | spring.cloud.azure.servicebus.connection-string | Service Bus Namespace connection string | Yes if connection string is used as credential configuration | spring.cloud.azure.servicebus.namespace | Service Bus Namespace. Auto creating if missing | Yes if service principal or MSI is used as credential configuration. | spring.cloud.azure.servicebus.transportType | Service Bus transportType, supported value of AMQP and AMQP_WEB_SOCKETS | No | AMQP spring.cloud.azure.servicebus.retry-Options | Service Bus retry options | No | Default value of AmqpRetryOptions

Partition configuration

The system will obtain the parameter PartitionSupply to send the message.

The following are configuration items related to the producer:

partition-count

The number of target partitions for the data, if partitioning is enabled.

Default: 1

partition-key-extractor-name

The name of the bean that implements PartitionKeyExtractorStrategy. The partition handler will first use the PartitionKeyExtractorStrategy#extractKey method to obtain the partition key value.

Default: null

partition-key-expression

A SpEL expression that determines how to partition outbound data. When interface PartitionKeyExtractorStrategy is not implemented, it will be called in the method PartitionHandler#extractKey.

Default: null

For more information about setting partition for the producer properties, please refer to the Producer Properties of Spring Cloud Stream.

Service Bus Queue Producer Properties

It supports the following configurations with the format of spring.cloud.stream.servicebus.queue.bindings.<channelName>.producer.

sync

Whether the producer should act in a synchronous manner with respect to writing messages into a stream. If true, the producer will wait for a response after a send operation.

Default: false

send-timeout

Effective only if sync is set to true. The amount of time to wait for a response after a send operation, in milliseconds.

Default: 10000

Service Bus Queue Consumer Properties

It supports the following configurations with the format of spring.cloud.stream.servicebus.queue.bindings.<channelName>.consumer.

checkpoint-mode

The mode in which checkpoints are updated.

RECORD, checkpoints occur after each record successfully processed by user-defined message handler without any exception.

MANUAL, checkpoints occur on demand by the user via the Checkpointer. You can get Checkpointer by Message.getHeaders.get(AzureHeaders.CHECKPOINTER)callback.

Default: RECORD

prefetch-count

Prefetch count of underlying service bus client.

Default: 1

maxConcurrentCalls

Controls the max concurrent calls of service bus message handler and the size of fixed thread pool that handles user's business logic

Default: 1

maxConcurrentSessions

Controls the maximum number of concurrent sessions to process at any given time.

Default: 1

concurrency

When sessionsEnabled is true, controls the maximum number of concurrent sessions to process at any given time. When sessionsEnabled is false, controls the max concurrent calls of service bus message handler and the size of fixed thread pool that handles user's business logic.

Deprecated, replaced with maxConcurrentSessions when sessionsEnabled is true and maxConcurrentCalls when sessionsEnabled is false

Default: 1

sessionsEnabled

Controls if is a session aware consumer. Set it to true if is a queue with sessions enabled.

Default: false

requeueRejected

Controls if is a message that trigger any exception in consumer will be force to DLQ. Set it to true if a message that trigger any exception in consumer will be force to DLQ. Set it to false if a message that trigger any exception in consumer will be re-queued.

Default: false

receiveMode

The modes for receiving messages.

PEEK_LOCK, received message is not deleted from the queue or subscription, instead it is temporarily locked to the receiver, making it invisible to other receivers.

RECEIVE_AND_DELETE, received message is removed from the queue or subscription and immediately deleted.

Default: PEEK_LOCK

enableAutoComplete

Enable auto-complete and auto-abandon of received messages. 'enableAutoComplete' is not needed in for RECEIVEANDDELETE mode.

Default: false

Support for Service Bus Message Headers and Properties

The following table illustrates how Spring message headers are mapped to Service Bus message headers and properties. When creat a message, developers can specify the header or property of a Service Bus message by below constants.

For some Service Bus headers that can be mapped to multiple Spring header constants, the priority of different Spring headers is listed.

Service Bus Message Headers and Properties | Spring Message Header Constants | Type | Priority Number (Descending priority) ---|---|---|--- MessageId | com.azure.spring.integration.servicebus.converter.ServiceBusMessageHeaders.MESSAGEID | String | 1 MessageId | com.azure.spring.integration.core.AzureHeaders.RAWID | String | 2 MessageId | org.springframework.messaging.MessageHeaders.ID | UUID | 3 ContentType | org.springframework.messaging.MessageHeaders.CONTENTTYPE | String | N/A ReplyTo | org.springframework.messaging.MessageHeaders.REPLYCHANNEL | String | N/A ScheduledEnqueueTimeUtc | com.azure.spring.integration.servicebus.converter.ServiceBusMessageHeaders.SCHEDULEDENQUEUETIME | OffsetDateTime | 1 ScheduledEnqueueTimeUtc | com.azure.spring.integration.core.AzureHeaders.SCHEDULEDENQUEUEMESSAGE | Integer | 2 TimeToLive | com.azure.spring.integration.servicebus.converter.ServiceBusMessageHeaders.TIMETOLIVE | Duration | N/A SessionID | com.azure.spring.integration.servicebus.converter.ServiceBusMessageHeaders.SESSIONID | String | N/A CorrelationId | com.azure.spring.integration.servicebus.converter.ServiceBusMessageHeaders.CORRELATIONID | String | N/A To | com.azure.spring.integration.servicebus.converter.ServiceBusMessageHeaders.TO | String | N/A ReplyToSessionId | com.azure.spring.integration.servicebus.converter.ServiceBusMessageHeaders.REPLYTOSESSIONID | String | N/A PartitionKey | com.azure.spring.integration.servicebus.converter.ServiceBusMessageHeaders.PARTITIONKEY | String | 1 PartitionKey | com.azure.spring.integration.core.AzureHeaders.PARTITION_KEY | String | 2

Examples

Usage examples

Example: Manually set the partition key for the message

This example demonstrates how to manually set the partition key for the message in the application.

Way 1: This example requires that spring.cloud.stream.default.producer.partitionKeyExpression be set "'partitionKey-' + headers[<message-header-key>]".

spring:
  cloud:
    azure:
      servicebus:
        connection-string: [servicebus-namespace-connection-string]
    stream:
      default:
        producer:
          partitionKeyExpression:  "'partitionKey-' + headers[<message-header-key>]"

@PostMapping("/messages")
public ResponseEntity<String> sendMessage(@RequestParam String message) {
    LOGGER.info("Going to add message {} to Sinks.Many.", message);
    many.emitNext(MessageBuilder.withPayload(message)
                                .setHeader("<message-header-key>", "Customize partirion key")
                                .build(), Sinks.EmitFailureHandler.FAIL_FAST);
    return ResponseEntity.ok("Sent!");
}

NOTE: When using application.yml to configure the partition key, its priority will be the lowest. It will take effect only when the ServiceBusMessageHeaders.SESSION_ID, ServiceBusMessageHeaders.PARTITION_KEY, AzureHeaders.PARTITION_KEY are not configured.

Way 2: Manually add the partition Key in the message header by code.

Recommended: Use ServiceBusMessageHeaders.PARTITION_KEY as the key of the header.

@PostMapping("/messages")
public ResponseEntity<String> sendMessage(@RequestParam String message) {
    LOGGER.info("Going to add message {} to Sinks.Many.", message);
    many.emitNext(MessageBuilder.withPayload(message)
                                .setHeader(ServiceBusMessageHeaders.PARTITION_KEY, "Customize partirion key")
                                .build(), Sinks.EmitFailureHandler.FAIL_FAST);
    return ResponseEntity.ok("Sent!");
}

Not recommended but currently supported: AzureHeaders.PARTITION_KEY as the key of the header.

@PostMapping("/messages")
public ResponseEntity<String> sendMessage(@RequestParam String message) {
    LOGGER.info("Going to add message {} to Sinks.Many.", message);
    many.emitNext(MessageBuilder.withPayload(message)
                                .setHeader(AzureHeaders.PARTITION_KEY, "Customize partirion key")
                                .build(), Sinks.EmitFailureHandler.FAIL_FAST);
    return ResponseEntity.ok("Sent!");
}

NOTE: When both ServiceBusMessageHeaders.PARTITION_KEY and AzureHeaders.PARTITION_KEY are set in the message headers, ServiceBusMessageHeaders.PARTITION_KEY is preferred.

Example: Set the session id for the message

This example demonstrates how to manually set the session id of a message in the application.

@PostMapping("/messages")
public ResponseEntity<String> sendMessage(@RequestParam String message) {
    LOGGER.info("Going to add message {} to Sinks.Many.", message);
    many.emitNext(MessageBuilder.withPayload(message)
                                .setHeader(ServiceBusMessageHeaders.SESSION_ID, "Customize session id")
                                .build(), Sinks.EmitFailureHandler.FAIL_FAST);
    return ResponseEntity.ok("Sent!");
}

NOTE: When the ServiceBusMessageHeaders.SESSION_ID is set in the message headers, and a different ServiceBusMessageHeaders.PARTITION_KEY (or AzureHeaders.PARTITION_KEY) header is also set, the value of the session id will eventually be used to overwrite the value of the partition key.

Please use this sample as a reference to learn more about how to use this binder in your project. - Service Bus Queue

Troubleshooting

Logging setting

Please refer to [spring logging document] to get more information about logging.

Logging setting examples

Example: Setting logging level of hibernate

logging.level.root=WARN
logging.level.org.springframework.web=DEBUG
logging.level.org.hibernate=ERROR

Next steps

The following section provides sample projects illustrating how to use the starter in different cases.

More sample code

Contributing

This project welcomes contributions and suggestions. Most contributions require you to agree to a Contributor License Agreement (CLA) declaring that you have the right to, and actually do, grant us the rights to use your contribution. For details, visit https://cla.microsoft.com.

Please follow instructions here to build from source or contribute.

Packages
Package	Description
com.azure.spring.servicebus.stream.binder	Package com.azure.spring.servicebus.stream.binder
com.azure.spring.servicebus.stream.binder.config	Package com.azure.spring.servicebus.stream.binder.config;
com.azure.spring.servicebus.stream.binder.properties	Package com.azure.spring.servicebus.stream.binder.properties;
com.azure.spring.servicebus.stream.binder.provisioning	Package com.azure.spring.servicebus.stream.binder.provisioning;

Azure SDK for Java Reference Documentation

Azure Service Bus Queue Spring Cloud Stream Binder client library for Java

Getting started

Prerequisites

Include the package

Key concepts

Scheduled Message

Consumer Group

Partitioning Support

Configuration Options

Spring Cloud Azure Properties

Partition configuration

Service Bus Queue Producer Properties

Service Bus Queue Consumer Properties

Support for Service Bus Message Headers and Properties

Examples

Usage examples

Troubleshooting

Logging setting

Logging setting examples

Next steps

More sample code

Contributing