Spring Cloud Functions, Kafka | How to interact asynchronous
Tags: springcloudstream, springcloudfunctions, kafka, java
Published: 2023-12-24 10:10:51

---
title: Spring Cloud Functions, Kafka | How to interact asynchronous
published: true
description: In this article I will show how you can use Spring Cloud Stream, Spring Cloud Functions and Kafka
tags: SpringCloudStream, SpringCloudFunctions, Kafka, Java
# cover_image: https://direct_url_to_image.jpg
# Use a ratio of 100:42 for best results.
# published_at: 2023-12-24 10:09 +0000
---


## Introduction

This article will show how to use the Spring Cloud Functions projects. The Spring Cloud Functions is the best approach to implementing a straightforward business logic.
Nowadays, the usage of message brokers such as Kafka has increased.
I will implement a simple system in this article, as shown below.

![Diagram of developing system](https://cdn-images-1.medium.com/max/3972/1*91p2A6X0hlcjhCdE7CVD6w.png)

## Implementation

I will start with a simple example of a cloud function, which is the same as a rest endpoint in the classical spring boot web application. You can create a project from your favorite IDE or from [Spring Initliazr.](https://start.spring.io/)

    <?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
             xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
        <parent>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-parent</artifactId>
            <version>3.2.1</version>
            <relativePath/> <!-- lookup parent from repository -->
        </parent>
        <groupId>io.vrnsky</groupId>
        <artifactId>cloud-function-push</artifactId>
        <version>0.0.1-SNAPSHOT</version>
        <name>cloud-function-push</name>
        <description>cloud-function-push</description>
        <properties>
            <java.version>17</java.version>
            <spring-cloud.version>2023.0.0</spring-cloud.version>
        </properties>
        <dependencies>
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-function-context</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-starter-function-web</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-test</artifactId>
                <scope>test</scope>
            </dependency>
        </dependencies>
        <dependencyManagement>
            <dependencies>
                <dependency>
                    <groupId>org.springframework.cloud</groupId>
                    <artifactId>spring-cloud-dependencies</artifactId>
                    <version>${spring-cloud.version}</version>
                    <type>pom</type>
                    <scope>import</scope>
                </dependency>
            </dependencies>
        </dependencyManagement>
    
        <build>
            <plugins>
                <plugin>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-maven-plugin</artifactId>
                </plugin>
            </plugins>
        </build>
    
    </project>

Your pom.xml file or build.gradle should like this.

The first step is to describe events related to service. I will stick with the following definition of an event.

    package io.vrnsky.cloudfunctionpush.domain;
    
    import java.util.List;
    
    public record PullRequestAccepted(
            String author,
            List<String> reviewers,
            
            boolean havePassedAllChecks
    ) {
    }

The first step is to describe events related to service. I will stick with the following definition of an event.
The second step is to create a configuration class within the bean function creation inside it.

    package io.vrnsky.cloudfunctionpush.config;
    
    import io.vrnsky.cloudfunctionpush.domain.PullRequestAccepted;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    import java.util.function.Function;
    
    
    @Configuration
    public class FunctionConfig {
    
        @Bean
        public Function<PullRequestAccepted, String> pullRequestAccepted() {
            return value -> """
                    Pull request from %s has been reviewed by %s. Have passed all checks: %b
                    """
                    .formatted(value.author(), value.reviewers().size(), value.havePassedAllChecks());
        }
    }

It’s time to start our cloud function push service and check that it works correctly.

    curl http://localhost:8080/pullRequestAccepted -H "Content-type:application/json" -d '{"author":"Yegor Voronianskii","reviewers":[],"havePassedAllChecks":true}'
    Pull request from Yegor Voronianskii has been reviewed by 0. Have passed all checks: true

To achieve that, we must add the following dependencies to pom.xml or build.gradle

            <dependency>
                <groupId>org.apache.kafka</groupId>
                <artifactId>kafka-streams</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-stream</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-stream-binder-kafka</artifactId>
            </dependency>
    

Before starting the service, it is required to start Kafka and Zookeeper. Here is a docker-compose file to start both.

    version: "3"
    services:
      zookeeper:
        image: confluentinc/cp-zookeeper:latest
        environment:
          ZOOKEEPER_CLIENT_PORT: 2181
        ports:
          - "22181:2181"
    
      kafka:
        image: confluentinc/cp-kafka:latest
        depends_on:
          - zookeeper
        environment:
          KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
          KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092,EXTERNAL://localhost:29092
          KAFKA_LISTENERS: PLAINTEXT://:9092,EXTERNAL://:29092
          KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,EXTERNAL:PLAINTEXT
          KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
          KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
        ports:
          - "9092:9092"
          - "29092:29092"

You can start Kafka and Zookeeper by following the command:

    docker compose up

The next step it to configure our application and add some code. Here, you can see the updated version of the FunctionConfig class, where an instance of the stream bridge is injected. You can use Lombok for brevity. It is up to you.

    package io.vrnsky.cloudfunctionpush.config;
    
    import io.vrnsky.cloudfunctionpush.domain.PullRequestAccepted;
    import org.springframework.cloud.stream.function.StreamBridge;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    import java.util.function.Function;
    
    
    @Configuration
    public class FunctionConfig {
    
        private final StreamBridge streamBridge;
    
        public FunctionConfig(StreamBridge streamBridge) {
            this.streamBridge = streamBridge;
        }
    
        @Bean
        public Function<PullRequestAccepted, String> pullRequestAccepted() {
            return value -> {
                streamBridge.send("pullRequestAccepted-out-0", value);
                return "Message have been sent";
            };
        }
    }

Below is the updated application configuration, where I specify the broker URL and other required settings.

    spring:
      kafka:
        bootstrap-servers:
          - http://localhost:29092
      cloud:
        stream:
          kafka:
            binder:
              brokers:
                - http://localhost:29092
          output-bindings: pullRequestAccepted-out-0
    
    logging:
      level:
        org.springframework.cloud.stream: TRACE

Now, we can check that the message has been sent to the broker.

    curl http://localhost:8080/pullRequestAccepted -H "Content-type:application/json" -d '{"author":"Yegor Voronianskii","reviewers":[],"havePassedAllChecks":true}'
    Message have been sent

In the cloud function push service, you should see something like this.

    --- [nio-8080-exec-1] o.s.c.s.binder.DefaultBinderFactory      : Retrieving cached binder: kafka
    --- [nio-8080-exec-1] o.s.c.s.m.DirectWithAttributesChannel    : preSend on channel 'bean 'pullRequestAccepted-out-0'', message: GenericMessage [payload=byte[73], headers={id=7c176f51-1a1a-63d9-b0ed-20cc1ce71fc8, contentType=application/json, target-protocol=kafka, timestamp=1703409370932}]
    --- [nio-8080-exec-1] tractMessageChannelBinder$SendingHandler : org.springframework.cloud.stream.binder.AbstractMessageChannelBinder$SendingHandler@7d99be9b received message: GenericMessage [payload=byte[73], headers={id=7c176f51-1a1a-63d9-b0ed-20cc1ce71fc8, contentType=application/json, target-protocol=kafka, timestamp=1703409370932}]
    --- [nio-8080-exec-1] nder$ProducerConfigurationMessageHandler : org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder$ProducerConfigurationMessageHandler@d36ee78 received message: GenericMessage [payload=byte[73], headers={id=d9ab9d27-fdf6-53a7-9db6-ae98e6130272, contentType=application/json, target-protocol=kafka, timestamp=1703409370933}]
    --- [nio-8080-exec-1] nder$ProducerConfigurationMessageHandler : handler 'org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder$ProducerConfigurationMessageHandler@d36ee78' produced no reply for request Message: GenericMessage [payload=byte[73], headers={id=d9ab9d27-fdf6-53a7-9db6-ae98e6130272, contentType=application/json, target-protocol=kafka, timestamp=1703409370933}]
    --- [nio-8080-exec-1] o.s.c.s.m.DirectWithAttributesChannel    : postSend (sent=true) on channel 'bean 'pullRequestAccepted-out-0'', message: GenericMessage [payload=byte[73], headers={id=7c176f51-1a1a-63d9-b0ed-20cc1ce71fc8, contentType=application/json, target-protocol=kafka, timestamp=1703409370932}]

The next step is implementing Spring Cloud Function to consume events from the message broker. The creation project’s steps are similar to those of the cloud push service. To configure our cloud pull service, we should have an application.properties or application.yml file with the following settings.

    spring:
      kafka:
        bootstrap-servers:
          - http://localhost:29092
      cloud:
        function:
          definition: pullRequestAccepted
        stream:
          kafka:
            binder:
              brokers:
                - http://localhost:29092
          input-bindings: pullRequestAccepted-in-0
          bindings:
            pullRequestAccepted-in-0:
              destination: pullRequestAccepted-out-0
              group: myGroup
    logging:
      level:
        org.springframework.cloud.stream: TRACE
    server:
      port: 9090

To handle incoming messages, we have to define the consumer of this message. I stick with a simple solution that prints the message into standard output.

    package com.example.cloudfunctionpull.config;
    
    import com.example.cloudfunctionpull.domain.PullRequestAccepted;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.messaging.Message;
    
    import java.util.function.Consumer;
    
    @Configuration
    public class FunctionsConfig {
    
        @Bean
        public Consumer<Message<PullRequestAccepted>> pullRequestAccepted() {
            return message -> {
                PullRequestAccepted payload = message.getPayload();
                System.out.println("Received message: " + payload);
            };
        }
    
        @Bean
        public Consumer<Message<?>> myErrorHandler() {
            return errorMessage -> {
                System.out.println("Handling error message " + errorMessage);
            };
        }
    }

After a successful start of consuming the service, we can again push the message to the message broker and check that the message was consumed to pull the service.

    kafka_consumer=org.springframework.kafka.core.DefaultKafkaConsumerFactory$ExtendedKafkaConsumer@1721c358, deliveryAttempt=1, kafka_timestampType=CREATE_TIME, kafka_receivedPartitionId=0, contentType=application/json, kafka_receivedTopic=pullRequestAccepted-out-0, kafka_receivedTimestamp=1703411981591, kafka_groupId=myGroup, target-protocol=kafka}]
    Received message: PullRequestAccepted[author=Yegor Voronianskii, reviewers=[], havePassedAllChecks=true]

## Conclusion

In my honest opinion, the Spring Cloud Function is a great tool for small, short-lived microservices. The Spring Cloud Stream is an excellent tool for building event-driven microservices architecture. But there are pros; of course, this project’s complexity can bring a lot of pain in terms of maintenance and understanding new paradigms.

## References

 1. [Spring Cloud Stream Documentation](https://docs.spring.io/spring-cloud-stream/docs/current/reference/html/)

 2. [Spring Cloud Functions Documentation](https://cloud.spring.io/spring-cloud-function/reference/html/)

 3. [Kafka Documentation](https://kafka.apache.org/documentation/)