You can read more about Istio, Spring Boot, and other Java and Kubernetes-related technologies in my latest book, Hands-On Java with Kubernetes.

You can also find many other articles on my blog about Istio. For example, this article is about Quarkus and tracing with Istio.

Source Code

Feel free to use my source code if you’d like to try it out yourself. To do that, you must clone my sample GitHub repository. Two sample applications for this exercise are available in the spring-boot-istio directory.

Prerequisites

We will start with the most straightforward Istio installation on a local Kubernetes cluster. This could be Minikube, which you can run with the following command. You can set slightly lower resource limits than I did.

minikube start --memory='8gb' --cpus='6'

minikube start --memory='8gb' --cpus='6'

To complete the exercise below, you need to install istioctl in addition to kubectl. Here you will find the available distributions for the latest versions of kubectl and istioctl. I install them on my laptop using Homebrew.

$ brew install kubectl
$ brew install istioctl

$ brew install kubectl
$ brew install istioctl

To install Istio with default parameters, run the following command:

istioctl install

istioctl install

After a moment, Istio should be running in the istio-system namespace.

It is also worth installing Kiali to verify the Istio resources we have created. Kiali is an observability and management tool for Istio that provides a web-based dashboard for service mesh monitoring. It visualizes service-to-service traffic, validates Istio configuration, and integrates with tools like Prometheus, Grafana, and Jaeger.

kubectl apply -f https://raw.githubusercontent.com/istio/istio/release-1.28/samples/addons/kiali.yaml

kubectl apply -f https://raw.githubusercontent.com/istio/istio/release-1.28/samples/addons/kiali.yaml

Once Kiali is successfully installed on Kubernetes, we can expose its web dashboard locally with the following istioctl command:

istioctl dashboard kiali

istioctl dashboard kiali

To test access to the Istio mesh from outside the cluster, you need to expose the ingress gateway. To do this, run the minikube tunnel command.

Use Spring Boot Istio Library

To test our library’s functionality, we will create a simple Spring Boot application that exposes a single REST endpoint.

@RestController
@RequestMapping("/callme")
public class CallmeController {

    private static final Logger LOGGER = LoggerFactory
        .getLogger(CallmeController.class);

    @Autowired
    Optional<BuildProperties> buildProperties;
    @Value("${VERSION}")
    private String version;

    @GetMapping("/ping")
    public String ping() {
        LOGGER.info("Ping: name={}, version={}", buildProperties.isPresent() ?
            buildProperties.get().getName() : "callme-service", version);
        return "I'm callme-service " + version;
    }
    
}

@RestController
@RequestMapping("/callme")
public class CallmeController {

    private static final Logger LOGGER = LoggerFactory
        .getLogger(CallmeController.class);

    @Autowired
    Optional<BuildProperties> buildProperties;
    @Value("${VERSION}")
    private String version;

    @GetMapping("/ping")
    public String ping() {
        LOGGER.info("Ping: name={}, version={}", buildProperties.isPresent() ?
            buildProperties.get().getName() : "callme-service", version);
        return "I'm callme-service " + version;
    }
    
}

Then, in addition to the standard Spring Web starter, add the istio-spring-boot-starter dependency.

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
  <groupId>com.github.piomin</groupId>
  <artifactId>istio-spring-boot-starter</artifactId>
  <version>1.2.1</version>
</dependency>

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
  <groupId>com.github.piomin</groupId>
  <artifactId>istio-spring-boot-starter</artifactId>
  <version>1.2.1</version>
</dependency>

Finally, we must add the @EnableIstio annotation to our application’s main class. We can also enable Istio Gateway to expose the REST endpoint outside the cluster. An Istio Gateway is a component that controls how external traffic enters or leaves a service mesh.

@SpringBootApplication
@EnableIstio(enableGateway = true)
public class CallmeApplication {

	public static void main(String[] args) {
		SpringApplication.run(CallmeApplication.class, args);
	}
	
}

@SpringBootApplication
@EnableIstio(enableGateway = true)
public class CallmeApplication {

	public static void main(String[] args) {
		SpringApplication.run(CallmeApplication.class, args);
	}
	
}

Let’s deploy our application on the Kubernetes cluster. To do this, we must first create a role with the necessary permissions to manage Istio resources in the cluster. The role must be assigned to the ServiceAccount used by the application.

apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  name: callme-service-with-starter
rules:
  - apiGroups: ["networking.istio.io"]
    resources: ["virtualservices", "destinationrules", "gateways"]
    verbs: ["create", "get", "list", "watch", "update", "patch", "delete"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  name: callme-service-with-starter
subjects:
  - kind: ServiceAccount
    name: callme-service-with-starter
    namespace: spring
roleRef:
  kind: Role
  name: callme-service-with-starter
  apiGroup: rbac.authorization.k8s.io
---
apiVersion: v1
kind: ServiceAccount
metadata:
  name: callme-service-with-starter

apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  name: callme-service-with-starter
rules:
  - apiGroups: ["networking.istio.io"]
    resources: ["virtualservices", "destinationrules", "gateways"]
    verbs: ["create", "get", "list", "watch", "update", "patch", "delete"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  name: callme-service-with-starter
subjects:
  - kind: ServiceAccount
    name: callme-service-with-starter
    namespace: spring
roleRef:
  kind: Role
  name: callme-service-with-starter
  apiGroup: rbac.authorization.k8s.io
---
apiVersion: v1
kind: ServiceAccount
metadata:
  name: callme-service-with-starter

Here are the Deployment and Service resources.

apiVersion: apps/v1
kind: Deployment
metadata:
  name: callme-service-with-starter
spec:
  replicas: 1
  selector:
    matchLabels:
      app: callme-service-with-starter
  template:
    metadata:
      labels:
        app: callme-service-with-starter
    spec:
      containers:
        - name: callme-service-with-starter
          image: piomin/callme-service-with-starter
          imagePullPolicy: IfNotPresent
          ports:
            - containerPort: 8080
          env:
            - name: VERSION
              value: "v1"
      serviceAccountName: callme-service-with-starter
---
apiVersion: v1
kind: Service
metadata:
  name: callme-service-with-starter
  labels:
    app: callme-service-with-starter
spec:
  type: ClusterIP
  ports:
  - port: 8080
    name: http
  selector:
    app: callme-service-with-starter

apiVersion: apps/v1
kind: Deployment
metadata:
  name: callme-service-with-starter
spec:
  replicas: 1
  selector:
    matchLabels:
      app: callme-service-with-starter
  template:
    metadata:
      labels:
        app: callme-service-with-starter
    spec:
      containers:
        - name: callme-service-with-starter
          image: piomin/callme-service-with-starter
          imagePullPolicy: IfNotPresent
          ports:
            - containerPort: 8080
          env:
            - name: VERSION
              value: "v1"
      serviceAccountName: callme-service-with-starter
---
apiVersion: v1
kind: Service
metadata:
  name: callme-service-with-starter
  labels:
    app: callme-service-with-starter
spec:
  type: ClusterIP
  ports:
  - port: 8080
    name: http
  selector:
    app: callme-service-with-starter

The application repository is configured to run it with Skaffold. Of course, you can apply YAML manifests to the cluster with kubectl apply. To do this, simply navigate to the callme-service-with-starter/k8s directory and apply the deployment.yaml file. As part of the exercise, we will run our applications in the spring namespace.

skaffold dev -n spring

skaffold dev -n spring

The sample Spring Boot application creates two Istio objects at startup: a VirtualService and a Gateway. We can verify them in the Kiali dashboard.

The default host name generated for the gateway includes the deployment name and the .ext suffix. We can change the suffix name using the domain field in the @EnableIstio annotation. Assuming you run the minikube tunnel command, you can call the service using the Host header with the hostname in the following way:

$ curl http://localhost/callme/ping -H "Host:callme-service-with-starter.ext"
I'm callme-service v1

$ curl http://localhost/callme/ping -H "Host:callme-service-with-starter.ext"
I'm callme-service v1

Additional Capabilities with Spring Boot Istio

The library’s behavior can be customized by modifying the @EnableIstio annotation. For example, you can enable the fault injection mechanism using the fault field. Both abort and delay are possible. You can make this change without redeploying the app. The library updates the existing VirtualService.

@SpringBootApplication
@EnableIstio(enableGateway = true, fault = @Fault(percentage = 50))
public class CallmeApplication {

	public static void main(String[] args) {
		SpringApplication.run(CallmeApplication.class, args);
	}
	
}

@SpringBootApplication
@EnableIstio(enableGateway = true, fault = @Fault(percentage = 50))
public class CallmeApplication {

	public static void main(String[] args) {
		SpringApplication.run(CallmeApplication.class, args);
	}
	
}

Now, you can call the same endpoint through the gateway. There is a 50% chance that you will receive this response.

Now let’s analyze a slightly more complex scenario. Let’s assume that we are running two different versions of the same application on Kubernetes. We use Istio to manage its versioning. Traffic is forwarded to the particular version based on the X-Version header in the incoming request. If the header value is v1 the request is sent to the application Pod with the label version=v1, and similarly, for the version v2. Here’s the annotation for the v1 application main class.

@SpringBootApplication
@EnableIstio(enableGateway = true, version = "v1",
	matches = { 
	  @Match(type = MatchType.HEADERS, key = "X-Version", value = "v1") })
public class CallmeApplication {

	public static void main(String[] args) {
		SpringApplication.run(CallmeApplication.class, args);
	}
	
}

@SpringBootApplication
@EnableIstio(enableGateway = true, version = "v1",
	matches = { 
	  @Match(type = MatchType.HEADERS, key = "X-Version", value = "v1") })
public class CallmeApplication {

	public static void main(String[] args) {
		SpringApplication.run(CallmeApplication.class, args);
	}
	
}

The Deployment manifest for the callme-service-with-starter-v1 should define two labels: app and version.

apiVersion: apps/v1
kind: Deployment
metadata:
  name: callme-service-with-starter-v1
spec:
  replicas: 1
  selector:
    matchLabels:
      app: callme-service-with-starter
      version: v1
  template:
    metadata:
      labels:
        app: callme-service-with-starter
        version: v1
    spec:
      containers:
        - name: callme-service-with-starter
          image: piomin/callme-service-with-starter
          imagePullPolicy: IfNotPresent
          ports:
            - containerPort: 8080
          env:
            - name: VERSION
              value: "v1"
      serviceAccountName: callme-service-with-starter

apiVersion: apps/v1
kind: Deployment
metadata:
  name: callme-service-with-starter-v1
spec:
  replicas: 1
  selector:
    matchLabels:
      app: callme-service-with-starter
      version: v1
  template:
    metadata:
      labels:
        app: callme-service-with-starter
        version: v1
    spec:
      containers:
        - name: callme-service-with-starter
          image: piomin/callme-service-with-starter
          imagePullPolicy: IfNotPresent
          ports:
            - containerPort: 8080
          env:
            - name: VERSION
              value: "v1"
      serviceAccountName: callme-service-with-starter

Unlike the skaffold dev command, the skaffold run command simply launches the application on the cluster and terminates. Let’s first release version v1, and then move on to version v2.

skaffold run -n spring

skaffold run -n spring

Then, we can deploy the v2 version of our sample Spring Boot application. In this exercise, it is just an “artificial” version, since we deploy the same source code, but with different labels and environment variables injected in the Deployment manifest.

@SpringBootApplication
@EnableIstio(enableGateway = true, version = "v2",
	matches = { 
	  @Match(type = MatchType.HEADERS, key = "X-Version", value = "v2") })
public class CallmeApplication {

	public static void main(String[] args) {
		SpringApplication.run(CallmeApplication.class, args);
	}
	
}

@SpringBootApplication
@EnableIstio(enableGateway = true, version = "v2",
	matches = { 
	  @Match(type = MatchType.HEADERS, key = "X-Version", value = "v2") })
public class CallmeApplication {

	public static void main(String[] args) {
		SpringApplication.run(CallmeApplication.class, args);
	}
	
}

Here’s the callme-service-with-starter-v2 Deployment manifest.

apiVersion: apps/v1
kind: Deployment
metadata:
  name: callme-service-with-starter-v2
spec:
  replicas: 1
  selector:
    matchLabels:
      app: callme-service-with-starter
      version: v2
  template:
    metadata:
      labels:
        app: callme-service-with-starter
        version: v2
    spec:
      containers:
        - name: callme-service-with-starter
          image: piomin/callme-service-with-starter
          imagePullPolicy: IfNotPresent
          ports:
            - containerPort: 8080
          env:
            - name: VERSION
              value: "v2"
      serviceAccountName: callme-service-with-starter

apiVersion: apps/v1
kind: Deployment
metadata:
  name: callme-service-with-starter-v2
spec:
  replicas: 1
  selector:
    matchLabels:
      app: callme-service-with-starter
      version: v2
  template:
    metadata:
      labels:
        app: callme-service-with-starter
        version: v2
    spec:
      containers:
        - name: callme-service-with-starter
          image: piomin/callme-service-with-starter
          imagePullPolicy: IfNotPresent
          ports:
            - containerPort: 8080
          env:
            - name: VERSION
              value: "v2"
      serviceAccountName: callme-service-with-starter

Service, on the other hand, remains unchanged. It still refers to Pods labeled with app=callme-service-with-starter. However, this time it includes both application instances marked as v1 or v2.

apiVersion: v1
kind: Service
metadata:
  name: callme-service-with-starter
  labels:
    app: callme-service-with-starter
spec:
  type: ClusterIP
  ports:
  - port: 8080
    name: http
  selector:
    app: callme-service-with-starter

apiVersion: v1
kind: Service
metadata:
  name: callme-service-with-starter
  labels:
    app: callme-service-with-starter
spec:
  type: ClusterIP
  ports:
  - port: 8080
    name: http
  selector:
    app: callme-service-with-starter

Version v2 should be run in the same way as before, using the skaffold run command. There are three Istio objects generated during apps startup: Gateway, VirtualService and DestinationRule.

A generated DestinationRule contains two subsets for both v1 and v2 versions.

The automatically generated VirtualService looks as follows.

kind: VirtualService
apiVersion: networking.istio.io/v1
metadata:
  name: callme-service-with-starter-route
spec:
  hosts:
  - callme-service-with-starter
  - callme-service-with-starter.ext
  gateways:
  - callme-service-with-starter
  http:
  - match:
    - headers:
        X-Version:
          prefix: v1
    route:
    - destination:
        host: callme-service-with-starter
        subset: v1
    timeout: 6s
    retries:
      attempts: 3
      perTryTimeout: 2s
      retryOn: 5xx
  - match:
    - headers:
        X-Version:
          prefix: v2
    route:
    - destination:
        host: callme-service-with-starter
        subset: v2
      weight: 100
    timeout: 6s
    retries:
      attempts: 3
      perTryTimeout: 2s
      retryOn: 5xx

kind: VirtualService
apiVersion: networking.istio.io/v1
metadata:
  name: callme-service-with-starter-route
spec:
  hosts:
  - callme-service-with-starter
  - callme-service-with-starter.ext
  gateways:
  - callme-service-with-starter
  http:
  - match:
    - headers:
        X-Version:
          prefix: v1
    route:
    - destination:
        host: callme-service-with-starter
        subset: v1
    timeout: 6s
    retries:
      attempts: 3
      perTryTimeout: 2s
      retryOn: 5xx
  - match:
    - headers:
        X-Version:
          prefix: v2
    route:
    - destination:
        host: callme-service-with-starter
        subset: v2
      weight: 100
    timeout: 6s
    retries:
      attempts: 3
      perTryTimeout: 2s
      retryOn: 5xx

To test the versioning mechanism generated using the Spring Boot Istio library, set the X-Version header for each call.

$ curl http://localhost/callme/ping -H "Host:callme-service-with-starter.ext" -H "X-Version:v1"
I'm callme-service v1

$ curl http://localhost/callme/ping -H "Host:callme-service-with-starter.ext" -H "X-Version:v2"
I'm callme-service v2

$ curl http://localhost/callme/ping -H "Host:callme-service-with-starter.ext" -H "X-Version:v1"
I'm callme-service v1

$ curl http://localhost/callme/ping -H "Host:callme-service-with-starter.ext" -H "X-Version:v2"
I'm callme-service v2

Conclusion

I am still working on this library, and new features will be added in the near future. I hope it will be helpful for those of you who want to get started with Istio without getting into the details of its configuration.

The post Istio Spring Boot Library Released appeared first on Piotr's TechBlog.

Here is a brief overview of all my published books.

Motivation

I won’t hide that this post is mainly directed at my blog subscribers and people who enjoy reading it and value my writing style. As you know, all posts and content on my blog, along with sample application repositories on GitHub, are always accessible to you for free. Over the past eight years, I have worked to publish high-quality content on my blog, and I plan to keep doing so. It is a part of my life, a significant time commitment, but also a lot of fun and a hobby.

I want to explain why I decided to write this book, why now, and why in this way. But first, a bit of background. I wrote my last book first, then my first book, over seven years ago. It focused on topics I was mainly involved with at the time, specifically Spring Boot and Spring Cloud. Since then, a lot of time has passed, and much has changed – not only in the technology itself but also a little in my personal life. Today, I am more involved in Kubernetes and container topics than, for example, Spring Cloud. For years, I have been helping various organizations transition from traditional application architectures to cloud-native models based on Kubernetes. Of course, Java remains my main area of expertise. Besides Spring Boot, I also really like the Quarkus framework. You can read a lot about both in my book on Kubernetes.

Based on my experience over the past few years, involving development teams is a key factor in the success of the Kubernetes platform within an organization. Ultimately, it is the applications developed by these teams that are deployed there. For developers to be willing to use Kubernetes, it must be easy for them to do so. That is why I persuade organizations to remove barriers to using Kubernetes and to design it in a way that makes it easier for development teams. On my blog and in this book, I aim to demonstrate how to quickly and simply launch applications on Kubernetes using frameworks such as Spring Boot and Quarkus.

It’s an unusual time to publish a book. AI agents are producing more and more technical content online. More often than not, instead of grabbing a book, people turn to an AI chatbot for a quick answer, though not always the best one. Still, a book that thoroughly introduces a topic and offers a step-by-step guide remains highly valuable.

Content of the Book

This book demonstrates that Java is an excellent choice for building applications that run on Kubernetes. In the first chapter, I’ll show you how to quickly build your application, create its image, and run it on Kubernetes without writing a single line of YAML or Dockerfile. This chapter also covers the minimum Kubernetes architecture you must understand to manage applications effectively in this environment. The second chapter, on the other hand, demonstrates how to effectively organize your local development environment to work with a Kubernetes cluster. You’ll see several options for running a distribution of your cluster locally and learn about the essential set of tools you should have. The third chapter outlines best practices for building applications on the Kubernetes platform. Most of the presented requirements are supported by simple examples and explanations of the benefits of meeting them. The fourth chapter presents the most valuable tools for the inner development loop with Kubernetes. After reading the first four chapters, you will understand the main Kubernetes components related to application management, enabling you to navigate the platform efficiently. You’ll also learn to leverage Spring Boot and Quarkus features to adapt your application to Kubernetes requirements.

In the following chapters, I will focus on the benefits of migrating applications to Kubernetes. The first area to cover is security. Chapter five discusses mechanisms and tools for securing applications running in a cluster. Chapter six describes Spring and Quarkus projects that enable native integration with the Kubernetes API from within applications. In chapter seven, you’ll learn about the service mesh tool and the benefits of using it to manage HTTP traffic between microservices. Chapter eight addresses the performance and scalability of Java applications in a Kubernetes environment. Chapter Eight demonstrates how to design a CI/CD process that runs entirely within the cluster, leveraging Kubernetes-native tools for pipeline building and the GitOps approach. This book also covers AI. In the final, ninth chapter, you’ll learn how to run a simple Java application that integrates with an AI model deployed on Kubernetes.

Publication

I decided to publish my book on Leanpub. Leanpub is a platform for writing, publishing, and selling books, especially popular among technical content authors. I previously published a book with Packt, but honestly, I was alone during the writing process. Leanpub is similar but offers several key advantages over publishers like Packt. First, it allows you to update content collaboratively with readers and keep it current. Even though my book is finished, I don’t rule out adding more chapters, such as on AI on Kubernetes. I also look forward to your feedback and plan to improve the content and examples in the repository continuously. Overall, this has been another exciting experience related to publishing technical content.

And when you buy such a book, you can be sure that most of the royalties go to me as the author, unlike with other publishers, where most of the royalties go to them as promoters. So, I’m looking forward to improving my book with you!

Conclusion

My book aims to bring together all the most interesting elements surrounding Java application development on Kubernetes. It is intended not only for developers but also for architects and DevOps teams who want to move to the Kubernetes platform.

The post A Book: Hands-On Java with Kubernetes appeared first on Piotr's TechBlog.

If you are looking for other interesting articles about Quarkus, you will find them in my blog. For example, you will read more about testing strategies with Quarkus and Pact here.

Source Code

If you would like to try it by yourself, you may always take a look at my source code. To do that, you must clone my sample GitHub repository. Then you should only follow my instructions

Architecture

Before proceeding to the implementation, let’s take a look at the diagram of our system architecture. There are three microservices: employee-service, departament-service, and organization-service. They are communicating with each other through a REST API. They use the Consul Key-Value store as a distributed configuration backend. Every instance of service is registering itself in Consul. A load balancer is included in the application. It reads a list of registered instances of a target service from the Consul using the Quarkus Stork extension. Then it chooses an instance using a provided algorithm.

Running Consul Instance

We will run a single-node Consul instance as a Docker container. By default, Consul exposes HTTP API and a UI console on the 8500 port. Let’s expose that port outside the container.

docker run -d --name=consul \
   -e CONSUL_BIND_INTERFACE=eth0 \
   -p 8500:8500 \
   consul

Dependencies

Let’s analyze a list of the most important Maven dependencies using the department-service application as an example. Our application exposes REST endpoints and connects to the in-memory H2 database. We use the Quarkus REST client and the SmallRye Stork Service Discovery library to implement communication between the microservices. On the other hand, the io.quarkiverse.config:quarkus-config-consul is responsible for reading configuration properties the Consul Key-Value store. With the smallrye-mutiny-vertx-consul-client library the application is able to interact directly with the Consul HTTP API. This may not be necessary in the future, once the Stork project will implement the registration and deregistration mechanism. Currently it is not ready. Finally, we will Testcontainers to run Consul and tests our apps against it with the Quarkus JUnit support.

	<dependencies>
		<dependency>
			<groupId>io.quarkus</groupId>
			<artifactId>quarkus-rest-jackson</artifactId>
		</dependency>
		<dependency>
			<groupId>io.quarkus</groupId>
			<artifactId>quarkus-rest-client-jackson</artifactId>
		</dependency>
		<dependency>
			<groupId>io.quarkus</groupId>
			<artifactId>quarkus-hibernate-orm-panache</artifactId>
		</dependency>
		<dependency>
			<groupId>io.quarkus</groupId>
			<artifactId>quarkus-jdbc-h2</artifactId>
		</dependency>
		<dependency>
			<groupId>com.h2database</groupId>
			<artifactId>h2</artifactId>
			<scope>runtime</scope>
		</dependency>
		<dependency>
			<groupId>io.quarkus</groupId>
			<artifactId>quarkus-smallrye-stork</artifactId>
		</dependency>
		<dependency>
			<groupId>io.smallrye.reactive</groupId>
			<artifactId>smallrye-mutiny-vertx-consul-client</artifactId>
		</dependency>
		<dependency>
			<groupId>io.smallrye.stork</groupId>
			<artifactId>stork-service-discovery-consul</artifactId>
		</dependency>
		<dependency>
			<groupId>io.smallrye.stork</groupId>
			<artifactId>stork-service-registration-consul</artifactId>
		</dependency>
		<dependency>
			<groupId>io.quarkus</groupId>
			<artifactId>quarkus-scheduler</artifactId>
		</dependency>
		<dependency>
			<groupId>io.quarkiverse.config</groupId>
			<artifactId>quarkus-config-consul</artifactId>
			<version>${quarkus-consul.version}</version>
		</dependency>
		<dependency>
			<groupId>io.rest-assured</groupId>
			<artifactId>rest-assured</artifactId>
			<scope>test</scope>
		</dependency>
		<dependency>
			<groupId>io.quarkus</groupId>
			<artifactId>quarkus-junit5</artifactId>
			<scope>test</scope>
		</dependency>
		<dependency>
			<groupId>org.testcontainers</groupId>
			<artifactId>consul</artifactId>
			<version>1.20.3</version>
			<scope>test</scope>
		</dependency>
		<dependency>
			<groupId>org.testcontainers</groupId>
			<artifactId>junit-jupiter</artifactId>
			<version>1.20.3</version>
			<scope>test</scope>
		</dependency>
	</dependencies>

Discovery and Load Balancing with Quarkus Stork for Consul

Let’s begin with the Quarkus Stork part. In the previous section, we included libraries required to provide service discovery and load balancing with Stork: quarkus-smallrye-stork and stork-service-discovery-consul. Now, we can proceed to the implementation. Here’s the EmployeeClient interface from the department-service responsible for calling the GET /employees/department/{departmentId} endpoint exposed by the employee-service. Instead of setting the target URL inside the @RegisterRestClient annotation we should refer to the name of the service registered in Consul.

@Path("/employees")
@RegisterRestClient(baseUri = "stork://employee-service")
public interface EmployeeClient {

    @GET
    @Path("/department/{departmentId}")
    @Produces(MediaType.APPLICATION_JSON)
    List<Employee> findByDepartment(@PathParam("departmentId") Long departmentId);

}

That service name should also be used in the configuration properties. The following property indicates that Stork will use Consul as a discovery server for the employee-service name.

quarkus.stork.employee-service.service-discovery.type = consul

Once we create a REST client with the additional annotations, we must inject it into the DepartmentResource class using the @RestClient annotation. Afterward, we can use that client to interact with the employee-service while calling the GET /departments/organization/{organizationId}/with-employees from the department-service.

@Path("/departments")
@Produces(MediaType.APPLICATION_JSON)
public class DepartmentResource {

    private Logger logger;
    private DepartmentRepository repository;
    private EmployeeClient employeeClient;

    public DepartmentResource(Logger logger,
                              DepartmentRepository repository,
                              @RestClient EmployeeClient employeeClient) {
        this.logger = logger;
        this.repository = repository;
        this.employeeClient = employeeClient;
    }

    // ... other methods for REST endpoints 

    @Path("/organization/{organizationId}")
    @GET
    public List<Department> findByOrganization(@PathParam("organizationId") Long organizationId) {
        logger.infof("Department find: organizationId=%d", organizationId);
        return repository.findByOrganization(organizationId);
    }

    @Path("/organization/{organizationId}/with-employees")
    @GET
    public List<Department> findByOrganizationWithEmployees(@PathParam("organizationId") Long organizationId) {
        logger.infof("Department find with employees: organizationId=%d", organizationId);
        List<Department> departments = repository.findByOrganization(organizationId);
        departments.forEach(d -> d.setEmployees(employeeClient.findByDepartment(d.getId())));
        return departments;
    }

}

Let’s take a look at the implementation of the GET /employees/department/{departmentId} in the employee-service called by the EmployeeClient in the department-service.

@Path("/employees")
@Produces(MediaType.APPLICATION_JSON)
public class EmployeeResource {

    private Logger logger;
    private EmployeeRepository repository;

    public EmployeeResource(Logger logger,
                            EmployeeRepository repository) {
        this.logger = logger;
        this.repository = repository;
    }

    @Path("/department/{departmentId}")
    @GET
    public List<Employee> findByDepartment(@PathParam("departmentId") Long departmentId) {
        logger.infof("Employee find: departmentId=%s", departmentId);
        return repository.findByDepartment(departmentId);
    }

    @Path("/organization/{organizationId}")
    @GET
    public List<Employee> findByOrganization(@PathParam("organizationId") Long organizationId) {
        logger.infof("Employee find: organizationId=%s", organizationId);
        return repository.findByOrganization(organizationId);
    }
    
    // ... other methods for REST endpoints

}

Similarly in the organization-service, we define two REST clients for interacting with employee-service and department-service.

@Path("/departments")
@RegisterRestClient(baseUri = "stork://department-service")
public interface DepartmentClient {

    @GET
    @Path("/organization/{organizationId}")
    @Produces(MediaType.APPLICATION_JSON)
    List<Department> findByOrganization(@PathParam("organizationId") Long organizationId);

    @GET
    @Path("/organization/{organizationId}/with-employees")
    @Produces(MediaType.APPLICATION_JSON)
    List<Department> findByOrganizationWithEmployees(@PathParam("organizationId") Long organizationId);

}

@Path("/employees")
@RegisterRestClient(baseUri = "stork://employee-service")
public interface EmployeeClient {

    @GET
    @Path("/organization/{organizationId}")
    @Produces(MediaType.APPLICATION_JSON)
    List<Employee> findByOrganization(@PathParam("organizationId") Long organizationId);

}

It involves the need to include the following two configuration properties that set the discovery service type for the target services.

quarkus.stork.employee-service.service-discovery.type = consul
quarkus.stork.department-service.service-discovery.type = consul

The OrganizationResource class injects and uses both previously created clients.

@Path("/organizations")
@Produces(MediaType.APPLICATION_JSON)
public class OrganizationResource {

    private Logger logger;
    private OrganizationRepository repository;
    private DepartmentClient departmentClient;
    private EmployeeClient employeeClient;

    public OrganizationResource(Logger logger,
                                OrganizationRepository repository,
                                @RestClient DepartmentClient departmentClient,
                                @RestClient EmployeeClient employeeClient) {
        this.logger = logger;
        this.repository = repository;
        this.departmentClient = departmentClient;
        this.employeeClient = employeeClient;
    }

    // ... other methods for REST endpoints

    @Path("/{id}/with-departments")
    @GET
    public Organization findByIdWithDepartments(@PathParam("id") Long id) {
        logger.infof("Organization find with departments: id={}", id);
        Organization organization = repository.findById(id);
        organization.setDepartments(departmentClient.findByOrganization(organization.getId()));
        return organization;
    }

    @Path("/{id}/with-departments-and-employees")
    @GET
    public Organization findByIdWithDepartmentsAndEmployees(@PathParam("id") Long id) {
        logger.infof("Organization find with departments and employees: id={}", id);
        Organization organization = repository.findById(id);
        organization.setDepartments(departmentClient.findByOrganizationWithEmployees(organization.getId()));
        return organization;
    }

    @Path("/{id}/with-employees")
    @GET
    public Organization findByIdWithEmployees(@PathParam("id") Long id) {
        logger.infof("Organization find with employees: id={}", id);
        Organization organization = repository.findById(id);
        organization.setEmployees(employeeClient.findByOrganization(organization.getId()));
        return organization;
    }

}

Registration in Consul with Quarkus

After including Stork, the Quarkus REST client automatically splits traffic between all the instances of the application existing in the discovery server. However, each application must register itself in the discovery server. Quarkus Stork won’t do that. Theoretically, there is the stork-service-registration-consul module that should register the application instance on startup. As far as I know, this feature is still under active development. For now, we will include a mentioned library and use the same property for enabling the registrar feature.

quarkus.stork.employee-service.service-registrar.type = consul

Our sample applications will interact directly with the Consul server using the SmallRye Mutiny reactive client. Let’s define the ClientConsul bean. It is registered only if the quarkus.stork.employee-service.service-registrar.type property with the consul value exists.

@ApplicationScoped
public class EmployeeBeanProducer {

    @ConfigProperty(name = "consul.host", defaultValue = "localhost")  String host;
    @ConfigProperty(name = "consul.port", defaultValue = "8500") int port;

    @Produces
    @LookupIfProperty(name = "quarkus.stork.employee-service.service-registrar.type", 
                      stringValue = "consul")
    public ConsulClient consulClient(Vertx vertx) {
        return ConsulClient.create(vertx, new ConsulClientOptions()
                .setHost(host)
                .setPort(port));
    }

}

The bean responsible for catching the startup and shutdown events is annotated with @ApplicationScoped. It defines two methods: onStart and onStop. It also injects the ConsulClient bean. Quarkus dynamically generates the HTTP listen port number on startup and saves it in the quarkus.http.port property. Therefore, the startup task needs to wait a moment to ensure that the application is running. We will run it 3 seconds after receiving the startup event. Every instance of the application needs to have a unique id in Consul. Therefore, we retrieve the number of running port and use that number as the id suffix. The name of the service is taken from the quarkus.application.name property. The instance of the application should save id in order to be able to deregister itself on shutdown.

@ApplicationScoped
public class EmployeeLifecycle {

    @ConfigProperty(name = "quarkus.application.name")
    private String appName;
    private int port;

    private Logger logger;
    private Instance<ConsulClient> consulClient;
    private ScheduledExecutorService executor;

    public EmployeeLifecycle(Logger logger,
                             Instance<ConsulClient> consulClient,
                             ScheduledExecutorService executor) {
        this.logger = logger;
        this.consulClient = consulClient;
        this.executor = executor;
    }

    void onStart(@Observes StartupEvent ev) {
        if (consulClient.isResolvable()) {
            executor.schedule(() -> {
                port = ConfigProvider.getConfig().getValue("quarkus.http.port", Integer.class);
                consulClient.get().registerService(new ServiceOptions()
                                .setPort(port)
                                .setAddress("localhost")
                                .setName(appName)
                                .setId(appName + "-" + port),
                        result -> logger.infof("Service %s-%d registered", appName, port));
            }, 3000, TimeUnit.MILLISECONDS);
        }
    }

    void onStop(@Observes ShutdownEvent ev) {
        if (consulClient.isResolvable()) {
            consulClient.get().deregisterService(appName + "-" + port,
                    result -> logger.infof("Service %s-%d deregistered", appName, port));
        }
    }
}

Read Configuration Properties from Consul

The io.quarkiverse.config:quarkus-config-consul is already included in dependencies. Once the quarkus.consul-config.enabled property is set to true, the Quarkus application tries to read properties from the Consul Key-Value store. The quarkus.consul-config.properties-value-keys property indicates the location of the properties file stored in Consul. Here are the properties that exists in the classpath application.properties. For example, the default config location for the department-service is config/department-service.

quarkus.application.name = department-service
quarkus.application.version = 1.1
quarkus.consul-config.enabled = true
quarkus.consul-config.properties-value-keys = config/${quarkus.application.name}

Let’s switch to the Consul UI. It is available under the same 8500 port as the API. In the “Key/Value” section we create configuration for all three sample applications.

These are configuration properties for department-service. They are targeted for the development mode. We enable the dynamically generated port number to run several instances on the same workstation. Our application use an in-memory H2 database. It loads the import.sql script on startup to initialize a demo data store. We also enable Quarkus Stork service discovery for the employee-service REST client and registration in Consul.

quarkus.http.port = 0
quarkus.datasource.db-kind = h2
quarkus.hibernate-orm.database.generation = drop-and-create
quarkus.hibernate-orm.sql-load-script = src/main/resources/import.sql
quarkus.stork.employee-service.service-discovery.type = consul
quarkus.stork.department-service.service-registrar.type = consul

Here are the configuration properties for the employee-service.

Finally, let’s take a look at the organization-service configuration in Consul.

Run Applications in the Development Mode

Let’s run our three sample Quarkus applications in the development mode. Both employee-service and department-service should have two instances running. We don’t have to take care about port conflicts, since they are aqutomatically generated on startup.

$ cd employee-service
$ mvn quarkus:dev
$ mvn quarkus:dev

$ cd department-service
$ mvn quarkus:dev
$ mvn quarkus:dev

$ cd organization-service
$ mvn quarkus:dev

Once we start all the instances we can switch to the Consul UI. You should see exactly the same services in your web console.

There are two instances of the employee-service and deparment-service. We can check out the list of registered instances for the selected application.

This step is optional. To simplify tests I also included API gateway that integrates with Consul discovery. It listens on the static 8080 port and forwards requests to the downstream services, which listen on the dynamic ports. Since Quarkus does not provide a module dedicates for the API gateway, I used Spring Cloud Gateway with Spring Cloud Consul for that. Therefore, you need to use the following command to run the application:

$ cd gateway-service
$ mvn spring-boot:run

Afterward, we can make some API tests with or without the gateway. With the gateway-service, we can use the 8080 port with the /api base context path. Let’s call the following three endpoints. The first one is exposed by the department-service, while the another two by the organization-service.

$ curl http://localhost:8080/api/departments/organization/1/with-employees
$ curl http://localhost:8080/api/organizations/1/with-departments
$ curl http://localhost:8080/api/organizations/1/with-departments-and-employees

Each Quarkus service listens on the dynamic port and register itself in Consul using that port number. Here’s the department-service logs from startup and during test communication.

After including the quarkus-micrometer-registry-prometheus module each application instance exposes metrics under the GET /q/metrics endpoint. There are several metrics related to service discovery published by the Quarkus Stork extension.

$ curl http://localhost:51867/q/metrics | grep stork
# TYPE stork_service_discovery_instances_count counter
# HELP stork_service_discovery_instances_count The number of service instances discovered
stork_service_discovery_instances_count_total{service_name="employee-service"} 12.0
# TYPE stork_service_selection_duration_seconds summary
# HELP stork_service_selection_duration_seconds The duration of the selection operation
stork_service_selection_duration_seconds_count{service_name="employee-service"} 6.0
stork_service_selection_duration_seconds_sum{service_name="employee-service"} 9.93934E-4
# TYPE stork_service_selection_duration_seconds_max gauge
# HELP stork_service_selection_duration_seconds_max The duration of the selection operation
stork_service_selection_duration_seconds_max{service_name="employee-service"} 0.0
# TYPE stork_service_discovery_failures counter
# HELP stork_service_discovery_failures The number of failures during service discovery
stork_service_discovery_failures_total{service_name="employee-service"} 0.0
# TYPE stork_service_discovery_duration_seconds_max gauge
# HELP stork_service_discovery_duration_seconds_max The duration of the discovery operation
stork_service_discovery_duration_seconds_max{service_name="employee-service"} 0.0
# TYPE stork_service_discovery_duration_seconds summary
# HELP stork_service_discovery_duration_seconds The duration of the discovery operation
stork_service_discovery_duration_seconds_count{service_name="employee-service"} 6.0
stork_service_discovery_duration_seconds_sum{service_name="employee-service"} 2.997176541
# TYPE stork_service_selection_failures counter
# HELP stork_service_selection_failures The number of failures during service selection
stork_service_selection_failures_total{service_name="employee-service"} 0.0

Advanced Load Balancing with Quarkus Stork and Consul

Quarkus Stork provides several load balancing strategies to efficiently distribute requests across multiple instances of a application. It can ensure optimal resource usage, better performance, and high availability. By default, Quarkus Stork uses round-robin algorithm. To override the default strategy, we first need to include a library responsible for providing the selected load-balancing algorithm. For example, let’s choose the least-response-time strategy, which collects response times of the calls made with service instances and picks an instance based on this information.

<dependency>
    <groupId>io.smallrye.stork</groupId>
    <artifactId>stork-load-balancer-least-response-time</artifactId>
</dependency>

Then, we have to change the default strategy in configuration properties for the selected client. Let’s add the following property to the config/department-service in Consul Key-Value store.

quarkus.stork.employee-service.load-balancer.type=least-response-time

After that, we can restart the instance of department-service and retest the communication between services.

Testing Integration Between Quarkus and Consul

We have already included the org.testcontainers:consul artifact to the Maven dependencies. Thanks to that, we can create JUnit tests with Quarkus and Testcontainers Consul. Since Quarkus doen’t provide a built-in support for testing Consul container, we need to create the class that implements the QuarkusTestResourceLifecycleManager interface. It is responsible for starting and stopping Consul container during JUnit tests. After starting the container, we add required configuration properties to enable in-memory database creation and a service registration in Consul.

public class ConsulResource implements QuarkusTestResourceLifecycleManager {

    private ConsulContainer consulContainer;

    @Override
    public Map<String, String> start() {
        consulContainer = new ConsulContainer("hashicorp/consul:latest")
                .withConsulCommand(
                """
                kv put config/department-service - <<EOF
                department.name=abc
                quarkus.datasource.db-kind=h2
                quarkus.hibernate-orm.database.generation=drop-and-create
                quarkus.stork.department-service.service-registrar.type=consul
                EOF
                """
                );

        consulContainer.start();

        String url = consulContainer.getHost() + ":" + consulContainer.getFirstMappedPort();

        return ImmutableMap.of(
                "quarkus.consul-config.agent.host-port", url,
                "consul.host", consulContainer.getHost(),
                "consul.port", consulContainer.getFirstMappedPort().toString()
        );
    }

    @Override
    public void stop() {
        consulContainer.stop();
    }
}

To start Consul container during the test, we need to annotate the test class with @QuarkusTestResource(ConsulResource.class). The test loads configuration properties from Consul on startup and registers the service. Then, it verifies that REST endpoints exposed by the department-service work fine and the registered service exists in Consul.

@QuarkusTest
@QuarkusTestResource(ConsulResource.class)
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
public class DepartmentResourceConsulTests {

    @ConfigProperty(name = "department.name", defaultValue = "")
    private String name;
    @Inject
    ConsulClient consulClient;

    @Test
    @Order(1)
    void add() {
        Department d = new Department();
        d.setOrganizationId(1L);
        d.setName(name);

        given().body(d).contentType(ContentType.JSON)
                .when().post("/departments").then()
                .statusCode(200)
                .body("id", notNullValue())
                .body("name", is(name));
    }

    @Test
    @Order(2)
    void findAll() {
        when().get("/departments").then()
                .statusCode(200)
                .body("size()", is(4));
    }

    @Test
    @Order(3)
    void checkRegister() throws InterruptedException {
        Thread.sleep(5000);
        Uni<ServiceList> uni = Uni.createFrom().completionStage(() -> consulClient.catalogServices().toCompletionStage());
        List<Service> services = uni.await().atMost(Duration.ofSeconds(3)).getList();
        final long count = services.stream()
                .filter(svc -> svc.getName().equals("department-service")).count();
        assertEquals(1 ,count);
    }
}

Final Thoughts

This article introduces Quarkus Stork for Consul discovery and client-side load balancing. It shows how to integrate Quarkus with Consul Key-Value store for distributed configuration. It also covers the topics like integration testing with Testcontainers support, metrics, service registration and advanced load-balancing strategies.

The post Consul with Quarkus and SmallRye Stork appeared first on Piotr's TechBlog.

If you are looking for more articles related to Spring Boot 3 you can refer to the following one, about microservices with Spring Cloud.

Source Code

If you would like to try it by yourself, you may always take a look at my source code. Since I’m using Testcontainers often, you can find examples in my several repositories. Here’s a list of repositories we will use today:

• https://github.com/piomin/sample-spring-boot-on-kubernetes.git
• https://github.com/piomin/sample-spring-microservices-advanced.git
• https://github.com/piomin/sample-spring-kafka-microservices.git

You can clone them and then follow my instruction to see how to leverage Spring Boot built-in support for Testcontainers and Docker Compose in development mode.

Use Testcontainers in Tests

Let’s start with the standard usage example. The first repository has a single Spring Boot app that connects to the Mongo database. In order to build automated tests we have to include the following Maven dependencies:

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-test</artifactId>
  <scope>test</scope>
</dependency>
<dependency>
  <groupId>org.testcontainers</groupId>
  <artifactId>mongodb</artifactId>
  <scope>test</scope>
</dependency>
<dependency>
  <groupId>org.testcontainers</groupId>
  <artifactId>junit-jupiter</artifactId>
  <scope>test</scope>
</dependency>

Now, we can create the tests. We need to annotate our test class with @Testcontainers. Then, we have to declare the MongoDBContainer bean. Before Spring Boot 3.1, we would have to use DynamicPropertyRegistry to set the Mongo address automatically generated by Testcontainers.

@SpringBootTest(webEnvironment = 
   SpringBootTest.WebEnvironment.RANDOM_PORT)
@Testcontainers
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
public class PersonControllerTest {

   @Container
   static MongoDBContainer mongodb = 
      new MongoDBContainer("mongo:5.0");

   @DynamicPropertySource
   static void registerMongoProperties(DynamicPropertyRegistry registry) {
      registry.add("spring.data.mongodb.uri", mongodb::getReplicaSetUrl);
   }

   // ... test methods

}

Fortunately, beginning from Spring Boot 3.1 we can simplify that notation with @ServiceConnection annotation. Here’s the full test implementation with the latest approach. It verifies some REST endpoints exposed by the app.

@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@Testcontainers
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
public class PersonControllerTest {

    private static String id;

    @Container
    @ServiceConnection
    static MongoDBContainer mongodb = new MongoDBContainer("mongo:5.0");

    @Autowired
    TestRestTemplate restTemplate;

    @Test
    @Order(1)
    void add() {
        Person p = new Person(null, "Test", "Test", 100, Gender.FEMALE);
        Person personAdded = restTemplate
            .postForObject("/persons", p, Person.class);
        assertNotNull(personAdded);
        assertNotNull(personAdded.getId());
        assertEquals(p.getLastName(), personAdded.getLastName());
        id = personAdded.getId();
    }

    @Test
    @Order(2)
    void findById() {
        Person person = restTemplate
            .getForObject("/persons/{id}", Person.class, id);
        assertNotNull(person);
        assertNotNull(person.getId());
        assertEquals(id, person.getId());
    }

    @Test
    @Order(2)
    void findAll() {
        Person[] persons = restTemplate
            .getForObject("/persons", Person[].class);
        assertEquals(6, persons.length);
    }

}

Now, we can build the project with the standard Maven command. Then Testcontainers will automatically start the Mongo database before the test. Of course, we need to have Docker running on our machine.

$ mvn clean package

Tests run fine. But what will happen if we would like to run our app locally for development? We can do it by running the app main class directly from IDE or with the mvn spring-boot:run Maven command. Here’s our main class:

@SpringBootApplication
@EnableMongoRepositories
public class SpringBootOnKubernetesApp implements ApplicationListener<ApplicationReadyEvent> {

    public static void main(String[] args) {
        SpringApplication.run(SpringBootOnKubernetesApp.class, args);
    }

    @Autowired
    PersonRepository repository;

    @Override
    public void onApplicationEvent(ApplicationReadyEvent applicationReadyEvent) {
        if (repository.count() == 0) {
            repository.save(new Person(null, "XXX", "FFF", 20, Gender.MALE));
            repository.save(new Person(null, "AAA", "EEE", 30, Gender.MALE));
            repository.save(new Person(null, "ZZZ", "DDD", 40, Gender.FEMALE));
            repository.save(new Person(null, "BBB", "CCC", 50, Gender.MALE));
            repository.save(new Person(null, "YYY", "JJJ", 60, Gender.FEMALE));
        }
    }
}

Of course, unless we start the Mongo database our app won’t be able to connect it. If we use Docker, we first need to execute the docker run command that runs MongoDB and exposes it on the local port.

Use Testcontainers in Development Mode with Spring Boot

Fortunately, with Spring Boot 3.1 we can simplify that process. We don’t have to Mongo before starting the app. What we need to do – is to enable development mode with Testcontainers. Firstly, we should include the following Maven dependency in the test scope:

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-testcontainers</artifactId>
  <scope>test</scope>
</dependency>

Then we need to prepare the @TestConfiguration class with the definition of containers we want to start together with the app. For me, it is just a single MongoDB container as shown below:

@TestConfiguration
public class MongoDBContainerDevMode {

    @Bean
    @ServiceConnection
    MongoDBContainer mongoDBContainer() {
        return new MongoDBContainer("mongo:5.0");
    }

}

After that, we have to “override” the Spring Boot main class. It should have the same name as the main class with the Test suffix. Then we pass the current main method inside the SpringApplication.from(...) method. We also need to set @TestConfiguration class using the with(...) method.

public class SpringBootOnKubernetesAppTest {

    public static void main(String[] args) {
        SpringApplication.from(SpringBootOnKubernetesApp::main)
                .with(MongoDBContainerDevMode.class)
                .run(args);
    }

}

Finally, we can start our “test” main class directly from the IDE or we can just execute the following Maven command:

$ mvn spring-boot:test-run

Once the app starts you will see that the Mongo container is up and running and connection to it is established.

Since we are in dev mode we will also include the Spring Devtools module to automatically restart the app after the source code change.

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-devtools</artifactId>
  <optional>true</optional>
</dependency>

Let’s what happened. Once we provide a change in the source code Spring Devtools will restart the app and the Mongo container. You can verify it in the app logs and also on the list of running Docker containers. As you see the Testcontainer ryuk has been initially started a minute ago, while Mongo was restarted after the app restarted 9 seconds ago.

In order to prevent restarting the container on app restart with Devtools we need to annotate the MongoDBContainer bean with @RestartScope.

@TestConfiguration
public class MongoDBContainerDevMode {

    @Bean
    @ServiceConnection
    @RestartScope
    MongoDBContainer mongoDBContainer() {
        return new MongoDBContainer("mongo:5.0");
    }

}

Now, Devtools just restart the app without restarting the container.

Sharing Container across Multiple Apps

In the previous example, we have a single app that connects to the database on a single container. Now, we will switch to the repository with some microservices that communicates with each other via the Kafka broker. Let’s say I want to develop and test all three apps simultaneously. Of course, our services need to have the following Maven dependencies:

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-testcontainers</artifactId>
  <scope>test</scope>
</dependency>
<dependency>
  <groupId>org.testcontainers</groupId>
  <artifactId>kafka</artifactId>
  <version>1.18.1</version>
  <scope>test</scope>
</dependency>

Then we need to do a very similar thing as before – declare the @TestConfiguration bean with a list of required containers. However, this time we need to make our Kafka container reusable between several apps. In order to do that, we will invoke the withReuse(true) on the KafkaContainer. By the way, it is also possible to use Kafka Raft mode instead of Zookeeper.

@TestConfiguration
public class KafkaContainerDevMode {

    @Bean
    @ServiceConnection
    public KafkaContainer kafka() {
        return new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:7.4.0"))
                .withKraft()
                .withReuse(true);
    }

}

The same as before we have to create a “test” main class that uses the @TestConfiguration bean. We will do the same thing for two other apps inside the repository: payment-service and stock-service.

public class OrderAppTest {

    public static void main(String[] args) {
        SpringApplication.from(OrderApp::main)
                .with(KafkaContainerDevMode.class)
                .run(args);
    }

}

Let’s run our three microservices. Just to remind you, it is possible to run the “test” main class directly from IDE or with the mvn spring-boot:test-run command. As you see, I run all three apps.

Now, if we display a list of running containers, there is only one Kafka broker shared between all the apps.

Use Spring Boot support for Docker Compose

Beginning from version 3.1 Spring Boot provides built-in support for Docker Compose. Let’s switch to our last sample repository. It consists of several microservices that connect to the Mongo database and the Netflix Eureka discovery server. We can go to the directory with one of the microservices, e.g. customer-service. Assuming we include the following Maven dependency, Spring Boot looks for a Docker Compose configuration file in the current working directory. Let’s activate that mechanism only for a specific Maven profile:

<profiles>
  <profile>
    <id>compose</id>
    <dependencies>
      <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-docker-compose</artifactId>
        <optional>true</optional>
      </dependency>
    </dependencies>
  </profile>
</profiles>

Our goal is to run all the required external services before running the customer-service app. The customer-service app connects to Mongo, Eureka, and calls endpoint exposed by the account-service. Here’s the implementation of the REST client that communicates to the account-service.

@FeignClient("account-service")
public interface AccountClient {

    @RequestMapping(method = RequestMethod.GET, value = "/accounts/customer/{customerId}")
    List<Account> getAccounts(@PathVariable("customerId") String customerId);

}

We need to prepare the docker-compose.yml with all required containers definition. As you see, there is the mongo service and two applications discovery-service and account-service, which uses local Docker images.

version: "3.8"
services:
  mongo:
    image: mongo:5.0
    ports:
      - "27017:27017"
  discovery-service:
    image: sample-spring-microservices-advanced/discovery-service:1.0-SNAPSHOT
    ports:
      - "8761:8761"
    healthcheck:
      test: curl --fail http://localhost:8761/eureka/v2/apps || exit 1
      interval: 4s
      timeout: 2s
      retries: 3
    environment:
      SPRING_PROFILES_ACTIVE: docker
  account-service:
    image: sample-spring-microservices-advanced/account-service:1.0-SNAPSHOT
    ports:
      - "8080"
    depends_on:
      discovery-service:
        condition: service_healthy
    links:
      - mongo
      - discovery-service
    environment:
      SPRING_PROFILES_ACTIVE: docker

Before we run the service, let’s build the images with our apps. We could as well use built-in Spring Boot mechanisms based on Buildpacks, but I’ve got some problems with it. Jib works fine in my case.

<profile>
  <id>build-image</id>
  <build>
    <plugins>
      <plugin>
        <groupId>com.google.cloud.tools</groupId>
        <artifactId>jib-maven-plugin</artifactId>
        <version>3.3.2</version>
        <configuration>
          <to>
            <image>sample-spring-microservices-advanced/${project.artifactId}:${project.version}</image>
          </to>
        </configuration>
        <executions>
          <execution>
            <goals>
              <goal>dockerBuild</goal>
            </goals>
            <phase>package</phase>
          </execution>
        </executions>
      </plugin>
    </plugins>
  </build>
</profile>

Let’s execute the following command on the repository root directory:

$ mvn clean package -Pbuild-image -DskipTests

After a successful build, we can verify a list of available images with the docker images command. As you see, there are two images used in our docker-compose.yml file:

Finally, the only thing you need to do is to run the customer-service app. Let’s switch to the customer-service directory once again and execute the mvn spring-boot:run with a profile that includes the spring-boot-docker-compose dependency:

$ mvn spring-boot:run -Pcompose

As you see, our app locates docker-compose.yml.

Once we start our app, it also starts all required containers.

For example, we can take a look at the Eureka dashboard available at http://localhost:8761. There are two apps registered there. The account-service is running on Docker, while the customer-service has been started locally.

Final Thoughts

Spring Boot 3.1 comes with several improvements in the area of containerization. Especially the feature related to the ability to run Testcontainers in development together with the app was something that I was waiting for. I hope this article will clarify how you can take advantage of the latest Spring Boot features for better integration with Testcontainers and Docker Compose.

The post Spring Boot Development Mode with Testcontainers and Docker appeared first on Piotr's TechBlog.

There are some significant changes in the approach to observability between Spring Boot 2 and 3. Tracing will no longer be part of Spring Cloud through the Spring Cloud Sleuth project. The core of that project has been moved to Micrometer Tracing. You can read more about the reasons and future plans in this post on the Spring blog.

The main goal of that article is to give you a simple receipt for how to enable observability for your microservices written in Spring Boot using a new approach. In order to simplify our exercise, we will use a fully managed Grafana instance in their cloud. We will build a very basic architecture with two microservices running locally. Let’s take a moment to discuss our architecture in great detail.

Source Code

If you would like to try it by yourself, you may always take a look at my source code. In order to do that you need to clone my GitHub repository. It contains several tutorials. You need to go to the inter-communication directory. After that, you should just follow my instructions.

Spring Boot Observability Architecture

There are two applications: inter-callme-service and inter-caller-service. The inter-caller-service app calls the HTTP endpoint exposed by the inter-callme-service app. We run two instances of inter-callme-service. We will also configure a static load balancing between those two instances using Spring Cloud Load Balancer. All the apps will expose Prometheus metrics using Spring Boot Actuator and the Micrometer project. For tracing, we are going to use Open Telemetry with Micrometer Tracing and OpenZipkin. In order to send all the data including logs, metrics, and traces from our local Spring Boot instances to the cloud, we have to use Grafana Agent.

On the other hand, there is a stack responsible for collecting and visualizing all the data. As I mentioned before we will leverage Grafana Cloud for that. It is a very comfortable way since we don’t have to install and configure all the required tools. First of all, Grafana Cloud offers a ready instance of Prometheus responsible for collecting metrics. We also need a log aggregation tool for storing and querying logs from our apps. Grafana Cloud offers a preconfigured instance of Loki for that. Finally, we have a distributed tracing backend through the Grafana Tempo. Here’s the visualization of our whole architecture.

Enable Metrics and Tracing with Micrometer

In order to export metrics in Prometheus format, we need to include the micrometer-registry-prometheus dependency. For tracing context propagation we should add the micrometer-tracing-bridge-otel module. We should also export tracing spans using one of the formats supported by Grafana Tempo. It will be OpenZipkin through the opentelemetry-exporter-zipkin dependency.

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<dependency>
  <groupId>io.micrometer</groupId>
  <artifactId>micrometer-registry-prometheus</artifactId>
</dependency>
<dependency>
  <groupId>io.micrometer</groupId>
  <artifactId>micrometer-tracing-bridge-otel</artifactId>
</dependency>
<dependency>
  <groupId>io.opentelemetry</groupId>
  <artifactId>opentelemetry-exporter-zipkin</artifactId>
</dependency>

We need to use the latest available version of Spring Boot 3. Currently, it is 3.0.0-RC1. As a release candidate that version is available in the Spring Milestone repository.

<parent>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-parent</artifactId>
  <version>3.0.0-RC1</version>
  <relativePath/>
</parent>

One of the more interesting new features in Spring Boot 3 is the support for Prometheus exemplars. Exemplars are references to data outside of the metrics published by an application. They allow linking metrics data to distributed traces. In that case, the published metrics contain a reference to the traceId. In order to enable exemplars for the particular metrics, we need to expose percentiles histograms. We will do that for http.server.requests metric (1). We will also send all the traces to Grafana Cloud by setting the sampling probability to 1.0 (2). Finally, just to verify it works properly we print traceId and spanId in the log line (3).

spring:
  application:
    name: inter-callme-service
  output.ansi.enabled: ALWAYS

management:
  endpoints.web.exposure.include: '*'
  metrics.distribution.percentiles-histogram.http.server.requests: true # (1)
  tracing.sampling.probability: 1.0 # (2)

logging.pattern.console: "%clr(%d{HH:mm:ss.SSS}){blue} %clr(%5p [${spring.application.name:},%X{traceId:-},%X{spanId:-}]){yellow} %clr(:){red} %clr(%m){faint}%n" # (3)

The inter-callme-service exposes the POST endpoint that just returns the message in the reversed order. We don’t need to add here anything, just standard Spring Web annotations.

@RestController
@RequestMapping("/callme")
class CallmeController(private val repository: ConversationRepository) {

   private val logger: Logger = LoggerFactory
       .getLogger(CallmeController::class.java)

   @PostMapping("/call")
   fun call(@RequestBody request: CallmeRequest) : CallmeResponse {
      logger.info("In: {}", request)
      val response = CallmeResponse(request.id, request.message.reversed())
      repository.add(Conversation(request = request, response = response))
      return response
   }

}

Load Balancing with Spring Cloud

In the endpoint exposed by the inter-caller-service we just call the endpoint from inter-callme-service. We use Spring RestTemplate for that. You can also declare Spring Cloud OpenFeign client, but it seems it does not currently support Micrometer Tracing out-of-the-box.

@RestController
@RequestMapping("/caller")
class CallerController(private val template: RestTemplate) {

   private val logger: Logger = LoggerFactory
      .getLogger(CallerController::class.java)

   private var id: Int = 0

   @PostMapping("/send/{message}")
   fun send(@PathVariable message: String): CallmeResponse? {
      logger.info("In: {}", message)
      val request = CallmeRequest(++id, message)
      return template.postForObject("http://inter-callme-service/callme/call",
         request, CallmeResponse::class.java)
   }

}

In this exercise, we will use a static client-side load balancer that distributes traffic across two instances of inter-callme-service. Normally, you would integrate Spring Cloud Load Balancer with service discovery based e.g. on Eureka. However, I don’t want to complicate our demo with external components in the architecture. Assuming we are running inter-callme-service on 55800 and 55900 here’s the load balancer configuration in the application.yml file:

spring:
  application:
    name: inter-caller-service
  cloud:
    loadbalancer:
      cache:
        ttl: 1s
      instances:
        - name: inter-callme-service
          servers: localhost:55800, localhost:55900

Since there is no built-in static load balancer implementation we need to add some code. Firstly, we have to inject configuration properties into the Spring bean.

@Configuration
@ConfigurationProperties("spring.cloud.loadbalancer")
class LoadBalancerConfigurationProperties {

   val instances: MutableList<ServiceConfig> = mutableListOf()

   class ServiceConfig {
      var name: String = ""
      var servers: String = ""
   }

}

Then we need to create a bean that implements the ServiceInstanceListSupplier interface. It just returns a list of ServiceInstance objects that represents all defined static addresses.

class StaticServiceInstanceListSupplier(private val properties: LoadBalancerConfigurationProperties,
                                        private val environment: Environment): 
   ServiceInstanceListSupplier {

   override fun getServiceId(): String = environment
      .getProperty(LoadBalancerClientFactory.PROPERTY_NAME)!!

   override fun get(): Flux<MutableList<ServiceInstance>> {
      val serviceConfig: LoadBalancerConfigurationProperties.ServiceConfig? =
                properties.instances.find { it.name == serviceId }
      val list: MutableList<ServiceInstance> =
                serviceConfig!!.servers.split(",", ignoreCase = false, limit = 0)
                        .map { StaticServiceInstance(serviceId, it) }.toMutableList()
      return Flux.just(list)
   }

}

Finally, we need to enable Spring Cloud Load Balancer for the app and annotate RestTemplate with @LoadBalanced.

@SpringBootApplication
@LoadBalancerClient(value = "inter-callme-service", configuration = [CustomCallmeClientLoadBalancerConfiguration::class])
class InterCallerServiceApplication {

    @Bean
    @LoadBalanced
    fun template(builder: RestTemplateBuilder): RestTemplate = 
       builder.build()

}

Here’s the client-side load balancer configuration. We are providing our custom StaticServiceInstanceListSupplier implementation as a default ServiceInstanceListSupplier. Then we set RandomLoadBalancer as a default implementation of a load-balancing algorithm.

class CustomCallmeClientLoadBalancerConfiguration {

   @Autowired
   lateinit var properties: LoadBalancerConfigurationProperties

   @Bean
   fun clientServiceInstanceListSupplier(
      environment: Environment, context: ApplicationContext):     
      ServiceInstanceListSupplier {
        val delegate = StaticServiceInstanceListSupplier(properties, environment)
        val cacheManagerProvider = context
           .getBeanProvider(LoadBalancerCacheManager::class.java)
        return if (cacheManagerProvider.ifAvailable != null) {
            CachingServiceInstanceListSupplier(delegate, cacheManagerProvider.ifAvailable)
        } else delegate
    }

   @Bean
   fun loadBalancer(environment: Environment, 
      loadBalancerClientFactory: LoadBalancerClientFactory):
            ReactorLoadBalancer<ServiceInstance> {
        val name: String? = environment.getProperty(LoadBalancerClientFactory.PROPERTY_NAME)
        return RandomLoadBalancer(
           loadBalancerClientFactory
              .getLazyProvider(name, ServiceInstanceListSupplier::class.java),
            name!!
        )
    }
}

Testing Observability with Spring Boot

Let’s see how it works. In the first step, we are going to run two instances of inter-callme-service. Since we set a static value of the listen port we need to override the property server.port for each instance. We can do it with the env variable SERVER_PORT. Go to the inter-communication/inter-callme-service directory and run the following commands:

$ SERVER_PORT=55800 mvn spring-boot:run
$ SERVER_PORT=55900 mvn spring-boot:run

Then, go to the inter-communication/inter-caller-service directory and run a single instance on the default port 8080:

$ mvn spring-boot:run

Then, let’s call our endpoint POST /caller/send/{message} several times with parameters, for example:

$ curl -X POST http://localhost:8080/caller/call/hello1

Here are the logs from inter-caller-service with the highlighted value of the traceId parameter:

Let’s take a look at the logs from inter-callme-service. As you see the traceId parameter is the same as the traceId for that request on the inter-caller-service side.

Here are the logs from the second instance of inter-callme-service:

ou could also try the same exercise with Spring Cloud OpenFeign. It is configured and ready to use. However, for me, it didn’t propagate the traceId parameter properly. Maybe, it is the case with the current non-GA versions of Spring Boot and Spring Cloud.

Let’s verify another feature – Prometheus exemplars. In order to do that we need to call the /actuator/prometheus endpoint with the Accept header that is asking for the OpenMetrics format. This is the same header Prometheus will use to scrape the metrics.

$ curl -H 'Accept: application/openmetrics-text; version=1.0.0' http://localhost:8080/actuator/prometheus

As you see several metrics for the result contain traceId and spanId parameters. These are our exemplars.

Install and Configure Grafana Agent

Our sample apps are ready. Now, the main goal is to send all the collected observables to our account on Grafana Cloud. There are various ways of sending metrics, logs, and traces to the Grafana Stack. In this article, I will show you how to use the Grafana Agent for that. Firstly, we need to install it. You can find detailed installation instructions depending on your OS here. Since I’m using macOS I can do it with Homebrew:

$ brew install grafana-agent

Before we start the agent, we need to prepare a configuration file. The location of that file also depends on your OS. For me it is $(brew --prefix)/etc/grafana-agent/config.yml. The configuration YAML manifests contain information on how we want to collect and send metrics, traces, and logs. Let’s begin with the metrics. Inside the scrape_configs section, we need to set a list of endpoints for scraping (1) and a default path (2). Inside the remote_write section, we have to pass our Grafana Cloud instance auth credentials (3) and URL (4). By default, Grafana Agent does not send exemplars. Therefore we need to enable it with the send_exemplars property (5).

metrics:
  configs:
    - name: integrations
      scrape_configs:
        - job_name: agent
          static_configs:
            - targets: ['127.0.0.1:55800','127.0.0.1:55900','127.0.0.1:8080'] # (1)
          metrics_path: /actuator/prometheus # (2)
      remote_write:
        - basic_auth: # (3)
            password: <YOUR_GRAFANA_CLOUD_TOKEN>
            username: <YOUR_GRAFANA_CLOUD_USER>
          url: https://prometheus-prod-01-eu-west-0.grafana.net/api/prom/push # (4)
          send_exemplars: true # (5)
  global:
    scrape_interval: 60s

You can find all information about your instance of Prometheus in the Grafana Cloud dashboard.

In the next step, we prepare a configuration for collecting and sending logs to Grafana Loki. The same as before we need to set auth credentials (1) and the URL (2) of our Loki instance. The most important thing here is to pass the location of log files (3).

logs:
  configs:
    - clients:
        - basic_auth: # (1)
            password: <YOUR_GRAFANA_CLOUD_TOKEN>
            username: <YOUR_GRAFANA_CLOUD_USER>
          # (2)
          url: https://logs-prod-eu-west-0.grafana.net/loki/api/v1/push
      name: springboot
      positions:
        filename: /tmp/positions.yaml
      scrape_configs:
        - job_name: springboot-json
          static_configs:
            - labels:
                __path__: ${HOME}/inter-*/spring.log # (3)
                job: springboot-json
              targets:
                - localhost
      target_config:
        sync_period: 10s

By default, Spring Boot logs only to the console and does not write log files. In our case, the Grafana Agent reads log lines from output files. In order to write log files, we need to set a logging.file.name or logging.file.path property. Since there are two instances of inter-callme-service we need to distinguish somehow their log files. We will use the server.port property for that. The logs inside files are stored in JSON format.

logging:
  pattern:
    console: "%clr(%d{HH:mm:ss.SSS}){blue} %clr(%5p [${spring.application.name:},%X{traceId:-},%X{spanId:-}]){yellow} %clr(:){red} %clr(%m){faint}%n"
    file: "{\"timestamp\":\"%d{HH:mm:ss.SSS}\",\"level\":\"%p\",\"traceId\":\"%X{traceId:-}\",\"spanId\":\"%X{spanId:-}\",\"appName\":\"${spring.application.name}\",\"message\":\"%m\"}%n"
  file:
    path: ${HOME}/inter-callme-service-${server.port}

Finally, we will configure trace collecting. Besides auth credentials and URL of the Grafana Tempo instance, we need to enable OpenZipkin receiver (1).

traces:
  configs:
    - name: default
      remote_write:
        - endpoint: tempo-eu-west-0.grafana.net:443
          basic_auth:
            username: <YOUR_GRAFANA_CLOUD_USER>
            password: <YOUR_GRAFANA_CLOUD_TOKEN>
      # (1)
      receivers:
        zipkin:

Then, we can start the agent with the following command:

$ brew services restart grafana-agent

Grafana agent contains a Zipkin collector that listens on the default port 9411. There is also an HTTP API exposed outside the agent on the port 12345 for verifying agent status.

For example, we can use Grafana Agent HTTP API to verify how many log files it is monitoring. To do that just call the endpoint GET agent/api/v1/logs/targets. As you see, for me, it is monitoring three files. So that’s what we exactly wanted to achieve since there are two running instances of inter-callme-service and a single instance of inter-caller-service.

Visualize Spring Boot Observability with Grafana Stack

One of the most important advantages of Grafana Cloud in our exercise is that we have all the required things configured. After you navigate to the Grafana dashboard you can display a list of available data sources. As you see, there are Loki, Prometheus, and Tempo already configured.

By default, Grafana Cloud enables exemplars in the Prometheus data source. If you are running Grafana by yourself, don’t forget to enable it on your Prometheus data source.

Let’s start with the logs. We will analyze exactly the same logs as in the section “Testing Observability with Spring Boot”. We will get all the logs sent by the Grafana Agent. As you probably remember, we formatted all the logs as JSON. Therefore, we will parse them using the Json parser on the server side. Thanks to that, we would be able to filter by all the log fields. For example, we can use the traceId label a filter expression: {job="springboot-json"} | json | traceId = 1bb1d7d78a5ac47e8ebc2da961955f87.

Here’s a full list of logs without any filtering. The highlighted lines contain logs of two analyzed traces.

In the next step, we will configure Prometheus metrics visualization. Since we enabled percentile histograms for the http.server.requests metrics, we have multiple buckets represented by the http_server_requests_seconds_bucket values. A set of multiple buckets _bucket with a label le which contains a count of all samples whose values are less than or equal to the numeric value contained in the le label. We need to count histograms for 90% and 60% percent of requests. Here are our Prometheus queries:

Here’s our histogram. Exemplars are shown as green diamonds.

When you hover over the selected exemplar, you will see more details. It includes, for example, the traceId value.

Finally, the last part of our exercise. We would like to analyze the particular traces with Grafana Tempo. The only thing you need to do is to choose the grafanacloud-*-traces data source and set the value of the searched traceId. Here’s a sample result.

Final Thoughts

The first GA release of Spring Boot 3 is just around the corner. Probably one of the most important things you will have to handle during migration from Spring Boot 2 is observability. In this article, you can find a detailed description of the current Spring Boot approach. If you are interested in Spring Boot, it’s worth reading about its best practices for building microservices in this article.

The post Spring Boot 3 Observability with Grafana appeared first on Piotr's TechBlog.

If you are also interested in Spring Boot and microservices you can my article about best practices for building microservices using the Spring Boot framework.

Source Code

If you would like to try this exercise yourself, you may always take a look at my source code. In order to do that, you need to clone my GitHub repository. This time we have two apps since there are backend and frontend. If you would like to run the Spring Boot directly from the code also clone the following repository. After that, just follow my instructions.

Prerequisites

Before we begin, we need to install some tools. Of course, you need npm to build and run our React app. I used a version 8.19.2 of npm. In order to run the Spring Boot backend app locally, you should have Docker or Maven with JDK.

Assuming you have Maven and JDK and you want to run it directly from the code just execute the following command:

$ mvn spring-boot:run

With Docker just run the app using the latest image from my registry:

$ docker run -d --name sample-spring-boot -p 8080:8080 \
  piomin/sample-spring-kotlin-microservice:latest

After running the Spring Boot app you can display the list of available REST endpoints by opening the Swagger UI page http://localhost:8080/swagger-ui.html.

Micro Frontend with React – Architecture

Here’s our architecture. First, we are going to run the Spring Boot app and expose it on the local port 8080. Then we will run the React app that listens on a port 3000 and communicates with the backend over REST API.

We will use the following React libraries:

• MUI (Material UI for React) – React UI components, which implement Google’s Material Design 
• React Redux – an implementation of Redux JS for React to centralize the state of apps using the store component 
• Redux Saga – an intuitive Redux side effect manager that allows us to dispatch an action asynchronously and connect to the Redux store
• Axios – the promise-based HTTP client for the browser and node.js
• React Router – declarative, client-side routing for React

Later, I will show you how those libraries will help you to organize your project. For now, let’s just take a look at the structure of our source code. There are three components: Home displays the list of all persons, AddPerson allows adding of a new person, and GetPerson displays the details of a selected person.

Let’s take a look at our package.json file.

{
  "name": "react",
  "version": "1.0.0",
  "description": "React Micro Frontend",
  "keywords": [
    "react",
    "starter"
  ],
  "main": "src/index.js",
  "dependencies": {
    "@emotion/react": "11.10.4",
    "@emotion/styled": "11.10.4",
    "@mui/material": "5.10.8",
    "@mui/x-data-grid": "latest",
    "axios": "1.1.2",
    "react": "18.2.0",
    "react-dom": "18.2.0",
    "react-redux": "8.0.2",
    "react-router-dom": "6.4.2",
    "react-scripts": "5.0.1",
    "redux": "4.2.0",
    "redux-saga": "1.2.1"
  },
  "devDependencies": {
    "@babel/runtime": "7.13.8",
    "typescript": "4.1.3"
  },
  "scripts": {
    "start": "react-scripts start",
    "build": "react-scripts build",
    "test": "react-scripts test --env=jsdom",
    "eject": "react-scripts eject"
  },
  "browserslist": [
    ">0.2%",
    "not dead",
    "not ie <= 11",
    "not op_mini all"
  ]
}

We can create React apps using two different approaches. The first of them is based on functional components, while the second is based on class components. I won’t compare them, since these are React basics and you can read more about them in tutorials. I’ll choose the first approach based on functions.

Communicate with the Backend over REST API

Let’s start unusual – with the REST client implementation. We use the Axios library for communication over HTTP and redux-saga for watching and propagating events (actions). For each type of action, there are two functions. The “watch” function waits on dispatched action. Then it just calls another function for performing HTTP calls. All the “watch” functions are our sagas (in fact they implement the popular SAGA pattern), so we need to export them outside the module. The Axios client is pretty intuitive. We can call for example GET endpoint without any parameters or POST JSON payload. Here’s the implementation available in the sagas/index.js file.

import { call, put, takeEvery, all } from "redux-saga/effects";
import axios from "axios";
import { 
  ADD_PERSON, 
  ADD_PERSON_FAILURE, 
  ADD_PERSON_SUCCESS, 
  GET_ALL_PERSONS, 
  GET_ALL_PERSONS_FAILURE, 
  GET_ALL_PERSONS_SUCCESS, 
  GET_PERSON_BY_ID, 
  GET_PERSON_BY_ID_FAILURE, 
  GET_PERSON_BY_ID_SUCCESS } from "../actions/types";

const apiUrl = "http://localhost:8080/persons";

function* getPersonById(action) {
  try {
    const person = yield call(axios, apiUrl + "/" + action.payload.id);
    yield put({ type: GET_PERSON_BY_ID_SUCCESS, payload: person });
  } catch (e) {
    yield put({ type: GET_PERSON_BY_ID_FAILURE, message: e.message });
  }
}

function* getAllPersons(action) {
  try {
    const persons = yield call(axios, apiUrl);
    yield put({ type: GET_ALL_PERSONS_SUCCESS, payload: persons });
  } catch (e) {
    yield put({ type: GET_ALL_PERSONS_FAILURE, message: e.message });
  }
}

function* addPerson(action) {
  try {
    const person = yield call(axios, {
      method: "POST",
      url: apiUrl,
      data: action.payload
    });
    yield put({ type: ADD_PERSON_SUCCESS, payload: person });
  } catch (e) {
    yield put({ type: ADD_PERSON_FAILURE, message: e.message });
  }
}

function* watchGetPerson() {
  yield takeEvery(GET_PERSON_BY_ID, getPersonById);
}

function* watchGetAllPersons() {
  yield takeEvery(GET_ALL_PERSONS, getAllPersons);
}

function* watchAddPerson() {
  yield takeEvery(ADD_PERSON, addPerson);
}

export default function* rootSaga() {
  yield all([watchGetPerson(), watchGetAllPersons(), watchAddPerson()]);
}

Redux Saga works asynchronously. It listens for the action and propagates a new event after receiving a response from the backend. There are three actions handled by the component visible above: GET /persons, GET /persons/{id}, and POST /persons. Depending on the result they emit *_SUCCESS or *_FAILURE events. Here’s a dictionary in the file actions/types.js with all the events handled/emitted by our app:

export const GET_ALL_PERSONS = "GET_ALL_PERSONS";
export const GET_ALL_PERSONS_SUCCESS = "GET_ALL_PERSONS_SUCCESS";
export const GET_ALL_PERSONS_FAILURE = "GET_ALL_PERSONS_FAILURE";

export const GET_PERSON_BY_ID = "GET_PERSON_BY_ID";
export const GET_PERSON_BY_ID_SUCCESS = "GET_PERSON_BY_ID_SUCCESS";
export const GET_PERSON_BY_ID_FAILURE = "GET_PERSON_BY_ID_FAILURE";

export const ADD_PERSON = "ADD_PERSON";
export const ADD_PERSON_SUCCESS = "ADD_PERSON_SUCCESS";
export const ADD_PERSON_FAILURE = "ADD_PERSON_FAILURE";

Also, let’s take a look a the action/index.js file. It contains three functions for dispatching actions. Those functions are then used by the React components. Each action has a type field and payload. The payload may e.g. contain a body that is sent as a JSON to the backend (1).

import { 
    ADD_PERSON, 
    GET_PERSON_BY_ID, 
    GET_ALL_PERSONS } from "./types";

export function getPersonById(payload) {
  return { type: GET_PERSON_BY_ID, payload };
}

export function getAllPersons(payload) {
  return { type: GET_ALL_PERSONS, payload };
}

export function addPerson(payload) { // (1)
  return { type: ADD_PERSON, payload };
}

Configure React Redux and Redux Saga

To make everything work properly we need to prepare some configurations. In the previous step, we have already created an implementation of sagas responsible for handling asynchronous actions dispatched by the React components. Now, we need to configure the Redux Saga library to handle those actions properly. In the same step, we also create a Redux store to handle the current global state of the React app. The configuration is available in the store/index.js file.

import { createStore, applyMiddleware, compose } from "redux";
import createSagaMiddleware from "redux-saga";
import rootReducer from "../reducers/index";
import rootSaga from "../sagas/index";

const sagaMiddleware = createSagaMiddleware(); // (1)
const composeEnhancers = window.__REDUX_DEVTOOLS_EXTENSION_COMPOSE__ || compose; // (2)

const store = createStore(
  rootReducer,
  composeEnhancers(applyMiddleware(sagaMiddleware))
); // (3)

sagaMiddleware.run(rootSaga); // (4)

export default store;

The only way to change the global state of the app is to take action. In order to handle the actions, let’s create a component called sagaMiddleware (1). Then we need to register sagas (4) and connect the store to the redux-saga middleware (3). We will also enable Redux Dev Tools for the Saga middleware (2). It would be helpful during development. The store requires reducers. That’s a very important part of the Redux concept. In redux nomenclature “reducer” is a function that takes a current state value and an action object that described “what happened”. As a result, it returns a new state value.

Here’s our reducer implementation provided in the reducers/index.js file:

import { 
  ADD_PERSON,
  ADD_PERSON_SUCCESS, 
  GET_ALL_PERSONS_SUCCESS, 
  GET_PERSON_BY_ID_SUCCESS } from "../actions/types";

const initialState = {
  persons: [],
  person: {},
  newPersonId: null,
}; // (1)

function rootReducer(state = initialState, action) {
  switch(action.type) {
    case GET_ALL_PERSONS_SUCCESS: // (2)
      return {
        ...state,
        persons: action.payload.data
      };
    case ADD_PERSON:
      return {
        ...state,
        person: action.payload.data,
        newPersonId: null
      };
    case ADD_PERSON_SUCCESS: // (3)
      return {
        ...state,
        person: {
          name: "",
          gender: "",
          age: 0
        },
        newPersonId: action.payload.data.id
      };
    case GET_PERSON_BY_ID_SUCCESS: // (4)
      return {
        ...state,
        person: action.payload.data
      };
    default:
      return state;
  }
}

export default rootReducer;

Let’s analyze what happened here. We need to define the initial state of the store for our micro frontend React app (1). It contains the list of all persons retrieved from the backend (persons), the current displayed or newly added person (person) and the id of a new person (newPersonId). For the GET_ALL_PERSONS action it puts the elements received from the backend API to the persons array (2). For the ADD_PERSON result, it resets the state of the person object and set the id of the new person in the newPersonId field (3). Finally, we set the current person details in the person object for the GET_PERSON_BY_ID result (4).

Create React Components

We have already created all the components responsible for handling actions, the state store, and communicating with the backend. It’s time to create our first React component. We will start with the Home component responsible for getting and displaying a list of all persons. Here’s the full code of the component available in components/Home.js. Let’s analyze step-by-step what happened here. The order of further steps is logical.

import { connect } from "react-redux";
import React, { useEffect } from "react";
import { useNavigate } from "react-router-dom"; // (9)

import { Button, Stack } from "@mui/material";
import { DataGrid } from '@mui/x-data-grid';

import { getAllPersons } from "../actions/index"; // (4)

// (7)
const columns = [
  { field: 'id', headerName: 'ID', width: 70 },
  { field: 'name', headerName: 'Name', width: 130, editable: true },
  { field: 'age', headerName: 'Age', type: 'number', width: 90, editable: true },
  { field: 'gender', headerName: 'Gender', width: 100 },
];

function Home({ getAllPersons, persons }) { // (5)

  let navigate = useNavigate(); // (10)

  // (8)
  useEffect(() => {
    getAllPersons()
  }, []);

  function handleClick() { // (11)
    navigate("/add");
  }

  function handleSelection(p, e) { // (13)
    navigate("/details/" + p.id);
  }

  return(
    <Stack spacing={2}>
      <Stack direction="row">
        <Button variant="outlined" onClick={handleClick}>Add person</Button>
      </Stack>
      <div style={{ height: 400, width: '100%' }}> // (6)
        <DataGrid
          rows={persons}
          columns={columns}
          pageSize={5}
          onRowDoubleClick={handleSelection} // (12)
        />
      </div>
    </Stack>
  );
}

function mapStateToProps(state) { // (2)
  return {
    persons: state.persons,
  };
}

function mapDispatchToProps(dispatch) { // (3)
  return {
    getAllPersons: () => dispatch(getAllPersons({})),
  };
}

export default connect(mapStateToProps, mapDispatchToProps)(Home); // (1)

(1) – we need to connect our component to the Redux store. The react-redux connect method takes two input arguments mapStateToProps and mapDispatchToProps

(2) – the mapStateToProps is used for selecting the part of the data from the store that the connected component needs. It’s frequently referred to as just mapState for short. The Home component requires the persons array from the global state store

(3) – as the second argument passed into connect, mapDispatchToProps is used for dispatching actions to the store – dispatch is a function of the Redux store. You can call store.dispatch to dispatch an action. This is the only way to trigger a state change. Since we just need to dispatch the GET_ALL_PERSONS action in the Home component we define a single action there

(4) – we need to import the action definition

(5) – the actions and state fields mapped by the connect method need to be declared as the component props

(6) – we use the Material DataGrid component to display the table with persons. It takes the persons prop as the input argument. We also need to define a list of table columns (7).

(7) – the definition of columns contained by the DataGrid component. It displays the id, name, age and gender fields of each person on the list.

(8) – with the React useEffect method we dispatch the GET_ALL_PERSONS action on load. In fact, we are just calling the getAllPersons() function defined within the actions, which creates and fires events asynchronously

(9) – from the Home component we can navigate to the other app pages represented by two other components AddPerson and GetPerson. In order to do that we first need to import the useNavigate method provided by React Router.

(10) – let’s call the useNavigate method declared in the previous step to get a handle to the navigate component

(11) – there is a Material Button on the page that redirects us the /add context handled by the AddPerson component

(12) – firstly let’s add the onRowDoubleClick listener to our DataGrid. It fires after you double-click on the selected row from the table

(13) – then we get the id field of the row and navigate to the /details/:id context.

Configure React App and Routing

That could be our first step. However, now we can analyze from the perspective of all previously created components or definitions as a final part of our configuration. We need to import the Redux store definition (1) and our React components (2). We also need to configure routing for our three components (3) using React Router library. Especially the last path is interesting. We use a dynamic parameter based on the person id field. Finally, let’s set the store and router providers (4).

import React from "react";
import { createRoot } from "react-dom/client";
import { Provider } from "react-redux";
import {
  createBrowserRouter,
  RouterProvider
} from "react-router-dom";
import store from "./store/index"; // (1)
import Home from "./components/Home"; // (2)
import AddPerson from "./components/AddPerson";
import GetPerson from "./components/GetPerson";

const root = document.getElementById("root");
const rootReact = createRoot(root);

const router = createBrowserRouter([
  {
    path: "/",
    element: <Home />,
  },
  {
    path: "/add",
    element: <AddPerson />,
  },
  {
    path: "/details/:id",
    element: <GetPerson />,
  },
]); // (3)

rootReact.render(
  <Provider store={store}>
    <RouterProvider router={router} />
  </Provider>
); // (4)

Let’s build the app by executing the following command:

$ npm install

Now, we can run our micro frontend React app with the following command:

$ npm start

Here’s our app home page:

Add and Get Data in React Micro Frontend

There are two other components responsible for adding (AddPerson) and getting (GetPerson) data. Let’s start with the AddPerson component. The logic of that component is pretty similar to the previously described Home component. We need to import the addPerson method form actions (1). We also use person and newPersonId field from the state store (2). The ADD_PERSON action is dispatched on the “Save” button clicked (3). After adding a new person we are displaying a message with the id generated by the backend app (4).

import { connect } from "react-redux";
import { Form } from "react-router-dom";
import { TextField, Button, MenuItem, Alert, Grid } from "@mui/material"

import { addPerson } from "../actions/index"; // (1)

function AddPerson({ addPerson, person, newPersonId }) { // (2)

  function handleChangeName(e) {
    person.name = e.target.value;
  }

  function handleChangeAge(e) {
    person.age = e.target.value;
  }

  function handleChangeGender(e) {
    person.gender = e.target.value;
  }

  function handleClick(e) {
    addPerson(person); // (3)
  }

  return(
    <Form method="post">
      <Grid container spacing={2} direction="column">
        <Grid item xs={6}> // (4)
          {newPersonId != null ?
          <Alert variant="filled" severity="success">New person added: {newPersonId}</Alert> : ""
          }
        </Grid>
        <Grid item xs={3}>
          <TextField id="name" label="Name" variant="outlined" onChange={handleChangeName} value={person?.name} />
        </Grid>
        <Grid item xs={3}>
          <TextField id="gender" select label="Gender" onChange={handleChangeGender} value={person?.gender} >
            <MenuItem value={'MALE'}>Male</MenuItem>
            <MenuItem value={'FEMALE'}>Female</MenuItem>
          </TextField>
        </Grid>
        <Grid item xs={3}>
          <TextField id="age" label="Age" inputProps={{ inputMode: 'numeric' }} onChange={handleChangeAge} value={person?.age} />
        </Grid>
        <Grid item xs={3}>
          <Button variant="outlined" onClick={handleClick}>Save</Button>
        </Grid>
      </Grid>
    </Form>
  );
}

function mapStateToProps(state) {
    return {
      person: state.person,
      newPersonId: state.newPersonId,
    };
  }
  
function mapDispatchToProps(dispatch) {
  return {
    addPerson: (payload) => dispatch(addPerson(payload)),
  };
}

export default connect(mapStateToProps, mapDispatchToProps)(AddPerson);

Here’s our page for adding a new person:

Just click the “SAVE” button. After a successful operation you will see the following message on the same page:

We can back to the list. As you see our new person is there:

Now we double-click on the selected row. I would probably need to work on the look of that component But it works fine – it displays the details of the person with the id equal to 4.

Let’s take a look at the code of the component responsible for displaying those details. We need to import the getPersonById method from actions (1). The component dispatches the GET_PERSON_BY_ID action on the page load (2). It takes the id parameter from the route context path /details/:id with the React Router useParams method (3). Then it just displays all the current person fields (4).

import { connect } from "react-redux";
import React, { useEffect } from "react";
import { useParams } from "react-router-dom";
import { Paper, Avatar, Grid } from "@mui/material"

import { getPersonById } from "../actions/index"; // (1)

function GetPerson({ getPersonById, person }) {

  let { id } = useParams(); // (3)

  // (2)
  useEffect(() => {
    getPersonById({id: id})
  }, []);

  // (4)
  return(
    <Grid container spacing={2} direction="column">
      <Grid item direction="row">
        <Grid item><Avatar>U</Avatar></Grid> 
        <Grid item>USER DETAILS</Grid>
      </Grid>
      <Grid item xs={3}>
        <Paper>Name: <b>{person?.name}</b></Paper>
      </Grid>
      <Grid item xs={3}>
        <Paper>Gender: <b>{person?.gender}</b></Paper>
      </Grid>
      <Grid item xs={3}>
        <Paper>Age: <b>{person?.age}</b></Paper>
      </Grid>
    </Grid>
    
  )
}

function mapStateToProps(state) {
    return {
      person: state.person,
    };
  }
  
function mapDispatchToProps(dispatch) {
  return {
    getPersonById: (payload) => dispatch(getPersonById(payload)),
  };
}

export default connect(mapStateToProps, mapDispatchToProps)(GetPerson);

Final Thoughts

I read some tutorials about React, but I didn’t find any that is providing detailed, step-by-step instructions on how to build a micro frontend that communicates with the backend over REST API. Some of them were too complicated, some were too basic or outdated. My point is to give you an up-to-date receipt on how to build a micro-frontend using the most interesting and useful libraries that help you organize your project well.

The post Micro Frontend with React appeared first on Piotr's TechBlog.

Introduction

First of all, let’s recap the approach described in the previous article. We used Kafka Streams to process order transactions on the order-service side. To handle orders coming to the stock-service and payment-service we used a standard Spring @KafkaListener. There are also two databases – a single database per every service. The stock-service stores data related to the number of available products and updates them after receiving an order. The same with the payment-service. It updates the customer’s account on every single order. Both applications receive orders from Kafka topic. They send responses to other topics. But just to simplify, we will skip it as shown in the figure below. We treat the Kafka orders topic as a stream of events and also as a table with the latest order’s status.

What may go wrong with that approach? In fact, we have two data sources here. We use Kafka as the order store. On the other hand, there are SQL databases (in my case H2, but you can use any other) that store stock and payment data. Once we send an order with a reservation to the Kafka topic, we need to update a database. Since Kafka does not support XA transactions, it may result in data inconsistency. Of course, Kafka doesn’t support XA transactions the same as many other systems including e.g. RabbitMQ.

The question is what can we do with that? One of the possible options you may use is an approach called Change Data Capture (CDC) with the outbox pattern. CDC identifies and tracks changes to data in a database. Then it may emit those changes as events and send them, for example to the Kafka topic. I won’t go into the details of that process. If you are interested in you may read this article written by Gunnar Morling.

Architecture with Kafka Streams

The approach I will describe today is fully based on the Kafka Streams. We won’t use any SQL databases. When the order-service sends a new order its id is the message key. With Kafka Streams, we may change a message key in the stream. It results in creating new topics and repartitioning. With new message keys, we may perform calculations just for the specific customerId or productId. The result of such calculation may be saved in the persistent store. For example, Kafka automatically creates and manages such state stores when you are calling stateful operations like count() or aggregate(). We will aggregate the orders related to the particular customer or product. Here’s the illustration of our architecture. Here’s the visualization of our process.

Now, let’s consider a scenario for the payment-service in details. In the incoming stream of orders the payment-service calls the selectKey() operation. It changes the key from the order’s id into the order’s customerId. Then it groups all the orders by the new key and invokes the aggregate() operation. In the aggregate() method it calculates the available amount and reserved amount based on the order’s price and status (whether it is a new order or a confirmation order). If there are sufficient funds on the customer account it sends the ACCEPT order to the payment-orders topic. Otherwise, it sends the REJECT order. Then the order-service process responses by joining streams from payment-orders and stock-orders by the order’s id. As the result, it sends a confirmation or a rollback order.

Finally, let’s proceed to the implementation!

Source Code

If you would like to try it by yourself, you may always take a look at my source code. In order to do that you need to clone my GitHub repository. Then switch to the streams-full branch. After that, you should just follow my instructions.

Aggregation with Kafka Streams

Let’s begin with the payment-service. The implementation of KStream in not complicated here. In the first step (1), we invoke the selectKey() method and get the customerId value of the Order object as a new key. Then we call groupByKey() method (2) to receive KGroupedStream as a result. While we have KGroupedStream we may invoke one of the calculation methods. In that case, we need to use aggregate(), since we have a little bit more advanced calculation than just a simple count (3). The last two steps are just for printing the value after calculation.

@Bean
public KStream<Long, Order> stream(StreamsBuilder builder) {
   JsonSerde<Order> orderSerde = new JsonSerde<>(Order.class);
   JsonSerde<Reservation> rsvSerde = new JsonSerde<>(Reservation.class);
   KStream<Long, Order> stream = builder
      .stream("orders", Consumed.with(Serdes.Long(), orderSerde))
      .peek((k, order) -> LOG.info("New: {}", order));

   KeyValueBytesStoreSupplier customerOrderStoreSupplier =
      Stores.persistentKeyValueStore("customer-orders");

   stream.selectKey((k, v) -> v.getCustomerId()) // (1)
      .groupByKey(Grouped.with(Serdes.Long(), orderSerde)) // (2)
      .aggregate(
         () -> new Reservation(random.nextInt(1000)),
         aggregatorService,
         Materialized.<Long, Reservation>as(customerOrderStoreSupplier)
            .withKeySerde(Serdes.Long())
            .withValueSerde(rsvSerde)) // (3)
      .toStream()
      .peek((k, trx) -> LOG.info("Commit: {}", trx));

   return stream;
}

However, the most important step in the fragment of code visible above is the class called inside the aggregate() method. The aggregate() method takes three input arguments. The first of them indicates the starting value of our compute object. That object represents the current state of the customer’s account. It has two fields: amountAvailable and amountReserved. To clarify, we use that object instead of the entity that stores available and reserved amounts on the customer account. Each customer is represented by the customerId (key) and the Reservation object (value) in Kafka KTable. Just for the test purpose, we are generating the starting value of amountAvailable as a random number between 0 and 1000.

public class Reservation {
   private int amountAvailable;
   private int amountReserved;

   public Reservation() {
   
   }

   public Reservation(int amountAvailable) {
      this.amountAvailable = amountAvailable;
   }

   // GETTERS AND SETTERS ...

}

Ok, let’s take a look at our aggregation method. It needs to implement the Kafka Aggregate interface and its method apply(). It may handle three types of orders. One of them is a confirmation of the order (1). It confirms the distributed transaction, so we just need to cancel a reservation by subtracting the order’s price from the amountReserved field. On the other, in the case of rollback, we need to increase the value of amountAvailable by the order’s price and decrease the value amountRerserved accordingly (2). Finally, if we receive a new order we need to perform a reservation if there are sufficient funds on the customer account, or otherwise, reject an order.

Aggregator<Long, Order, Reservation> aggregatorService = (id, order, rsv) -> {
   switch (order.getStatus()) {
      case "CONFIRMED" -> // (1)
         rsv.setAmountReserved(rsv.getAmountReserved() 
               - order.getPrice());
      case "ROLLBACK" -> { // (2)
         if (!order.getSource().equals("PAYMENT")) {
            rsv.setAmountAvailable(rsv.getAmountAvailable() 
                  + order.getPrice());
            rsv.setAmountReserved(rsv.getAmountReserved() 
                  - order.getPrice());
         }
      }
      case "NEW" -> { // (3)
         if (order.getPrice() <= rsv.getAmountAvailable()) {
            rsv.setAmountAvailable(rsv.getAmountAvailable() 
                  - order.getPrice());
            rsv.setAmountReserved(rsv.getAmountReserved() 
                  + order.getPrice());
            order.setStatus("ACCEPT");
         } else {
            order.setStatus("REJECT");
         }
         template.send("payment-orders", order.getId(), order);
      }
   }
   LOG.info("{}", rsv);
   return rsv;
};

State Store with the Kafka Streams Table

The implementation of the stock-service is pretty similar to the payment-service. With the difference that we count a number of available products on stock instead of available funds on the customer account. Here’s our Reservation object:

public class Reservation {
   private int itemsAvailable;
   private int itemsReserved;

   public Reservation() {
    
   }

   public Reservation(int itemsAvailable) {
      this.itemsAvailable = itemsAvailable;
   }

   // GETTERS AND SETTERS ...

}

The implementation of the aggregation method is also very similar to the payment-service. However, this time, let’s focus on another thing. Once we process a new order we need to send a response to the stock-orders topic. We use KafkaTemplate for that. In the case of payment-service we also send a response, but to the payment-orders topic. The send method from the KafkaTemplate does not block the thread. It returns the ListenableFuture objects. We may add a callback to the send method using it and the result after sending the message (1). Finally, let’s log the current state of the Reservation object (2).

Aggregator<Long, Order, Reservation> aggrSrv = (id, order, rsv) -> {
   switch (order.getStatus()) {
      case "CONFIRMED" -> rsv.setItemsReserved(rsv.getItemsReserved() 
            - order.getProductCount());
      case "ROLLBACK" -> {
         if (!order.getSource().equals("STOCK")) {
            rsv.setItemsAvailable(rsv.getItemsAvailable() 
                  + order.getProductCount());
            rsv.setItemsReserved(rsv.getItemsReserved() 
                  - order.getProductCount());
         }
      }
      case "NEW" -> {
         if (order.getProductCount() <= rsv.getItemsAvailable()) {
            rsv.setItemsAvailable(rsv.getItemsAvailable() 
                  - order.getProductCount());
            rsv.setItemsReserved(rsv.getItemsReserved() 
                  + order.getProductCount());
            order.setStatus("ACCEPT");
         } else {
            order.setStatus("REJECT");
         }
         // (1)
         template.send("stock-orders", order.getId(), order)
            .addCallback(r -> LOG.info("Sent: {}", 
               result != null ? result.getProducerRecord().value() : null),
               ex -> {});
      }
   }
   LOG.info("{}", rsv); // (2)
   return rsv;
};

After that, we are also logging the value of the Reservation object (1). In order to do that we need to convert KTable into KStream and then call the peek method. This log is printed just after Kafka Streams commits the offset in the source topic.

@Bean
public KStream<Long, Order> stream(StreamsBuilder builder) {
   JsonSerde<Order> orderSerde = new JsonSerde<>(Order.class);
   JsonSerde<Reservation> rsvSerde = new JsonSerde<>(Reservation.class);
   KStream<Long, Order> stream = builder
      .stream("orders", Consumed.with(Serdes.Long(), orderSerde))
      .peek((k, order) -> LOG.info("New: {}", order));

   KeyValueBytesStoreSupplier stockOrderStoreSupplier =
      Stores.persistentKeyValueStore("stock-orders");

   stream.selectKey((k, v) -> v.getProductId())
      .groupByKey(Grouped.with(Serdes.Long(), orderSerde))
      .aggregate(() -> new Reservation(random.nextInt(100)), aggrSrv,
         Materialized.<Long, Reservation>as(stockOrderStoreSupplier)
            .withKeySerde(Serdes.Long())
            .withValueSerde(rsvSerde))
      .toStream()
      .peek((k, trx) -> LOG.info("Commit: {}", trx)); // (1)

   return stream;
}

What will happen if you send the test order? Let’s see the logs. You can see the difference in time between processing the message and offset commit. You won’t have any problems with that until your application is running or it has been stopped gracefully. But if you, for example, kill the process using the kill -9 command? After restart, our application will receive the same messages once again. Since we use KafkaTemplate to send the response to the stock-orders topic, we need to commit the offset as soon as possible.

What can we do to avoid such problems? We may override the default value (30000) of the commit.interval.ms Kafka Streams property. If you set it to 0, it commits immediately after processing finishes.

spring.kafka:  
  streams:
    properties:
      commit.interval.ms: 0

On the other hand, we can also set the property processing.guarantee to exactly_once. It also changes the default value of commit.interval.ms to 100ms and enables idempotence for a producer. You can read more about it here in Kafka documentation.

spring.kafka:  
  streams:
    properties:
      processing.guarantee: exactly_once

Running Kafka on Upstash

For the purpose of today’s exercise, we will use a serverless Kafka cluster on Upstash. You can create it with a single click. If you would like to test JAAS authentication for your application I’ve got good news The authentication on that cluster is enabled by default. You can find and copy username and password from the cluster’s main panel.

Now, let’s configure Kafka connection settings and credentials for the Spring Boot application. There is a developer free tier on Upstash up to 10k messages per day. It will be enough for our tests.

spring.kafka:
  bootstrap-servers: topical-gar-11460-us1-kafka.upstash.io:9092
  properties:
    security.protocol: SASL_SSL
    sasl.mechanism: SCRAM-SHA-256
    sasl.jaas.config: org.apache.kafka.common.security.scram.ScramLoginModule required username="${USERNAME}" password="${PASSWORD}";

With Upstash you can easily display a list of topics. In total, there are 10 topics used in our sample system. Three of them are used directly by the Spring Boot applications, while the rest of them by the Kafka Streams in order to process stateful operations.

After starting the order-service application we can call its REST endpoint to create and send an order to the Kafka topic.

private static final Logger LOG = 
   LoggerFactory.getLogger(OrderController.class);
private AtomicLong id = new AtomicLong();
private KafkaTemplate<Long, Order> template;

@PostMapping
public Order create(@RequestBody Order order) {
   order.setId(id.incrementAndGet());
   template.send("orders", order.getId(), order);
   LOG.info("Sent: {}", order);
   return order;
}

Let’s call the endpoint using the following curl command. You can use any customerId or productId you want.

$ curl -X 'POST' \
  'http://localhost:8080/orders' \
  -H 'Content-Type: application/json' \
  -d '{
    "customerId": 20,
    "productId": 20,
    "productCount": 2,
    "price": 10,
    "status": "NEW"
  }'

All three sample applications use Kafka Streams to process distributed transactions. Once the order is accepted by both stock-service and payment-service you should see the following entry in the order-service logs.

You can easily simulate rejection of transactions with Kafka Streams just by setting e.g. productCount higher than the value generated by the product-service as available items.

With Upstash UI you can also easily verify the number of messages incoming to the topics. Let’s see the current statistics for the orders topic.

Final Thoughts

In order to fully understand what happens in this example, you should be also familiar with the Kafka Streams threading model. It is worth reading the following article, which explains it in a clean manner. First of all, each stream partition is a totally ordered sequence of data records and maps to a Kafka topic partition. It means, that even if we have multiple orders at the same time related to e.g. same product, they are all processed sequentially since they have the same message key (productId in that case).

Moreover, by default, there is only a single stream thread that handles all the partitions. You can see this in the logs below. However, there are stream tasks that act as the lowest-level units of parallelism. As a result, stream tasks can be processed independently and in parallel without manual intervention.

I hope this article helps you to better understand Kafka Streams. I just wanted to give you a simple example of how you can use Kafka Streams with Saga transactions in order to simplify your current architecture.

The post Deep Dive into Saga Transactions with Kafka Streams and Spring Boot appeared first on Piotr's TechBlog.

This time I won’t tell you about Istio basics. Although our configuration is not complicated you may read the following introduction before we start.

Source Code

If you would like to try it by yourself, you may always take a look at my source code. In order to do that you need to clone my GitHub repository. Then you should go to the mesh-with-db directory. After that, you should just follow my instructions

Service Mesh Architecture

Let’s start with our microservices architecture. There are two applications: person-app and insurance-app. As you probably guessed, the person-app stores and returns information about insured people. On the other hand, the insurance-app keeps insurances data. Each service has a separate database. We deploy the person-app in two versions. The v2 version contains one additional field externalId.

The following picture illustrates our scenario. Istio splits traffic between two versions of the person-app. By default, it splits the traffic 50% to 50%. If it receives the X-Version header in the request it calls the particular version of the person-app. Of course, the possible values of the header are v1 or v2.

Distributed Tracing with Istio

Istio generates distributed trace spans for each managed service. It means that every request sent inside the Istio will have the following HTTP headers:

So, every single request incoming from the Istio gateway contains X-B3-SpanId, X-B3-TraceId, and some other B3 headers. The X-B3-SpanId indicates the position of the current operation in the trace tree. On the other hand, every span in a trace should share the X-B3-TraceId header. At first glance, you can feel surprised that Istio does not propagate B3 headers in client calls. To clarify, if one service communicates with another service using e.g. REST client, you will see two different traces. The first of them is related to the API endpoint call, while the second with the client call of another API endpoint. That’s not exactly what we would like to achieve, right?

Let’s visualize our problem. If you call the insurance-app through the Istio gateway you will have the first trace in Jaeger. During that call the insurance-app calls endpoint from the person-app using Quarkus REST client. That’s another separate trace in Jeager. Our goal here is to propagate all required B3 headers to the person-app also. You can find a list of required headers here in the Istio documentation.

Of course, that’s not the only thing we will do today. We will also prepare Istio rules in order to simulate latency in our communication. It is a good scenario to use the tracing tool. Also, I’m going to show you how to easily deploy microservice on Kubernetes using Quarkus features for that.

I’m running Istio and Jaeger on OpenShift. More precisely, I’m using OpenShift Service Mesh that is the RedHat’s SM implementation based on Istio. I doesn’t have any impact on the exercise, so you as well repeat all the steps on Kubernetes.

Create Microservices with Quarkus

Let’s begin with the insurance-app. Here’s the class responsible for the REST endpoints implementation. There are several methods there. However, the most important for us is the getInsuranceDetailsById method that calls the GET /persons/{id} endpoint in the person-app. In order to use the Quarkus REST client extension, we need to inject client bean with @RestClient annotation.

@Path("/insurances")
public class InsuranceResource {

   @Inject
   Logger log;
   @Inject
   InsuranceRepository insuranceRepository;
   @Inject @RestClient
   PersonService personService;

   @POST
   @Transactional
   public Insurance addInsurance(Insurance insurance) {
      insuranceRepository.persist(insurance);
      return insurance;
   }

   @GET
   public List<Insurance> getInsurances() {
      return insuranceRepository.listAll();
   }

   @GET
   @Path("/{id}")
   public Insurance getInsuranceById(@PathParam("id") Long id) {
      return insuranceRepository.findById(id);
   }

   @GET
   @Path("/{id}/details")
   public InsuranceDetails getInsuranceDetailsById(@PathParam("id") Long id, @HeaderParam("X-Version") String version) {
      log.infof("getInsuranceDetailsById: id=%d, version=%s", id, version);
      Insurance insurance = insuranceRepository.findById(id);
      InsuranceDetails insuranceDetails = new InsuranceDetails();
      insuranceDetails.setPersonId(insurance.getPersonId());
      insuranceDetails.setAmount(insurance.getAmount());
      insuranceDetails.setType(insurance.getType());
      insuranceDetails.setExpiry(insurance.getExpiry());
      insuranceDetails.setPerson(personService.getPersonById(insurance.getPersonId()));
      return insuranceDetails;
   }

}

As you probably remember from the previous section, we need to propagate several headers responsible for Istio tracing to the downstream Quarkus service. Let’s take a look at the implementation of the REST client in the PersonService interface. In order to send some additional headers to the request, we need to annotate the interface with @RegisterClientHeaders. Then we have two options. We can provide our custom headers factory as shown below. Otherwise, we may use the property org.eclipse.microprofile.rest.client.propagateHeaders with a list of headers.

@Path("/persons")
@RegisterRestClient
@RegisterClientHeaders(RequestHeaderFactory.class)
public interface PersonService {

   @GET
   @Path("/{id}")
   Person getPersonById(@PathParam("id") Long id);
}

Let’s just take a look at the implementation of the REST endpoint in the person-app. The method getPersonId at the bottom is responsible for finding a person by the id field. It is a target method called by the PersonService client.

@Path("/persons")
public class PersonResource {

   @Inject
   Logger log;
   @Inject
   PersonRepository personRepository;

   @POST
   @Transactional
   public Person addPerson(Person person) {
      personRepository.persist(person);
      return person;
   }

   @GET
   public List<Person> getPersons() {
      return personRepository.listAll();
   }

   @GET
   @Path("/{id}")
   public Person getPersonById(@PathParam("id") Long id) {
      log.infof("getPersonById: id=%d", id);
      Person p = personRepository.findById(id);
      log.infof("getPersonById: %s", p);
      return p;
   }
}

Finally, the implementation of our customer client headers factory. It needs to implement the ClientHeadersFactory interface and its update() method. We are not doing anything complicated here. We are forwarding B3 tracing headers and the X-Version used by Istio to route between two versions of the person-app. I also added some logs. There I didn’t use the already mentioned option of header propagation based on the property org.eclipse.microprofile.rest.client.propagateHeaders.

@ApplicationScoped
public class RequestHeaderFactory implements ClientHeadersFactory {

   @Inject
   Logger log;

   @Override
   public MultivaluedMap<String, String> update(MultivaluedMap<String, String> inHeaders,
                                                 MultivaluedMap<String, String> outHeaders) {
      String version = inHeaders.getFirst("x-version");
      log.infof("Version Header: %s", version);
      String traceId = inHeaders.getFirst("x-b3-traceid");
      log.infof("Trace Header: %s", traceId);
      MultivaluedMap<String, String> result = new MultivaluedHashMap<>();
      result.add("X-Version", version);
      result.add("X-B3-TraceId", traceId);
      result.add("X-B3-SpanId", inHeaders.getFirst("x-b3-spanid"));
      result.add("X-B3-ParentSpanId", inHeaders.getFirst("x-b3-parentspanid"));
      return result;
   }
}

Run Quarkus Applications on Kubernetes

Before we test Istio tracing we need to deploy our Quarkus microservices on Kubernetes. We may do it in several different ways. One of the methods is provided directly by Quarkus. Thanks to that we can generate the Deployment manifests automatically during the build. It turns out that we can also apply Istio manifests to the Kubernetes cluster as well. Firstly, we need to include the following two modules.

<dependency>
  <groupId>io.quarkus</groupId>
  <artifactId>quarkus-openshift</artifactId>
</dependency>
<dependency>
  <groupId>me.snowdrop</groupId>
  <artifactId>istio-client</artifactId>
  <version>1.7.7.1</version>
</dependency>

If you deploy on Kubernetes just replace the quarkus-openshift module with the quarkus-kubernetes module. In the next step, we need to provide some configuration settings in the application.properties file. In order to enable deployment during the build, we need to set the property quarkus.kubernetes.deploy to true. We can configure several aspects of the Kubernetes Deployment. For example, we may enable the Istio proxy by setting the annotation sidecar.istio.io/inject to true or add some labels required for routing: app and version (in that case). Finally, our application connects to the database, so we need to inject values from the Kubernetes Secret.

quarkus.container-image.group = demo-mesh
quarkus.container-image.build = true
quarkus.kubernetes.deploy = true
quarkus.kubernetes.deployment-target = openshift
quarkus.kubernetes-client.trust-certs = true

quarkus.openshift.deployment-kind = Deployment
quarkus.openshift.labels.app = quarkus-insurance-app
quarkus.openshift.labels.version = v1
quarkus.openshift.annotations."sidecar.istio.io/inject" = true
quarkus.openshift.env.mapping.postgres_user.from-secret = insurance-db
quarkus.openshift.env.mapping.postgres_user.with-key = database-user
quarkus.openshift.env.mapping.postgres_password.from-secret = insurance-db
quarkus.openshift.env.mapping.postgres_password.with-key = database-password
quarkus.openshift.env.mapping.postgres_db.from-secret = insurance-db
quarkus.openshift.env.mapping.postgres_db.with-key = database-name

Here’s a part of the configuration responsible for database connection settings. The name of the key after quarkus.openshift.env.mappings maps to the name of the environment variable, for example, the postgres_password property to the POSTGRES_PASSWORD env.

quarkus.datasource.db-kind = postgresql
quarkus.datasource.username = ${POSTGRES_USER}
quarkus.datasource.password = ${POSTGRES_PASSWORD}
quarkus.datasource.jdbc.url = 
jdbc:postgresql://person-db:5432/${POSTGRES_DB}

If there are any additional manifests to apply we should place them inside the src/main/kubernetes directory. This applies to, for example, the Istio configuration. So, now the only thing we need to do is to application build. Firstly go to the quarkus-person-app directory and run the following command. Then, go to the quarkus-insurance-app directory, and do the same.

$ mvn clean package

Traffic Management with Istio

There are two versions of the person-app application. So, let’s create the DestinationRule object containing two subsets v1 and v2 based on the version label.

apiVersion: networking.istio.io/v1beta1
kind: DestinationRule
metadata:
  name: quarkus-person-app-dr
spec:
  host: quarkus-person-app
  subsets:
    - name: v1
      labels:
        version: v1
    - name: v2
      labels:
        version: v2

In the next step, we need to create the VirtualService object for the quarkus-person-app service. The routing between versions can be based on the X-Version header. If that is not set Istio sends 50% to the v1 version, and 50% to the v2 version. Also, we will inject the delay into the v2 route using the Istio HTTPFaultInjection object. It adds 3 seconds delay for 50% of incoming requests.

apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: quarkus-person-app-vs
spec:
  hosts:
    - quarkus-person-app
  http:
    - match:
        - headers:
            X-Version:
              exact: v1
      route:
        - destination:
            host: quarkus-person-app
            subset: v1
    - match:
        - headers:
            X-Version:
              exact: v2
      route:
        - destination:
            host: quarkus-person-app
            subset: v2
      fault:
        delay:
          fixedDelay: 3s
          percentage:
            value: 50
    - route:
        - destination:
            host: quarkus-person-app
            subset: v1
          weight: 50
        - destination:
            host: quarkus-person-app
            subset: v2
          weight: 50

Now, let’s create the Istio Gateway object. Replace the CLUSTER_DOMAIN variable with your cluster’s domain name:

apiVersion: networking.istio.io/v1beta1
kind: Gateway
metadata:
  name: microservices-gateway
spec:
  selector:
    istio: ingressgateway
  servers:
    - port:
        number: 80
        name: http
        protocol: HTTP
      hosts:
        - quarkus-insurance-app.apps.$CLUSTER_DOMAIN
        - quarkus-person-app.apps.$CLUSTER_DOMAIN

In order to forward traffic from the gateway, the VirtualService needs to refer to that gateway.

apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: quarkus-insurance-app-vs
spec:
  hosts:
    - quarkus-insurance-app.apps.$CLUSTER_DOMAIN
  gateways:
    - microservices-gateway
  http:
    - match:
        - uri:
            prefix: "/insurance"
      rewrite:
        uri: " "
      route:
        - destination:
            host: quarkus-insurance-app
          weight: 100

Now you can call the insurance-app service:

$ curl http://quarkus-insurance-app.apps.${CLUSTER_DOMAIN}/insurance/insurances/1/details

We can verify all the existing Istio objects using Kiali.

Testing Istio Tracing with Quarkus

First of all, you can generate many requests using the siege tool. There are multiple ways to run it. We can prepare the file with example requests as shown below:

http://quarkus-person-app.apps.${CLUSTER_DOMAIN}/person/persons/1
http://quarkus-person-app.apps.${CLUSTER_DOMAIN}/person/persons/2
http://quarkus-person-app.apps.${CLUSTER_DOMAIN}/person/persons/3
http://quarkus-person-app.apps.${CLUSTER_DOMAIN}/person/persons/4
http://quarkus-person-app.apps.${CLUSTER_DOMAIN}/person/persons/5
http://quarkus-person-app.apps.${CLUSTER_DOMAIN}/person/persons/6
http://quarkus-person-app.apps.${CLUSTER_DOMAIN}/person/persons/7
http://quarkus-person-app.apps.${CLUSTER_DOMAIN}/person/persons/8
http://quarkus-person-app.apps.${CLUSTER_DOMAIN}/person/persons/9
http://quarkus-person-app.apps.${CLUSTER_DOMAIN}/person/persons
http://quarkus-insurance-app.apps.${CLUSTER_DOMAIN}/insurance/insurances/1/details
http://quarkus-insurance-app.apps.${CLUSTER_DOMAIN}/insurance/insurances/2/details
http://quarkus-insurance-app.apps.${CLUSTER_DOMAIN}/insurance/insurances/3/details
http://quarkus-insurance-app.apps.${CLUSTER_DOMAIN}/insurance/insurances/4/details
http://quarkus-insurance-app.apps.${CLUSTER_DOMAIN}/insurance/insurances/5/details
http://quarkus-insurance-app.apps.${CLUSTER_DOMAIN}/insurance/insurances/6/details
http://quarkus-insurance-app.apps.${CLUSTER_DOMAIN}/insurance/insurances/1
http://quarkus-insurance-app.apps.${CLUSTER_DOMAIN}/insurance/insurances/2
http://quarkus-insurance-app.apps.${CLUSTER_DOMAIN}/insurance/insurances/3
http://quarkus-insurance-app.apps.${CLUSTER_DOMAIN}/insurance/insurances/4
http://quarkus-insurance-app.apps.${CLUSTER_DOMAIN}/insurance/insurances/5
http://quarkus-insurance-app.apps.${CLUSTER_DOMAIN}/insurance/insurances/6

Now we need to set that file as an input for the siege command. We can also set the number of repeats (-r) and concurrent threads (-c).

$ siege -f k8s/traffic/urls.txt -i -v -r 500 -c 10 --no-parser

It takes some time before the command will finish. In the meanwhile, let’s try to send a single request to the insurance-app with the X-Version header set to v2. 50% of such requests are delayed by the Istio quarkus-person-app-vs VirtualService. Repeat the request until you have the response with 3s delay:

Here’s the access log from the Quarkus application:

Then, let’s switch to the Jaeger console. We can find the request by the guid:x-request-id tag.

Here’s the result of our search:

We have a full trace of the request including communication between Istio gateway and insurance-app, and also between the insurance-app and person-app. In order to print the details of the trace just click the record. You will see a trace timeline and requests/responses structure. You can easily verify that latency occurs somewhere between insurance-app and person-app, because the request has been processed only 4.46 ms by the person-app.

The post Distributed Tracing with Istio, Quarkus and Jaeger appeared first on Piotr's TechBlog.

Source Code

If you would like to try it by yourself, you may always take a look at my source code. In order to do that you need to clone my GitHub repository. After that, you should just follow my instructions.

Instead of Spring Kafka, you could as well use Spring Cloud Stream for Kafka. You can read more about it in this article. Spring Cloud Stream provides several useful features like DLQ support, serialization to JSON by default, or interactive queries.

Architecture

We will create a simple system that consists of three microservices. The order-service sends orders to the Kafka topic called orders. Both other microservices stock-service and payment-service listen for the incoming events. After receiving them they verify if it is possible to execute the order. For example, if there are no sufficient funds on the customer account the order is rejected. Otherwise, the payment-service accepts the order and sends a response to the payment-orders topic. The same with stock-service except that it verifies a number of products in stock and sends a response to the stock-orders topic.

Then, the order-service joins two streams from the stock-orders and payment-orders topics by the order’s id. If both orders were accepted it confirms a distributed transaction. On the other hand, if one order has been accepted and the second rejected it performs rollback. In that case, it just generates o new order event and sends it to the orders topic. We may treat the orders topic as a stream of the order’s status changes or just like a table with the last status. Here’s the picture that visualizes our scenario.

Kafka Streams with Spring Boot

Let’s begin our implementation from the order-service. Surprisingly there is no Spring Boot starter for Kafka (unless we use Spring Cloud Stream). Therefore we need to include the spring-kafka dependency. In order to process streams we also need to include the kafka-streams module directly. Since the order-service exposes some REST endpoints it is required to add the Spring Boot Web starter.

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
  <groupId>org.springframework.kafka</groupId>
  <artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-streams</artifactId>
</dependency>

The order-service is the most important microservice in our scenario. It acts as an order gateway and a saga pattern orchestrator. It requires all the three topics used in our architecture. In order to automatically create topics on application startup we need to define the following beans:

@Bean
public NewTopic orders() {
   return TopicBuilder.name("orders")
         .partitions(3)
         .compact()
         .build();
}

@Bean
public NewTopic paymentTopic() {
   return TopicBuilder.name("payment-orders")
         .partitions(3)
         .compact()
         .build();
}

@Bean
public NewTopic stockTopic() {
   return TopicBuilder.name("stock-orders")
         .partitions(3)
         .compact()
         .build();
}

Then, let’s define our first Kafka stream. To do that we need to use the StreamsBuilder bean. The order-service receives events from the payment-service (in the payment-events topic) and from the stock-service (in the stock-events topic). Every single event contains the id previously set by the order-service. If we join both these streams into a single stream by order’s id we will be able to determine the status of our transaction. The algorithm is pretty simple. If both payment-service and stock-service accepted the order the final status of transaction is CONFIRMED. If both services rejected the order the final status is REJECTED. The last option is ROLLBACK – when one service accepted the order, and one service rejected it. Here’s the described method inside the OrderManageService bean.

@Service
public class OrderManageService {

   public Order confirm(Order orderPayment, Order orderStock) {
      Order o = new Order(orderPayment.getId(),
             orderPayment.getCustomerId(),
             orderPayment.getProductId(),
             orderPayment.getProductCount(),
             orderPayment.getPrice());
      if (orderPayment.getStatus().equals("ACCEPT") &&
                orderStock.getStatus().equals("ACCEPT")) {
         o.setStatus("CONFIRMED");
      } else if (orderPayment.getStatus().equals("REJECT") &&
                orderStock.getStatus().equals("REJECT")) {
         o.setStatus("REJECTED");
      } else if (orderPayment.getStatus().equals("REJECT") ||
                orderStock.getStatus().equals("REJECT")) {
         String source = orderPayment.getStatus().equals("REJECT")
                    ? "PAYMENT" : "STOCK";
         o.setStatus("ROLLBACK");
         o.setSource(source);
      }
      return o;
   }

}

Finally, the implementation of our stream. We need to define the KStream bean. We are joining two streams using the join method of KStream. The joining window is 10 seconds. As the result, we are setting the status of the order and sending a new order to the orders topic. We use the same topic as for sending new orders.

@Autowired
OrderManageService orderManageService;

@Bean
public KStream<Long, Order> stream(StreamsBuilder builder) {
   JsonSerde<Order> orderSerde = new JsonSerde<>(Order.class);
   KStream<Long, Order> stream = builder
         .stream("payment-orders", Consumed.with(Serdes.Long(), orderSerde));

   stream.join(
         builder.stream("stock-orders"),
         orderManageService::confirm,
         JoinWindows.of(Duration.ofSeconds(10)),
         StreamJoined.with(Serdes.Long(), orderSerde, orderSerde))
      .peek((k, o) -> LOG.info("Output: {}", o))
      .to("orders");

   return stream;
}

Let’s see it also in the picture.

Configuration for Spring Boot

In Spring Boot the name of the application is by default the name of the consumer group for Kafka Streams. Therefore, we should set in application.yml. Of course, we also need to set the address of the Kafka bootstrap server. Finally, we have to configure the default key and value for events serialization. It applies to both standard and streams processing.

spring.application.name: orders
spring.kafka:
  bootstrap-servers: 127.0.0.1:56820
  producer:
    key-serializer: org.apache.kafka.common.serialization.LongSerializer
    value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
  streams:
    properties:
      default.key.serde: org.apache.kafka.common.serialization.Serdes$LongSerde
      default.value.serde: org.springframework.kafka.support.serializer.JsonSerde
      spring.json.trusted.packages: "*"

Sending and receiving events from Kafka topics

In the previous section, we discussed how to create a new Kafka stream as a result of joining two other streams. Now, let’s see how to process incoming messages. We can consider it on the example of the payment-service. It listens for the incoming orders. If it receives a new order it performs reservation on the customer’s account and sends a response with a reservation status to the payment-orders topic. If it receives confirmation of the transaction from the order-service, it commits the transaction or rollbacks it. In order to enable Kafka listener, we should annotate the main class with @EnableKafka. Additionally, the listening method must be annotated with the @KafkaListener. The following method listens for events on the orders topic and runs in the payment consumer group.

@SpringBootApplication
@EnableKafka
public class PaymentApp {

    private static final Logger LOG = LoggerFactory.getLogger(PaymentApp.class);

    public static void main(String[] args) {
        SpringApplication.run(PaymentApp.class, args);
    }

    @Autowired
    OrderManageService orderManageService;

    @KafkaListener(id = "orders", topics = "orders", groupId = "payment")
    public void onEvent(Order o) {
        LOG.info("Received: {}" , o);
        if (o.getStatus().equals("NEW"))
            orderManageService.reserve(o);
        else
            orderManageService.confirm(o);
    }
}

Here’s the implementation of the OrderManageService used in the previous code snippet. If it receives the order in the NEW status it performs reservation. During the reservation, it subtracts the order’s price from the amountAvailable field and adds the same value to the amountReserved field. Then it sets the status of the order and sends a response to the payment-orders topic using KafkaTemplate. During the confirmation phase, it doesn’t send any response event. It can perform a rollback, which means – subtracting the order’s price from the amountReserved field and adding it to the amountAvailable field. Otherwise it just “commits” the transaction by subtracting the price from the amountReserved field.

@Service
public class OrderManageService {

   private static final String SOURCE = "payment";
   private static final Logger LOG = LoggerFactory.getLogger(OrderManageService.class);
   private CustomerRepository repository;
   private KafkaTemplate<Long, Order> template;

   public OrderManageService(CustomerRepository repository, KafkaTemplate<Long, Order> template) {
      this.repository = repository;
      this.template = template;
   }

   public void reserve(Order order) {
      Customer customer = repository.findById(order.getCustomerId()).orElseThrow();
      LOG.info("Found: {}", customer);
      if (order.getPrice() < customer.getAmountAvailable()) {
         order.setStatus("ACCEPT");
         customer.setAmountReserved(customer.getAmountReserved() + order.getPrice());
         customer.setAmountAvailable(customer.getAmountAvailable() - order.getPrice());
      } else {
         order.setStatus("REJECT");
      }
      order.setSource(SOURCE);
      repository.save(customer);
      template.send("payment-orders", order.getId(), order);
      LOG.info("Sent: {}", order);
   }

   public void confirm(Order order) {
      Customer customer = repository.findById(order.getCustomerId()).orElseThrow();
      LOG.info("Found: {}", customer);
      if (order.getStatus().equals("CONFIRMED")) {
         customer.setAmountReserved(customer.getAmountReserved() - order.getPrice());
         repository.save(customer);
      } else if (order.getStatus().equals("ROLLBACK") && !order.getSource().equals(SOURCE)) {
         customer.setAmountReserved(customer.getAmountReserved() - order.getPrice());
         customer.setAmountAvailable(customer.getAmountAvailable() + order.getPrice());
         repository.save(customer);
      }

   }
}

A similar logic is implemented on the stock-service side. However, instead of the order’s price, it uses the field productCount and performs reservation for the desired number of ordered products. Here’s the implementation of the OrderManageService class in the stock-service.

@Service
public class OrderManageService {

   private static final String SOURCE = "stock";
   private static final Logger LOG = LoggerFactory.getLogger(OrderManageService.class);
   private ProductRepository repository;
   private KafkaTemplate<Long, Order> template;

   public OrderManageService(ProductRepository repository, KafkaTemplate<Long, Order> template) {
      this.repository = repository;
      this.template = template;
   }

   public void reserve(Order order) {
      Product product = repository.findById(order.getProductId()).orElseThrow();
      LOG.info("Found: {}", product);
      if (order.getStatus().equals("NEW")) {
         if (order.getProductCount() < product.getAvailableItems()) {
            product.setReservedItems(product.getReservedItems() + order.getProductCount());
            product.setAvailableItems(product.getAvailableItems() - order.getProductCount());
            order.setStatus("ACCEPT");
            repository.save(product);
         } else {
            order.setStatus("REJECT");
         }
         template.send("stock-orders", order.getId(), order);
         LOG.info("Sent: {}", order);
      }
   }

   public void confirm(Order order) {
      Product product = repository.findById(order.getProductId()).orElseThrow();
      LOG.info("Found: {}", product);
      if (order.getStatus().equals("CONFIRMED")) {
         product.setReservedItems(product.getReservedItems() - order.getProductCount());
         repository.save(product);
      } else if (order.getStatus().equals("ROLLBACK") && !order.getSource().equals(SOURCE)) {
         product.setReservedItems(product.getReservedItems() - order.getProductCount());
         product.setAvailableItems(product.getAvailableItems() + order.getProductCount());
         repository.save(product);
      }
   }

}

Query Kafka Stream with Spring Boot

Now, let’s consider the following scenario. Firstly, the order-service receives a new order (via REST API) and sends it to the Kafka topic. This order is then received by both other microservices. Once they send back a positive response (or negative) the order-service process them as streams and change the status of the order. The order with a new status is emitted to the same topic as before. So, where we are storing the data with the current status of an order? In Kafka topic. Once again we will use Kafka Streams in our Spring Boot application. But this time, we take advantage of KTable. Let’s at the visualization of our scenario.

Ok, so let’s define another Kafka Streams bean in the order-service. We are getting the same orders topic as a stream. We will convert it to the Kafka table and materialize it in a persistent store. Thanks to that we will be able to easily query the store from our REST controller.

@Bean
public KTable<Long, Order> table(StreamsBuilder builder) {
   KeyValueBytesStoreSupplier store =
         Stores.persistentKeyValueStore("orders");
   JsonSerde<Order> orderSerde = new JsonSerde<>(Order.class);
   KStream<Long, Order> stream = builder
         .stream("orders", Consumed.with(Serdes.Long(), orderSerde));
   return stream.toTable(Materialized.<Long, Order>as(store)
         .withKeySerde(Serdes.Long())
         .withValueSerde(orderSerde));
}

If we run more than one instance of the order-service on the same machine it is also important to override the default location of the state store. To do that we should define the following property unique per every instance:

spring.kafka.streams.state-dir: /tmp/kafka-streams/1

Unfortunately, there is no built-in support in Spring Boot for interactive queries of Kafka Streams. However, we can use auto-configured StreamsBuilderFactoryBean to inject KafkaStreams instance into the controller. Then we can query the state store under the “materialized” name. That’s of course very trivial sample. We are just getting all orders from KTable.

@GetMapping
public List<Order> all() {
   List<Order> orders = new ArrayList<>();
      ReadOnlyKeyValueStore<Long, Order> store = kafkaStreamsFactory
         .getKafkaStreams()
         .store(StoreQueryParameters.fromNameAndType(
                "orders",
                QueryableStoreTypes.keyValueStore()));
   KeyValueIterator<Long, Order> it = store.all();
   it.forEachRemaining(kv -> orders.add(kv.value));
   return orders;
}

In the same OrderController there is also a method for sending a new order to the Kafka topic.

@PostMapping
public Order create(@RequestBody Order order) {
   order.setId(id.incrementAndGet());
   template.send("orders", order.getId(), order);
   LOG.info("Sent: {}", order);
   return order;
}

Testing Scenario

Before we run our sample microservices, we need to start the local instance of Kafka. Usually, I’m using Redpanda for that. It is a Kafka API compatible streaming platform. In comparison to Kafka, it is relatively easy to run it locally. All you need to do is to install the rpk CLI locally (here is the instruction for macOS). After that, you can create a single-node instance using the following command:

$ rpk container start

After running, it will print the address of your node. For me, it is 127.0.0.1:56820. You should put that address as a value of the spring.kafka.bootstrap-servers property. You can also display a list of created topics using the following command:

$ rpk topic list --brokers 127.0.0.1:56820

Then, let’s run our microservices. Begin from the order-service since it is creating all the required topics and building the Kafka Streams instance. You can send a single order using the REST endpoint:

$ curl -X 'POST' \
  'http://localhost:8080/orders' \
  -H 'Content-Type: application/json' \
  -d '{
  "customerId": 10,
  "productId": 10,
  "productCount": 5,
  "price": 100,
  "status": "NEW"
}'

There is some test data inserted while payment-service and stock-service start. So, you can set the value of customerId or productId between 1 and 100 and it will work for you. However, you can use as well a method for generating a random stream of data. The following bean is responsible for generating 10000 random orders:

@Service
public class OrderGeneratorService {

   private static Random RAND = new Random();
   private AtomicLong id = new AtomicLong();
   private Executor executor;
   private KafkaTemplate<Long, Order> template;

   public OrderGeneratorService(Executor executor, KafkaTemplate<Long, Order> template) {
      this.executor = executor;
      this.template = template;
   }

   @Async
   public void generate() {
      for (int i = 0; i < 10000; i++) {
         int x = RAND.nextInt(5) + 1;
         Order o = new Order(id.incrementAndGet(), RAND.nextLong(100) + 1, RAND.nextLong(100) + 1, "NEW");
         o.setPrice(100 * x);
         o.setProductCount(x);
         template.send("orders", o.getId(), o);
      }
   }
}

You can start that process by calling the endpoint POST /orders/generate.

@PostMapping("/generate")
public boolean create() {
   orderGeneratorService.generate();
   return true;
}

No matter if decide to send single order or generate multiple random orders you can easily query the status of orders using the following endpoint:

$ curl http://localhost:8080/orders

Here’s a structure of topics generated by the application and by Kafka Streams to perform join operation and save the orders KTable as a state store.

The post Distributed Transactions in Microservices with Kafka Streams and Spring Boot appeared first on Piotr's TechBlog.

Introduction

What’s Telepresence? It’s a very simple and powerful CLI tool for fast and local development for Kubernetes. Why it is simple? Because you can do almost everything using a single command. Telepresence is a CNCF sandbox project originally created by Ambassador Labs. It lets you run and test microservices locally against a remote Kubernetes cluster. It intercepts remote traffic and sends it to your local running instance. I won’t focus on the technical aspects. If you want to read more about it, you can refer to the following link.

Firstly, let’s analyze our case. There are three microservices: first-service, caller-service and callme-service. All of them expose a single REST endpoint GET /ping, which returns basic information about each microservice. In order to create applications, I’m using the Spring Boot framework. Our architecture is visible in the picture below. The first-service is calling the endpoint exposed by the caller-service. Then the caller-service is calling the endpoint exposed by the callme-service. Of course, we are going to deploy all the microservices on the remote Kubernetes cluster.

Assuming we have something to do with the caller-service, we are also running it locally. So finally, our goal is to forward the traffic that is sent to the caller-service on the Kubernetes cluster to our local instance. On the other hand, the local instance of the caller-service should call the instance of the callme-service running on the remote cluster. Looks hard? Let’s check it out!

Source Code

If you would like to try it by yourself, you may always take a look at my source code. In order to do that you need to clone my GitHub repository. The code used in this article is available in the dev branch. Then you should just follow my instructions 

Prerequisites

Before we start, we need to install several tools. Of course, we also need to have a running Kubernetes cluster (or e.g. OpenShift). We will use the following CLI tools:

1. kubectl – to interact with the Kubernetes cluster. It is also used by Skaffold
2. skaffold – here are installation instructions. It works perfectly fine on Linux, Macos as well as Windows too
3. telepresence – here are installation instruction. I’m not sure about Windows since it is in developer preview there. However, I’m using it on Macos without any problems
4. Maven + JDK11 – of course we need to build applications locally before deploying to Kubernetes.

Build and deploy applications on Kubernetes with Skaffold and Jib

Our applications are as simple as possible. Let’s take a look at the callme-service REST endpoint implementation. It just returns the name of the microservice and its version (v1 everywhere in this article):

@RestController
@RequestMapping("/callme")
public class CallmeController {

   private static final Logger LOGGER = 
       LoggerFactory.getLogger(CallmeController.class);

   @Autowired
   BuildProperties buildProperties;
   @Value("${VERSION}")
   private String version;

   @GetMapping("/ping")
   public String ping() {
      LOGGER.info("Ping: name={}, version={}", 
         buildProperties.getName(), version);
      return "I'm callme-service " + version;
   }
}

The endpoint visible above is called by the caller-service. The same as before it also prints the name of the microservice and its version. But also, it appends the result received from the callme-service. It calls the callme-service endpoint using the Spring RestTemplate and the name of the Kubernetes Service.

@RestController
@RequestMapping("/caller")
public class CallerController {

   private static final Logger LOGGER = 
      LoggerFactory.getLogger(CallerController.class);

   @Autowired
   BuildProperties buildProperties;
   @Autowired
   RestTemplate restTemplate;
   @Value("${VERSION}")
   private String version;

   @GetMapping("/ping")
   public String ping() {
      LOGGER.info("Ping: name={}, version={}", 
         buildProperties.getName(), version);
      String response = restTemplate
         .getForObject("http://callme-service:8080/callme/ping", String.class);
      LOGGER.info("Calling: response={}", response);
      return "I'm caller-service " + version + ". Calling... " + response;
   }
}

Finally, let’s take a look at the implementation of the first-service @RestController. It calls the caller-service endpoint visible above.

@RestController
@RequestMapping("/first")
public class FirstController {

   private static final Logger LOGGER = 
      LoggerFactory.getLogger(FirstController.class);

   @Autowired
   BuildProperties buildProperties;
   @Autowired
   RestTemplate restTemplate;
   @Value("${VERSION}")
   private String version;

   @GetMapping("/ping")
   public String ping() {
      LOGGER.info("Ping: name={}, version={}", 
         buildProperties.getName(), version);
      String response = restTemplate
         .getForObject("http://caller-service:8080/caller/ping", String.class);
      LOGGER.info("Calling: response={}", response);
      return "I'm first-service " + version + ". Calling... " + response;
   }

}

Here’s a definition of the Kubernetes Service for e.g. callme-service:

apiVersion: v1
kind: Service
metadata:
  name: callme-service
  labels:
    app: callme-service
spec:
  type: ClusterIP
  ports:
  - port: 8080
    name: http
  selector:
    app: callme-service

You can see Kubernetes YAML manifests inside the k8s directory for every single microservice. Let’s take a look at the example Deployment manifest for the callme-service.

apiVersion: apps/v1
kind: Deployment
metadata:
  name: callme-service
spec:
  replicas: 1
  selector:
    matchLabels:
      app: callme-service
      version: v1
  template:
    metadata:
      labels:
        app: callme-service
        version: v1
    spec:
      containers:
        - name: callme-service
          image: piomin/callme-service
          imagePullPolicy: IfNotPresent
          ports:
            - containerPort: 8080
          env:
            - name: VERSION
              value: "v1"

We can deploy each microservice independently, or all of them at once. Here’s a global Skaffold configuration for the whole project. You can find it in the root directory. As you see it uses Jib as a build tool and tries to find manifests inside k8s directory of every single module.

apiVersion: skaffold/v2beta22
kind: Config
metadata:
  name: simple-istio-services
build:
  artifacts:
    - image: piomin/first-service
      jib:
        project: first-service
    - image: piomin/caller-service
      jib:
        project: caller-service
    - image: piomin/callme-service
      jib:
        project: callme-service
  tagPolicy:
    gitCommit: {}
deploy:
  kubectl:
    manifests:
      - '*/k8s/*.yaml'

Each Maven module has to include Jib Maven plugin.

<plugin>
  <groupId>com.google.cloud.tools</groupId>
  <artifactId>jib-maven-plugin</artifactId>
  <version>3.1.1</version>
</plugin>

Finally, we can deploy all our microservices on Kubernetes with Skaffold. Jib is working in Dockerless mode, so you don’t have to run Docker on your machine. By default, it uses adoptopenjdk:11-jre as a base following Java version defined in Maven pom.xml. If you want to observe logs after running applications on Kubernetes just activate the --tail option.

$ skaffold run --tail

Let’s just display a list of running pods to verify if the deployment was successful:

$ kubectl get pod
NAME                              READY   STATUS    RESTARTS   AGE
caller-service-688bd76c98-2m4gp   1/1     Running   0          3m1s
callme-service-75c7cf5bf-rfx69    1/1     Running   0          3m
first-service-7698465bcb-rvf77    1/1     Running   0          3m

Using Telepresence with Kubernetes

Let the party begin! After running all microservices on Kubernetes we will connect Telepresence to our cluster. The following command will run Telepresence deamon on your machine and connect it to the Kubernetes cluster (from current Kubecontext).

$ telepresence connect

If you see a similar result it means everything goes well.

Telepresence has already connected to your Kubernetes cluster, but it still not intercepting any traffic from the pods. You can verify it with the following command:

$ telepresence list
caller-service: ready to intercept (traffic-agent not yet installed)
callme-service: ready to intercept (traffic-agent not yet installed)
first-service : ready to intercept (traffic-agent not yet installed)

Ok, so now let’s intercept the traffic from the caller-service.

$ telepresence intercept caller-service --port 8080:8080

Here’s my result after running the command visible above.

Now, the only thing we need to do is to run the caller-service on the local machine. By default, it listens on the port 8080:

$ mvn clean spring-boot:run

We can do it event smarter with the single Telepresence command instead of running them separately:

$ telepresence intercept caller-service --port 8080:8080 -- mvn clean spring-boot:run

Before we send a test request let’s analyze what happened. After running the telepresence intercept command Telepresence injects a sidecar container into the application pod. The name of this container is traffic-agent. It is responsible for intercepting the traffic that comes to the caller-service.

$ kubectl get pod caller-service-7577b9f6fd-ww7nv \
  -o jsonpath='{.spec.containers[*].name}'
caller-service traffic-agent

Ok, now let’s just call the first-service running on the remote Kubernetes cluster. I deployed it on OpenShift, so I can easily call it externally using the Route object. If you run it on other plain Kubernetes you can create Ingress or just run the kubectl port-forward command. Alternatively, you may also enable port forwarding on the skaffold run command (--port-forward option). Anyway, let’s call the first-service /ping endpoint. Here’s my result.

Here are the application logs from the Kubernetes cluster printed by the skaffold run command. As you see it just prints the logs from callme-service and first-service:

Now, let’s take a look at the logs from the local instance of the caller-service. Telepresence intercepts the traffic and sends it to the local instance of the application. Then this instance call the callme-service on the remote cluster

Cleanup Kubernetes environment after using Telepresence

To clean up the environment just run the following command. It will remove the sidecar container from your application pod. However, if you start your Spring Boot application within the telepresence intercept command you just need to kill the local process with CTRL+C (however the traffic-agent container is still inside the pod).

$ telepresence uninstall --agent caller-service

After that, you can call the first-service once again. Now, all the requests are not going out of the cluster.

In order to shutdown the Telepresence daemon and disconnect from the Kubernetes cluster just run the following command:

$ telepresence quit

Final Thoughts

You can also easily debug your microservices locally, just by running the same telepresence intercept command and your application in debug mode. What’s important in this scenario – Telepresence does not force you to use any other particular tools or IDE. You can do everything the same as would you normally run or debug the application locally. I hope you will like that tool the same as me

The post Development on Kubernetes with Telepresence and Skaffold appeared first on Piotr's TechBlog.