maniish_jaiin

I am Manish and I am a Software Engineer by profession. I have been writing on Medium for a year now. I write majorly because it fulfills my desire to share my knowledge and gives me a creativity boost. <br> 
I write on a variety of topics like Software Engineering, Tech, Programming, Developer Productivity, Apple Products, and a few more. I have worked at ThoughtWorks and BCG Digital Ventures and I am currently living in Berlin. 

Software Engineer

Manish Jain

Learn how to process streams of data from Apache Kafka topics using Spring Cloud Streams

**TL;DR:** Have you ever wondered how features like Google Maps' live traffic work? These systems have to gather and process data in real-time. The architecture of these systems generally involves a data pipeline that processes and transfers data to be processed further until it reaches the clients. In this article, we will see something similar with a simple example using Kafka Streams. The sample app can be found [here](https://github.com/auth0-blog/spring-cloud-stream-sample).

## Introduction to Spring Cloud Stream

Spring Cloud Stream is a framework designed to support stream processing provided by various messaging systems like Apache Kafka, RabbitMQ, etc. The framework allows you to create processing logic without having to deal with any specific platform. It helps you build highly scalable event-driven microservices connected using these messaging systems.

The framework provides a flexible programming model built on already established and familiar Spring idioms and best practices. The way it works is simple; you have to provide implementations (called Binder implementations)for the messaging system that you are using. Spring cloud stream supports:

* [RabbitMQ](https://github.com/spring-cloud/spring-cloud-stream-binder-rabbit)
* [Apache Kafka](https://github.com/spring-cloud/spring-cloud-stream-binder-kafka)
* [Kafka Streams](https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/tree/main/spring-cloud-stream-binder-kafka-streams)
* [Amazon Kinesis](https://github.com/spring-cloud/spring-cloud-stream-binder-aws-kinesis)

And a few others. The links above will take you to the binder implementations. In this article, we will look into a simple application that uses Kafka Streams as a stream processor listening to events on a topic, processing the data, and publishing it to the outgoing topic.

## Introduction to Apache Kafka

Apache Kafka is a distributed publish-subscribe messaging system. It is a system that publishes and subscribes to a stream of records, similar to a message queue. Kafka is suitable for both offline and online message consumption. It is fault-tolerant, robust, and has a high throughput. Kafka is run as a cluster on one or more servers that can span multiple data centers. The Kafka cluster stores stream of records in categories called topics. Each record consists of a key, a value, and a timestamp. For more information on topics, Producer API, Consumer API, and event streaming, please visit this [link](https://kafka.apache.org/intro).

## Introduction to Kafka Streams

Kafka Streams is a library that can be used to consume data, process it, and produce new data, all in real-time. It works on a continuous, never-ending stream of data. Consider an example of the stock market. The stock prices fluctuate every second, and to be able to provide real-time value to the customer, you would use something like Kafka streams.

## Pre-requisites:

1. Basic knowledge of [Java 11](https://jdk.java.net/archive/).
2. Basic knowledge of [Spring Boot](https://spring.io/projects/spring-boot).
3. A basic understanding of [Apache Kafka](https://kafka.apache.org/intro).
4. [Docker](https://docs.docker.com/get-docker/) and [Docker Compose](https://docs.docker.com/compose/install/) for running Kafka locally.

## Setting up Spring Boot App

Let us first create a Spring Boot project with the help of the [Spring boot Initializr](https://start.spring.io/#!type=mavenproject&language=java&platformVersion=2.4.5.RELEASE&packaging=jar&jvmVersion=11&groupId=io.pratik&artifactId=springcloudsqs&name=springcloudsqs&description=Demo%20project%20for%20Spring%20cloud%20sqs&packageName=io.pratik.springcloudsqs&dependencies=web,lombok), and then open the project in our favorite IDE. Select Gradle project and Java language. Last but not least, select Spring boot version `2.5.4`. Fill in the project metadata and click generate.

For Spring Cloud, We need to configure Spring Kafka and Kafka Streams in our `gradle.build`:

```groovy

dependencies {
    implementation 'org.springframework.boot:spring-boot-starter'
    implementation 'org.apache.kafka:kafka-streams'
    implementation 'org.springframework.kafka:spring-kafka'
}

```

Let's setup the config for Kafka. We need to define a few parameters on how we want to serialize and deserialize the data. The config is easy to set up and understand. Since our application will be listening to a topic and producing the output to a different topic, our application is a producer and a consumer both. So, we need to define config for both producer and consumer.

In the application.yml file, we need to add these entries.

```yaml

kafka:
  bootstrap-servers: localhost:9092
  properties:
    schema.registry.url: http://localhost:8081

  producer:
    client-id: ${spring.application.name}
    key-serializer: org.apache.kafka.common.serialization.StringSerializer
    value-serializer: org.apache.kafka.common.serialization.StringSerializer

  consumer:
    client-id: ${spring.application.name}
    group-id: ${spring.application.name}-group
    key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    auto-offset-reset: earliest

  streams:
    client-id: ${spring.application.name}-stream
    application-id: ${spring.application.name}
    properties:
      default.key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
      default.value.serde: org.apache.kafka.common.serialization.Serdes$StringSerde

```

If you look at the config carefully, we are setting up serializers and de-serializers for the producer, the consumer, and the streams (serde is just short for serializer-deserializer). This is the only setup we need for the Spring boot project.

Let's jump into creating the producer, the consumer, and the stream processor. I have taken a simple example here. We are producing random numbers every 2 seconds using a scheduler.

```java

@Component
@AllArgsConstructor
public class NumberProducer {

    NumberPublisher numberPublisher;

    @Scheduled(fixedRate = 2000)
    public void produceIntStream() {
        Random random = new Random();
        numberPublisher.produce(random.nextInt(1000));
    }
}

```

The number publisher is the actual publisher that puts the data on a topic. We set a key for the message and the data (which is a random number in our case).

```java

@Slf4j
@Component
@AllArgsConstructor
public class NumberPublisher {

    private final KafkaTemplate<String, String> kafkaTemplate;

    public void produce(Integer randomNumber) {
        String s = "Odd";
        if (randomNumber % 2 == 0) s = "Even";
        System.out.println("Produced number: " + randomNumber);
        kafkaTemplate.send(INPUT_TOPIC_NAME, s, String.valueOf(randomNumber));
    }
}
```

The key is defined as a String, which is either even or odd based on the number. We use the Kafka template to send the message; this comes from the spring-kafka library. It abstracts out the logic for publishing and consuming the messages.

Next up, we set up our stream processor that listens to the topic on which the publisher is putting the messages. This is where it gets interesting. We listen to the `INPUT_TOPIC` and then process the data. In this case, the job of the stream processor is to filter out the odd numbers and only send the even numbers on the `OUTPUT_TOPIC`.

```java

@Configuration
@EnableKafkaStreams
public class KafkaStream {

    public static final String OUTPUT_TOPIC_NAME = "even-number-topic";
    public static final String INPUT_TOPIC_NAME = "number-topic";

    @Bean
    public KStream<String, String> evenNumbersStream(StreamsBuilder kStreamBuilder) {
        KStream<String, String> input = kStreamBuilder.stream(INPUT_TOPIC_NAME);

        KStream<String, String> output = input.filter((key, value) -> key.equals("Even"));

        output.to(OUTPUT_TOPIC_NAME);
        return output;
    }
}

```

You might be wondering about that KStream in the return type of our method. I will give a brief overview here as it is outside the scope of this article.

KStream -> A Kafka stream that is append-only. When you provide data with the same key, it will not update the previous record. It provides several operations that are very useful for data processing, like a filter, map, partition, flatMap, etc. You can read more about KStreams [here](https://www.confluent.co.uk/blog/introducing-kafka-streams-stream-processing-made-simple/).

Finally, when we have processed the data, we put it on an `OUTGOING_TOPIC`. For the sake of simplicity and completion, I am listening to that topic in our application. This generally will not be the case, as there would be another application that would be consuming from that topic and hence the name `OUTGOING_TOPIC`.

```java

@Component
@EnableKafka
public class EvenNumberConsumer {

    @KafkaListener(topics = OUTPUT_TOPIC_NAME)
    public void receive(String value) {
        System.out.println("Received number: " + value);
    }
}

```

The application code is complete. Let's set up Kafka locally.

## Setting up Kafka Locally

Setting up Kafka is easy, but it requires some dependency to run, you just need to use the docker-compose file below, and it will start the Kafka server locally. Add the `docker compose.yml` to the repository's root directory. Start the required dependency using: `docker-compose up`.

```yaml

version: '3.7'

services:
  kafka:
    image: confluentinc/cp-kafka:5.5.0
    container_name: kafka
    hostname: kafka
    restart: always
    environment:
      KAFKA_LISTENERS: LISTENER_DOCKER_INTERNAL://:29092,LISTENER_DOCKER_EXTERNAL://:9092
      KAFKA_ADVERTISED_LISTENERS: LISTENER_DOCKER_INTERNAL://kafka:29092,LISTENER_DOCKER_EXTERNAL://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: LISTENER_DOCKER_INTERNAL:PLAINTEXT,LISTENER_DOCKER_EXTERNAL:PLAINTEXT
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_INTER_BROKER_LISTENER_NAME: LISTENER_DOCKER_INTERNAL
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
    ports:
      - 9092:9092
      - 9999:9999
    depends_on:
      - zookeeper

  zookeeper:
    container_name: zookeeper
    hostname: zookeeper
    image: confluentinc/cp-zookeeper:5.5.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
    ports:
      - 2181:2181

  schema-registry:
    image: confluentinc/cp-schema-registry:5.5.0
    hostname: schema-registry
    container_name: schema-registry
    restart: always
    environment:
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: "kafka:29092"
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_LISTENERS: "http://0.0.0.0:8081"
    ports:
      - 8081:8081
    depends_on:
      - zookeeper

```

That's it!

## Verify Sending, Processing, and Receiving of Events

Run the Spring Boot app:

1. Clone the sample code from the repo. Go to the root directory.
2. Use the Gradle plugin to run your Spring Boot app using the command in the project directory.

```shell
./gradlew bootRun
```

Just run the application. You should see logs like this.

```shell
Received number: 910
Received number: 320
Received number: 16
Received number: 526
Received number: 76
Received number: 936
Received number: 642
Produced number: 510
Received number: 510
Produced number: 996
Received number: 996
Produced number: 897
```

## Conclusion

Spring Cloud Stream provides a simple and convenient way to create apps that can process streams and publish data to different topics. You can build micro-services that talk to each other using Kafka messages and process data like you would process in a single application.

In this article, we have learned how to build a Spring Cloud Stream app that uses Kafka Streams. We saw how Spring Cloud Stream provides an easy way to set up and run an application that can consumer, process, and publish messages to Kafka topics without the hassle of configuring each. With such little code, we could do so much.

You can refer to the repository used in the article on [Github](https://github.com/auth0-blog/spring-cloud-stream-sample).

Spring Cloud Streams with Apache Kafka

matthew-casperson

Technology enthusiast since my first C64. I love writing about new ideas and working with great people.

Product Manager at Octopus Deploy

Matthew Casperson

Learn how to integrate a browser client to reactive streams published by a Spring WebFlux server.

Reactive Streams with Kotlin, Webflux, and rsocket-js

amber-sharma

Amber is a Software Engineer at Auth0 who enjoys hard problems and learning. She switched to software engineering after falling in love with the wide-spread impact and embetterment offered by civic tech. In her free time, she can be found reading, eating pies, and finding adventure with her family.

Amber Sharma

Using Azure Event Grid with Dynamics 365 and Auth0 to Bolster Sales



This tutorial leverages the power of Auth0 Log Events to create a responsive sales pipeline using [Azure Event Grid](https://azure.microsoft.com/en-us/services/event-grid/overview).

Identity management is so much more than simply lock and key. While login events are the doorway into your product, sign-up events are the first opportunity to welcome your customers to their new home. Auth0 Log Streaming provides near real-time access to these events, providing you with the ability to respond instantly the moment they step foot through your door. 

With [Event Grid](https://azure.microsoft.com/en-us/services/event-grid/overview), Azure takes us one step further and provides us with a virtual valet &mdash; helping us to build event-driven applications, and routing our events to powerful, cloud-based services.

## What is Azure Event Grid?

Azure Event Grid is a highly-available event router that collects data from multiple sources and delivers them to Azure services. Whether it is streaming stock quotes via Service Bus and broadcasting to your users via WebSockets, or integrating your Auth0 identity data and kicking off serverless flows using Azure Functions, Event Grid is there to make it easy to develop serverless workflows, system automation, and distributed applications. 

One of the cool ways to leverage this integration is to automate intake tasks into your CRM. Instead of requiring your sales team to dig through email lists and enter in new leads, we can generate them and notify our teams within moments of their arrival.

In this tutorial, we will see just how easily we can leverage the power of this new integration by using Auth0 Log Events to build a CRM pipeline.

## What are we Building?

CRM professionals will tell you that organic growth is far more likely to lead to a sale than outreach. Now with Azure and [Auth0](https://auth0.com), you can feed every new sign-up directly into your CRM with minimal latency between customer action and engagement, generating new leads just as fast as they arrive. 

In this tutorial, you will be building an automatic Sales Leads pipeline using Microsoft’s CRM solution, Dynamics 365, in order to build a sales prospect pipeline.Every time a new user signs up for your service through Auth0’s user flow, a corresponding new potential sales lead will be populated in Dynamics 365. 

### Requirements:

- An Auth0 account. Currently, log-streaming is not supported for free past-the-trial-date or PSaaS accounts, however, you can <a href="https://auth0.com/signup" data-amp-replace="CLIENT_ID" data-amp-addparams="anonId=CLIENT_ID(cid-scope-cookie-fallback-name)">sign up for a free Auth0 trial account here</a>.
- A [Microsoft Azure](https://azure.microsoft.com/) subscription
- Access to a [Microsoft Dynamics 365: Sales account](https://dynamics.microsoft.com/en-us/sales/overview/) &mdash; we will run through the trial setup if you don't have one

Let’s get started!

## Step 1: Set up an Auth0 Event Source

In order to complete this step, you should know:

- Your Microsoft Azure Subscription ID
- Your relevant Azure Resource Group 
- The Azure region in which you will create your Partner Topic

Part of the integration process is to set Auth0 as an Azure event source. Here's how you do that:

- Log in to the [Auth0 Dashboard](https://manage.auth0.com).
- Navigate to "Logs" > "Streams.
- Click "+ Create Stream".
- Select Azure Event Grid and enter a unique name for your new stream.
- Create the event source by providing your Azure Subscription ID, Azure Region, and a Resource Group name. 
- Click Save.

## Step 2. Activate your Auth0 Partner Topic in Azure

Activating the Auth0 topic in Azure allows events to flow from Auth0 to Azure.

- After clicking "Save", click the link underneath your newly created Partner Topic in the Auth0 portal.
- You will now be redirected to your Partner Topic in the Azure Portal.
- Confirm the `Partner Topic` field matches your Auth0 account.
- Click "Activate".

## Step 3. Sign Up for a Dynamics 365: Sales Account

- Set up your [Dynamics 365 Sales Account trial](https://trials.dynamics.com/Dynamics365/Signup/), if applicable.
- Click "Sales", fill out the form, and get started!
- If you already have an account, sign in. 

## Step 4. Create a Logic App

- Search Logic Apps in the search bar in Azure.
- Create the app using your subscription, resource group, and desired region. 
- Go to resource once deployment completes. 

![Azure logic application designer](https://images.ctfassets.net/23aumh6u8s0i/3EEnKPw0vzPxYAOWl8MUWI/164da7c7c225e2f75f2c6b7ff0b64b08/azure-logics-app-designer)

- In "Logic App Designer", under "Start with a common trigger", click "When a Http Request is Received".

![Azure logic application designer common trigger](https://images.ctfassets.net/23aumh6u8s0i/7H9dSbzV64C61kXq8PYtUM/dcdffc08b6c863430903b77ec6c663f7/azure-app-trigger)

In the popup, add the sample schema of an Auth0 log. Alternatively, you can paste a sample payload to generate a schema following the link underneath the sample schema prompt:

```json
{
  "type": "object",
  "properties": {
    "specversion": {
      "type": "string"
    },
    "id": {
      "type": "string"
    },
    "source": {
      "type": "string"
    },
    "type": {
      "type": "string"
    },
    "datacontenttype": {
      "type": "string"
    },
    "data": {
      "type": "object",
      "properties": {
        "date": {
          "type": "string"
        },
        "type": {
          "type": "string"
        },
        "description": {
          "type": "string"
        },
        "connection": {
          "type": "string"
        },
        "connection_id": {
          "type": "string"
        },
        "client_id": {
          "type": "string"
        },
        "client_name": {
          "type": "string"
        },
        "ip": {
          "type": "string"
        },
        "user_agent": {
          "type": "string"
        },
        "details": {
          "type": "object",
          "properties": {
            "body": {
              "type": "object",
              "properties": {
                "tenant": {
                  "type": "string"
                },
                "client_id": {
                  "type": "string"
                },
                "connection": {
                  "type": "string"
                },
                "email": {
                  "type": "string"
                },
                "password": {
                  "type": "string"
                },
                "request_language": {
                  "type": "string"
                }
              }
            },
            "device_id": {
              "type": "string"
            }
          }
        },
        "user_id": {
          "type": "string"
        },
        "user_name": {
          "type": "string"
        },
        "strategy": {
          "type": "string"
        },
        "strategy_type": {
          "type": "string"
        },
        "log_id": {
          "type": "string"
        }
      }
    }
  }
}
```

Your Logic App will generate a URL upon save. Please copy this for later. Click "Run".

![Azure logic designer app URL](https://images.ctfassets.net/23aumh6u8s0i/4YGjYfnmpurh5gVIHQnXtC/c01d1420df34412ed7e90338fa75c61e/azure-app-url)

Next, you'll connect to Dynamics 365.

## Step 5. Connect to 365

First, click on "+ New step". Select Dynamics 365. If "Dynamics 365" does not appear, search for it in the search bar.

![Azure logic designer Dynamics 365](https://images.ctfassets.net/23aumh6u8s0i/lUjxWSqpUghLeK3J2joxP/5d63836aaf5fbec36f52e61eae02b8a8/dynamics-365)

- Click "Actions: Create a New Record".
- Sign in to 365 as asked. 
- Populate the Org name.
- Entity Name: Leads (for sales leads).
- Populate the fields as desired (with dynamic fields popup).
- Save and then click "Run". 

Next, let's create the new event subscription.

## Step 6. Create Event Subscription

- Create new event subscription.
- In Event Schema, select "Cloud Event Schema v1.0".
- For Topic Types, select "Event Grid Topics"
- Populate the Subscription and Resource Group.
- Under "Endpoint Details":
  - Select "Web Hook" as "Endpoint Type".
  - Then paste the endpoint.

![Azure create subscription](https://images.ctfassets.net/23aumh6u8s0i/1bOWMMGlzkF4LT71RNvxzU/e7990592ff14205295c8a043b408672c/azure-create-subscription)

- Next, click the "Filters" tab:
  - For Advanced Filters, user the following: 
    - Key: `data.type`
    - Operator: `string is in`
    - “ss”
  - Click "Create".

## Step 7. Generate a Sign up Log and View Your New Sales Leads

Congratulations! In seven simple steps, you have now created an automated sales pipeline!

![Azure new sales lead](https://images.ctfassets.net/23aumh6u8s0i/1OGkrKqMF2nIPkUvWgvQ8t/9ca3256420897f049e542f85c81e66f3/azure-new-sale)

## Learn more

Using Azure Event Grid with Auth0 amplifies the possibilities for customization and protection of the end-user experience. Learn how to [set up the  Azure Event Grid Log Streams integration here](https://auth0.com/docs/logs/streams/azure-event-grid), or [sign up for a free Auth0 trial](https://auth0.com/signup).

<include src="asides/AboutAuth0" />


Using Azure Event Grid and Auth0 to Bolster Sales

valery-calderon

Guest Author

Valery Calderon

Learn how to use reactive programming in Python to process data streams asynchronously.

"streams"