Spring Cloud Stream with Azure Event Hubs

Prerequisites

An Azure subscription - create one for free .

Java Development Kit (JDK) version 8 or higher.

Apache Maven , version 3.2 or higher.

cURL or a similar HTTP utility to test functionality.

An Azure Event hub. If you don't have one, create an event hub using Azure portal .

An Azure Storage Account for Event hub checkpoints. If you don't have one, create a storage account .

A Spring Boot application. If you don't have one, create a Maven project with the Spring Initializr . Be sure to select Maven Project and, under Dependencies , add the Spring Web and Azure Support dependencies, then select Java version 8 or higher.

To grant your account access to resources, in Azure Event Hubs, assign the Azure Event Hubs Data Receiver and Azure Event Hubs Data Sender role to the Azure AD account you're currently using. Then, in the Azure Storage account, assign the Storage Blob Data Contributor role to the Azure AD account you're currently using. For more information about granting access roles, see Assign Azure roles using the Azure portal and Authorize access to Event Hubs resources using Azure Active Directory .

Important

Spring Boot version 2.5 or higher is required to complete the steps in this tutorial.

Send and receive messages from Azure Event Hubs

With an Azure Storage Account and an Azure Event hub, you can send and receive messages using Spring Cloud Azure Stream Binder Event Hubs.

To install the Spring Cloud Azure Stream Binder Event Hubs module, add the following dependencies to your pom.xml file:

The Spring Cloud Azure Bill of Materials (BOM):

<dependencyManagement>
  <dependencies>
    <dependency>
      <groupId>com.azure.spring</groupId>
      <artifactId>spring-cloud-azure-dependencies</artifactId>
      <version>4.7.0</version>
      <type>pom</type>
      <scope>import</scope>
    </dependency>
  </dependencies>
</dependencyManagement>
If you're using Spring Boot 3.x, be sure to set the spring-cloud-azure-dependencies version to 5.0.0.
For more information about the spring-cloud-azure-dependencies version, see Which Version of Spring Cloud Azure Should I Use.
The Spring Cloud Azure Stream Binder Event Hubs artifact:
<dependency>
   <groupId>com.azure.spring</groupId>
   <artifactId>spring-cloud-azure-stream-binder-eventhubs</artifactId>
</dependency>
Code the application
Use the following steps to configure your application to produce and consume messages using Azure Event Hubs.
Configure the Event hub credentials by adding the following properties to your application.properties file.
 spring.cloud.azure.eventhubs.namespace=${AZURE_EVENTHUBS_NAMESPACE}
 spring.cloud.azure.eventhubs.processor.checkpoint-store.account-name=${AZURE_STORAGE_ACCOUNT_NAME}
 spring.cloud.azure.eventhubs.processor.checkpoint-store.container-name=${AZURE_STORAGE_CONTAINER_NAME}
 spring.cloud.stream.bindings.consume-in-0.destination=${AZURE_EVENTHUB_NAME}
 spring.cloud.stream.bindings.consume-in-0.group=${AZURE_EVENTHUB_CONSUMER_GROUP}
 spring.cloud.stream.bindings.supply-out-0.destination=${AZURE_EVENTHUB_NAME}
 spring.cloud.stream.eventhubs.bindings.consume-in-0.consumer.checkpoint.mode=MANUAL
 spring.cloud.stream.function.definition=consume;supply;
 spring.cloud.stream.poller.initial-delay=0
 spring.cloud.stream.poller.fixed-delay=1000
The following table describes the fields in the configuration:
Field
Description
spring.cloud.azure.eventhubs.namespace
Specify the namespace you obtained in your event hub from the Azure portal.
spring.cloud.azure.eventhubs.processor.checkpoint-store.container-name
Specify the container of your storage account.
spring.cloud.azure.eventhubs.processor.checkpoint-store.account-key
Specify the access-key of your storage account.
spring.cloud.azure.eventhubs.processor.checkpoint-store.account-name
Specify the storage account you created in this tutorial.
spring.cloud.stream.function.definition
Specify which functional bean to bind to the external destination(s) exposed by the bindings.
spring.cloud.stream.bindings.consume-in-0.destination
Specify the event hub you used in this tutorial.
spring.cloud.stream.bindings.consume-in-0.group
Specify the Consumer groups in your Event Hubs Instance.
spring.cloud.stream.bindings.supply-out-0.destination
Specify the same event hub you used in this tutorial.
spring.cloud.stream.eventhubs.bindings.consume-in-0.consumer.checkpoint.mode
Specify MANUAL.
spring.cloud.stream.poller.fixed-delay
Specify fixed delay for default poller in milliseconds. The default value is 1000 L.
spring.cloud.stream.poller.initial-delay
Specify initial delay for periodic triggers. The default value is 0.
Edit the startup class file to show the following content.
import com.azure.spring.messaging.checkpoint.Checkpointer;
import com.azure.spring.messaging.eventhubs.support.EventHubsHeaders;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;
import java.util.function.Consumer;
import java.util.function.Supplier;
import static com.azure.spring.messaging.AzureHeaders.CHECKPOINTER;
@SpringBootApplication
public class EventHubBinderApplication implements CommandLineRunner {
    private static final Logger LOGGER = LoggerFactory.getLogger(EventHubBinderApplication.class);
    private static final Sinks.Many<Message<String>> many = Sinks.many().unicast().onBackpressureBuffer();
    public static void main(String[] args) {
        SpringApplication.run(EventHubBinderApplication.class, args);
    @Bean
    public Supplier<Flux<Message<String>>> supply() {
        return ()->many.asFlux()
                       .doOnNext(m->LOGGER.info("Manually sending message {}", m))
                       .doOnError(t->LOGGER.error("Error encountered", t));
    @Bean
    public Consumer<Message<String>> consume() {
        return message->{
            Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER);
            LOGGER.info("New message received: '{}', partition key: {}, sequence number: {}, offset: {}, enqueued "
                    +"time: {}",
                message.getPayload(),
                message.getHeaders().get(EventHubsHeaders.PARTITION_KEY),
                message.getHeaders().get(EventHubsHeaders.SEQUENCE_NUMBER),
                message.getHeaders().get(EventHubsHeaders.OFFSET),
                message.getHeaders().get(EventHubsHeaders.ENQUEUED_TIME)
            checkpointer.success()
                        .doOnSuccess(success->LOGGER.info("Message '{}' successfully checkpointed",
                            message.getPayload()))
                        .doOnError(error->LOGGER.error("Exception found", error))
                        .block();
    @Override
    public void run(String... args) {
        LOGGER.info("Going to add message {} to sendMessage.", "Hello Word");
        many.emitNext(MessageBuilder.withPayload("Hello Word").build(), Sinks.EmitFailureHandler.FAIL_FAST);
In this tutorial, there are no authentication operations in the configurations or the code. However, connecting to Azure services requires authentication. To complete the authentication, you need to use Azure Identity. Spring Cloud Azure uses DefaultAzureCredential, which the Azure Identity library provides to help you get credentials without any code changes.
DefaultAzureCredential supports multiple authentication methods and determines which method to use at runtime. This approach enables your app to use different authentication methods in different environments (such as local and production environments) without implementing environment-specific code. For more information, see the Default Azure credential section of Authenticate Azure-hosted Java applications.
To complete the authentication in local development environments, you can use Azure CLI, Visual Studio Code, PowerShell or other methods. For more information, see Azure authentication in Java development environments. To complete the authentication in Azure hosting environments, we recommend using managed identity. For more information, see What are managed identities for Azure resources?
Start the application. Messages like this will be posted in your application log, as shown in the following example output:
New message received: 'Hello Word', partition key: 107207233, sequence number: 458, offset: 94256, enqueued time: 2023-02-17T08:27:59.641Z
Message 'Hello Word!' successfully checkpointed
Deploy to Azure Spring Apps
Now that you have the Spring Boot application running locally, it's time to move it to production. Azure Spring Apps makes it easy to deploy Spring Boot applications to Azure without any code changes. The service manages the infrastructure of Spring applications so developers can focus on their code. Azure Spring Apps provides lifecycle management using comprehensive monitoring and diagnostics, configuration management, service discovery, CI/CD integration, blue-green deployments, and more. To deploy your application to Azure Spring Apps, see Deploy your first application to Azure Spring Apps.
Next steps
Azure for Spring developers
Spring Cloud Azure Stream Binder Event Hubs Samples