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.
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.
<?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.
Top comments (0)