• Use Quarkus Dekorate extension to automatically generate Kubernetes manifests basing on the source code and configuration
• Build and push images to Docker registry with Jib extension
• Deploy your application on Kubernetes without any manually created YAML in one click
• Use Quarkus Kubernetes Config to inject configuration properties from ConfigMap

This guide is the second in series about Quarkus framework. If you are interested in the introduction to building Quarkus REST applications with Kotlin you may refer to my article Guide to Quarkus with Kotlin.

Source code

The source code with the sample Quarkus applications is available on GitHub. First, you need to clone the following repository: https://github.com/piomin/sample-quarkus-applications.git. Then, you need to go to the employee-service directory. We use the same repository as in my previous article about Quarkus.

1. Dependencies

Quarkus does not implement mechanisms for generating Kubernetes manifests, deploying them on the platform, or building images. It adds some logic to the existing tools. To enable extensions to Dekorate and Jib we should include the following dependencies.

<dependency>
   <groupId>io.quarkus</groupId>
   <artifactId>quarkus-kubernetes</artifactId>
</dependency>
<dependency>
   <groupId>io.quarkus</groupId>
   <artifactId>quarkus-container-image-jib</artifactId>
</dependency>

Jib builds optimized images for Java applications without a Docker daemon, and without deep mastery of Docker best-practices. It is available as plugins for Maven and Gradle and as a Java library. Dekorate is a Java library that makes generating and decorating Kubernetes manifests as simple as adding a dependency to your project. It may generate manifests basing on the source code, annotations, and configuration properties.

2. Preparation

In the first part of my guide to Kotlin, we were running our application in development mode with an embedded H2 database. In this part of the tutorial, we will integrate our application with Postgres deployed on Kubernetes. To do that we first need to change configuration settings for the data source. H2 database will be active only in dev and test mode. The configuration of Postgresql data source would be based on environment variables.

# kubernetes
quarkus.datasource.db-kind=postgresql
quarkus.datasource.username=${POSTGRES_USER}
quarkus.datasource.password=${POSTGRES_PASSWORD}
quarkus.datasource.jdbc.url=jdbc:postgresql://${POSTGRES_HOST}:5432/${POSTGRES_DB}
# dev
%dev.quarkus.datasource.db-kind=h2
%dev.quarkus.datasource.username=sa
%dev.quarkus.datasource.password=password
%dev.quarkus.datasource.jdbc.url=jdbc:h2:mem:testdb
# test
%test.quarkus.datasource.db-kind=h2
%test.quarkus.datasource.username=sa
%test.quarkus.datasource.password=password
%test.quarkus.datasource.jdbc.url=jdbc:h2:mem:testdb

3. Configure Kubernetes extension

With Quarkus Kubernetes extension we may customize the behavior of the manifest generator. To do that we need to provide configuration settings with the prefix quarkus.kubernetes.*. There are pretty many options like defining labels, annotations, environment variables, Secret and ConfigMap references, or mounting volumes. First, let’s take a look at the Secret and ConfigMap prepared for Postgres.

apiVersion: v1
kind: ConfigMap
metadata:
  name: postgres-config
  labels:
    app: postgres
data:
  POSTGRES_DB: quarkus
  POSTGRES_USER: quarkus
  POSTGRES_HOST: postgres
---
apiVersion: v1
kind: Secret
metadata:
  name: postgres-secret
  labels:
    app: postgres
data:
  POSTGRES_PASSWORD: YWRtaW4xMjM=

In this fragment of configuration, besides simple label and annotation, we are adding reference to all the keys inside postgres-config and postgres-secret.

quarkus.kubernetes.labels.app-type=demo
quarkus.kubernetes.annotations.app-type=demo
quarkus.kubernetes.env.secrets=postgres-secret
quarkus.kubernetes.env.configmaps=postgres-config

4. Build image and deploy

Before executing build and deploy we need to apply manifest with Postgres. Here’s Deployment definition of Postgres.

apiVersion: apps/v1
kind: Deployment
metadata:
  name: postgres
spec:
  replicas: 1
  selector:
    matchLabels:
      app: postgres
  template:
    metadata:
      labels:
        app: postgres
    spec:
      containers:
        - name: postgres
          image: postgres:latest
          imagePullPolicy: "IfNotPresent"
          ports:
            - containerPort: 5432
          env:
            - name: POSTGRES_DB
              valueFrom:
                configMapKeyRef:
                  key: POSTGRES_DB
                  name: postgres-config
            - name: POSTGRES_USER
              valueFrom:
                configMapKeyRef:
                  key: POSTGRES_USER
                  name: postgres-config
            - name: POSTGRES_PASSWORD
              valueFrom:
                secretKeyRef:
                  key: POSTGRES_PASSWORD
                  name: postgres-secret
          volumeMounts:
            - mountPath: /var/lib/postgresql/data
              name: postgredb
      volumes:
        - name: postgredb
          persistentVolumeClaim:
            claimName: postgres-claim

Let’s apply it on Kubernetes together with required ConfigMap, Secret, PersistenceVolume and PersistenceVolumeClaim. All the objects are available inside example repository in the file employee-service/k8s/postgres-deployment.yaml.

$ kubectl apply -f employee-service\k8s\postgresql-deployment.yaml

After deploying Postgres we may proceed to the main task. In order to build a Docker image with the application, we need to enable option quarkus.container-image.build during Maven build. If you also want to deploy and run a container with the application on your local Kubernetes instance you need to enable option quarkus.kubernetes.deploy.

$ clean package -Dquarkus.container-image.build=true -Dquarkus.kubernetes.deploy=true

If your Kubernetes cluster is located on the hosted cloud you should push the image to a remote Docker registry before deployment. To do that we should also activate option quarkus.container-image.push during Maven build. If you do not push to the default Docker registry you have to set parameter quarkus.container-image.registry=gcr.io inside the application.properties file. The only thing I need to set for building images is the following property, which is the same as my login to docker.io site.

quarkus.container-image.group=piomin

After running the required Maven command our application is deployed on Kubernetes. Let’s take a look at what happened during the Maven build. Here’s the fragment of logs during that build. You see that Quarkus extension generated two files kubernetes.yaml and kubernetes.json inside target/kubernetes directory. Then it proceeded to build a Docker image with our application. Because we didn’t specify any base image it takes a default one for Java 11 – fabric8/java-alpine-openjdk11-jre.

Let’s take a look on the Deployment definition automatically generated by Quarkus.

1. It adds some annotations like port or path to metrics endpoint used by Prometheus to monitor application and enabled scraping. It also adds Git commit id, repository URL, and our custom annotation defined in application.properties.
2. It adds labels with the application name, version (taken from Maven pom.xml), and our custom label app-type.
3. It injects Kubernetes namespace name into the container.
4. It injects the reference to the postgres-secret defined in application.properties.
5. It injects the reference to the postgres-config defined in application.properties.
6. The name of the image is automatically created. It is based on Maven artifactId and version.
7. The definition of liveness and readiness is generated if Maven module quarkus-smallrye-health is present

apiVersion: apps/v1
kind: Deployment
metadata:
  annotations: # (1)
    prometheus.io/path: /metrics
    prometheus.io/port: 8080
    app.quarkus.io/commit-id: f6ae37288ed445177f23291c921c6099cfc58c6e
    app.quarkus.io/vcs-url: https://github.com/piomin/sample-quarkus-applications.git
    app.quarkus.io/build-timestamp: 2020-08-10 - 13:22:32 +0000
    app-type: demo
    prometheus.io/scrape: "true"
  labels: # (2)
    app.kubernetes.io/name: employee-service
    app.kubernetes.io/version: 1.1
    app-type: demo
  name: employee-service
spec:
  replicas: 1
  selector:
    matchLabels:
      app.kubernetes.io/name: employee-service
      app.kubernetes.io/version: 1.1
  template:
    metadata:
      annotations:
        prometheus.io/path: /metrics
        prometheus.io/port: 8080
        app.quarkus.io/commit-id: f6ae37288ed445177f23291c921c6099cfc58c6e
        app.quarkus.io/vcs-url: https://github.com/piomin/sample-quarkus-applications.git
        app.quarkus.io/build-timestamp: 2020-08-10 - 13:22:32 +0000
        app-type: demo
        prometheus.io/scrape: "true"
      labels:
        app.kubernetes.io/name: employee-service
        app.kubernetes.io/version: 1.1
        app-type: demo
    spec:
      containers:
      - env:
        - name: KUBERNETES_NAMESPACE # (3)
          valueFrom:
            fieldRef:
              fieldPath: metadata.namespace
        envFrom:
        - secretRef: # (4)
            name: postgres-secret
        - configMapRef: # (5)
            name: postgres-config
        image: piomin/employee-service:1.1 # (6)
        imagePullPolicy: IfNotPresent
        livenessProbe: # (7)
          failureThreshold: 3
          httpGet:
            path: /health/live
            port: 8080
            scheme: HTTP
          initialDelaySeconds: 0
          periodSeconds: 30
          successThreshold: 1
          timeoutSeconds: 10
        name: employee-service
        ports:
        - containerPort: 8080
          name: http
          protocol: TCP
        readinessProbe:
          failureThreshold: 3
          httpGet:
            path: /health/ready
            port: 8080
            scheme: HTTP
          initialDelaySeconds: 0
          periodSeconds: 30
          successThreshold: 1
          timeoutSeconds: 10
      serviceAccount: employee-service

Once the image has been built it is available in local registry. Quarkus automatically deploy it to the current cluster using already generated Kubernetes manifests.

Here’s the list of pods in default namespace.

5. Using Kubernetes Config extension

With Kubernetes Config extension you can use ConfigMap as a configuration source, without having to mount them into the pod with the application. To use that extension we need to include the following Maven dependency.

<dependency>
    <groupId>io.quarkus</groupId>
    <artifactId>quarkus-kubernetes-config</artifactId>
</dependency>

This extension works directly with Kubernetes API using fabric8 KubernetesClient. That’s why we should set the proper permissions for ServiceAccount. Fortunately, all the required configuration is automatically generated by Quarkus Kubernetes extension. The RoleBinding object is appied automatically if quarkus-kubernetes-config module is present.

apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  annotations:
    prometheus.io/path: /metrics
    prometheus.io/port: 8080
    app.quarkus.io/commit-id: a5da459af01637657ebb0ec3a606eb53d13b8524
    app.quarkus.io/vcs-url: https://github.com/piomin/sample-quarkus-applications.git
    app.quarkus.io/build-timestamp: 2020-08-10 - 14:25:20 +0000
    app-type: demo
    prometheus.io/scrape: "true"
  labels:
    app.kubernetes.io/name: employee-service
    app.kubernetes.io/version: 1.1
    app-type: demo
  name: employee-service:view
roleRef:
  kind: ClusterRole
  apiGroup: rbac.authorization.k8s.io
  name: view
subjects:
- kind: ServiceAccount
  name: employee-service

Here’s our example ConfigMap that contains a single property property1.

apiVersion: v1
kind: ConfigMap
metadata:
  name: employee-config
data:
  application.properties: |-
    property1=one

The same property is defined inside application.properties available on the classpath, but there it has a different value.

property1=test

Before deploying a new version of application we need to add the following properties. First of them enables Kubernetes ConfigMap injection, while the second specifies the name of injected ConfigMap.

quarkus.kubernetes-config.enabled=true
quarkus.kubernetes-config.config-maps=employee-config

Finally we just need to implement a simple endpoint that injects and returns configuration property.

@ConfigProperty(name = "property1")
lateinit var property1: String

@GET
@Path("/property1")
fun property1(): String = property1

The properties obtained from the ConfigMap have a higher priority than any properties of the same name that are found in application.properties available on the classpath. Let’s test it.

The post Guide to Quarkus on Kubernetes appeared first on Piotr's TechBlog.

• Implementation of REST endpoints
• Integration with H2 with Hibernate and Panache project
• Generating and exposing OpenAPI/Swagger documentation
• Exposing health checks
• Exposing basic metrics
• Logging request and response
• Testing REST endpoints with RestAssured library

Source code

The source code with the sample Quarkus Kotlin applications is available on GitHub. First, you need to clone the following repository: https://github.com/piomin/sample-quarkus-applications.git. Then, you need to go to the employee-service directory.

1. Enable Quarkus Kotlin support

To enable Kotlin support in Quarkus we need to include quarkus-kotlin module. We also have to add kotlin-stdlib library.

<dependency>
   <groupId>io.quarkus</groupId>
   <artifactId>quarkus-kotlin</artifactId>
</dependency>
<dependency>
   <groupId>org.jetbrains.kotlin</groupId>
   <artifactId>kotlin-stdlib</artifactId>
</dependency>

In the next step we need to include kotlin-maven-plugin. Besides standard configuration, we have to use all-open Kotlin compiler plugin. The all-open compiler plugin makes classes annotated with a specific annotation and their members open without the explicit open keyword. Since classes annotated with @Path, @ApplicationScoped, or @QuarkusTest should not be final, we need to add all those annotations to the pluginOptions section.

<build>
   <sourceDirectory>src/main/kotlin</sourceDirectory>
   <testSourceDirectory>src/test/kotlin</testSourceDirectory>
   <plugins>
      <plugin>
         <groupId>io.quarkus</groupId>
         <artifactId>quarkus-maven-plugin</artifactId>
         <version>${quarkus-plugin.version}</version>
         <executions>
            <execution>
               <goals>
                  <goal>build</goal>
               </goals>
            </execution>
         </executions>
      </plugin>
      <plugin>
         <groupId>org.jetbrains.kotlin</groupId>
         <artifactId>kotlin-maven-plugin</artifactId>
         <version>${kotlin.version}</version>
         <executions>
            <execution>
               <id>compile</id>
               <goals>
                  <goal>compile</goal>
               </goals>
            </execution>
            <execution>
               <id>test-compile</id>
               <goals>
                  <goal>test-compile</goal>
               </goals>
            </execution>
         </executions>
         <dependencies>
            <dependency>
               <groupId>org.jetbrains.kotlin</groupId>
               <artifactId>kotlin-maven-allopen</artifactId>
               <version>${kotlin.version}</version>
            </dependency>
         </dependencies>
         <configuration>
            <javaParameters>true</javaParameters>
            <jvmTarget>11</jvmTarget>
            <compilerPlugins>
               <plugin>all-open</plugin>
            </compilerPlugins>
            <pluginOptions>
               <option>all-open:annotation=javax.ws.rs.Path</option>
               <option>all-open:annotation=javax.enterprise.context.ApplicationScoped</option>
               <option>all-open:annotation=io.quarkus.test.junit.QuarkusTest</option>
            </pluginOptions>
         </configuration>
      </plugin>
   </plugins>
</build>

2. Implement REST endpoint

In Quarkus support for REST is built on top of Resteasy and JAX-RS libraries. You can choose between two available extentions for JSON serialization/deserialization: JsonB and Jackson. Since I decided to use Jackson I need to include quarkus-resteasy-jackson dependency. It also includes quarkus-resteasy module.

<dependency>
   <groupId>io.quarkus</groupId>
   <artifactId>quarkus-resteasy-jackson</artifactId>
</dependency>

We mostly use JAX-RS annotations for mapping controller methods and fields into HTTP endpoints. We may also use Resteasy annotations like @PathParam, that does not require to set any fields. In order to interact with database, we are injecting a repository bean.

@Path("/employees")
@Produces(MediaType.APPLICATION_JSON)
@Consumes(MediaType.APPLICATION_JSON)
class EmployeeResource(val repository: EmployeeRepository) {

    @POST
    @Transactional
    fun add(employee: Employee): Response {
        repository.persist(employee)
        return Response.ok(employee).status(201).build()
    }

    @DELETE
    @Path("/{id}")
    @Transactional
    fun delete(@PathParam id: Long) {
        repository.deleteById(id)
    }

    @GET
    fun findAll(): List<Employee> = repository.listAll()

    @GET
    @Path("/{id}")
    fun findById(@PathParam id: Long): Employee? = repository.findById(id)

    @GET
    @Path("/first-name/{firstName}/last-name/{lastName}")
    fun findByFirstNameAndLastName(@PathParam firstName: String, @PathParam lastName: String): List<Employee>
            = repository.findByFirstNameAndLastName(firstName, lastName)

    @GET
    @Path("/salary/{salary}")
    fun findBySalary(@PathParam salary: Int): List<Employee> = repository.findBySalary(salary)

    @GET
    @Path("/salary-greater-than/{salary}")
    fun findBySalaryGreaterThan(@PathParam salary: Int): List<Employee>
            = repository.findBySalaryGreaterThan(salary)

}

3. Integration with database

Quarkus provides Panache JPA extension to simplify work with Hibernate ORM. It also provides driver extensions for the most popular SQL databases like Postgresql, MySQL, or H2. To enable both these features for H2 in-memory database we need to include the following dependencies.

<dependency>
   <groupId>io.quarkus</groupId>
   <artifactId>quarkus-hibernate-orm-panache-kotlin</artifactId>
</dependency>
<dependency>
   <groupId>io.quarkus</groupId>
   <artifactId>quarkus-jdbc-h2</artifactId>
</dependency>

We should also configure connection settings inside application.properties file.

quarkus.datasource.db-kind=h2
quarkus.datasource.username=sa
quarkus.datasource.password=password
quarkus.datasource.jdbc.url=jdbc:h2:mem:testdb

Panache extension allows to use well-known repository pattern. To use it we should first define entity that extends PanacheEntity class.

@Entity
data class Employee(var firstName: String = "",
                    var lastName: String = "",
                    var position: String = "",
                    var salary: Int = 0,
                    var organizationId: Int? = null,
                    var departmentId: Int? = null): PanacheEntity()

In the next step, we are defining repository bean that implements PanacheRepository interface. It comes with some basic methods like persist, deleteById or listAll. We may also use those basic methods to implement more advanced queries or operations.

@ApplicationScoped
class EmployeeRepository: PanacheRepository<Employee> {
    fun findByFirstNameAndLastName(firstName: String, lastName: String): List<Employee> =
           list("firstName = ?1 and lastName = ?2", firstName, lastName)

    fun findBySalary(salary: Int): List<Employee> = list("salary", salary)

    fun findBySalaryGreaterThan(salary: Int): List<Employee> = list("salary > ?1", salary)
}

4. Enable OpenAPI documentation for Quarkus Kotlin

It is possible to generate OpenAPI v3 specification automatically. To do that we need to include SmallRye OpenAPI extension. The specification is available under path /openapi.

<dependency>
   <groupId>io.quarkus</groupId>
   <artifactId>quarkus-smallrye-openapi</artifactId>
</dependency>

We may provide some additional informations to the generated OpenAPI specification like description or version number. To do that we need to create application class that extends javax.ws.rs.core.Application, and annotate it with @OpenAPIDefinition, as shown below.

@OpenAPIDefinition(info = Info(title = "Employee API", version = "1.0"))
class EmployeeApplication: Application()

Usually, we want to expose OpenAPI specification using Swagger UI. Such a feature may be enabled using configuration property quarkus.swagger-ui.always-include=true.

5. Health checks

We may expose built-in health checks implementation by including SmallRye Health extension.

<dependency>
   <groupId>io.quarkus</groupId>
   <artifactId>quarkus-smallrye-health</artifactId>
</dependency>

It exposes three REST endpoints compliant with Kubernetes health checks pattern:

• /health/live – The application is up and running (Kubernetes liveness probe).
• /health/ready – The application is ready to serve requests (Kubernetes readiness probe).
• /health – Accumulating all health check procedures in the application.

The default implementation of readiness health check verifies database connection status, while liveness just determines if the application is running.

6. Expose metrics

We may enable metrics collection by adding SmallRye Metrics extension. By default, it collects only JVM, CPU and processes metrics.

<dependency>
   <groupId>io.quarkus</groupId>
   <artifactId>quarkus-smallrye-metrics</artifactId>
</dependency>

We may force the library to collect metrics from JAX-RS endpoints. To do that we need to annotate the selected endpoints with @Timed.

@POST
@Transactional
@Timed(name = "add", unit = MetricUnits.MILLISECONDS)
fun add(employee: Employee): Response {
   repository.persist(employee)
   return Response.ok(employee).status(201).build()
}

Now, we may call endpoint POST /employee 100 times in a row. Here’s the list of metrics generated for the single endpoint. If you would like to ensure compatibility with Micrometer metrics format you need to set the following configuration property: quarkus.smallrye-metrics.micrometer.compatibility=true.

7. Logging request and response for Quarkus Kotlin application

There is no built-in mechanism for logging HTTP requests and responses. We may implement custom logging filter that implements interfaces ContainerRequestFilter, and ContainerResponseFilter.

@Provider
class LoggingFilter: ContainerRequestFilter, ContainerResponseFilter {

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

    @Context
    lateinit var info: UriInfo
    @Context
    lateinit var request: HttpServerRequest

    override fun filter(ctx: ContainerRequestContext) {
        logger.info("Request {} {}", ctx.method, info.path)
    }

    override fun filter(r: ContainerRequestContext, ctx: ContainerResponseContext) {
        logger.info("Response {} {}: {}", r.method, info.path, ctx.status)
    }
    
}

8. Testing

The module quarkus-junit5 is required for testing, as it provides the @QuarkusTest annotation that controls the testing framework. The extension rest-assured is not required, but is a convenient way to test HTTP endpoints.

<dependency>
   <groupId>io.quarkus</groupId>
   <artifactId>quarkus-junit5</artifactId>
   <scope>test</scope>
</dependency>
<dependency>
   <groupId>io.rest-assured</groupId>
   <artifactId>kotlin-extensions</artifactId>
   <scope>test</scope>
</dependency>

We are adding new Employee in the first test. Then the second test verifies if there is a single Employee stored inside in-memory database.

@QuarkusTest
class EmployeeResourceTest {

    @Test
    fun testAddEmployee() {
        val emp = Employee(firstName = "John", lastName = "Smith", position = "Developer", salary = 20000)
        given().body(emp).contentType(ContentType.JSON)
                .post("/employees")
                .then()
                .statusCode(201)
    }

    @Test
    fun testGetAll() {
        given().get("/employees")
                .then()
                .statusCode(200)
                .assertThat().body("size()", `is`(1))
    }

}

Conclusion

In this guide, I showed you how to build a Quarkus Kotlin application that connects to a database and follows some best practices like exposing health checks, metrics, or logging incoming requests and outgoing responses. The last step is to run our sample application. To do that in development mode we just need to execute command mvn compile quarkus:dev. Here’s my start screen. You can see there, for example, the list of included Quarkus modules.

If you are interested in Quarkus framework the next useful article for you is Guide to Quarkus on Kubernetes.

The post Guide to Quarkus with Kotlin appeared first on Piotr's TechBlog.

Example

As always a source code with examples is available on GitHub. The address of repository is https://github.com/piomin/sample-kotlin-playground.git. It contains example for another article, so for test of coroutines you should take a look on pl.piomin.services.test.CoroutinesTest class.

Implementation

Let’s start by creating a simple coroutine. Of course, there are some different ways to create coroutine, but I’m choosing the simplest one – with GlobalScope. It means that the lifetime of each coroutine is limited only by the lifetime of the whole application or a test as in our case. We are creating coroutine 10 times and after launching it we are printing the name of the currently used thread. We are not calling any suspend method here – I just want to show you how it works.

@Test
fun testSimpleCoroutine() {
   var i: Int = 0
   repeat(10) {
      GlobalScope.launch {
         println("${++i}: ${Thread.currentThread().name}")
      }
   }
   Thread.sleep(100)
}

The order of printing messages is indeterminate. However, you may see that it was using just 3 different thread during test execution. A default thread pool used here is called DefaultDispatcher-worker. It also add the name of coroutine to the thread name.

We may compare it to the result of this test. It does the same thing as the previous test, but uses Java thread instead of Kotlin coroutine.

@Test
fun testSimpleThread() {
   var j: Int = 0
   repeat(10) {
      Thread(Runnable {
         println("${++j}: ${Thread.currentThread().name}")
      }).start()
   }
   Thread.sleep(100)
}

Here’s the result. As you probably expect the order of printing messages is still indeterminate (even more than earlier, but we will discuss it in the next section), but of course there were 10 running threads during the test. In comparison to the previous test it had to create 10 threads instead of 3.

It is said that Kotlin coroutines are processed sequentially. What does it mean in practice? Let’s change the default thread pool used by our coroutines to force them using just a main thread. To do that we need to override default CoroutineDispatcher inside launch method with Dispatchers.Unconfined.

@Test
fun testSimpleCoroutineThreadMain() {
   var i: Int = 0
   repeat(10) {
      GlobalScope.launch(Dispatchers.Unconfined) {
         i++
         println("$i: ${Thread.currentThread().name}")
      }
   }
   Thread.sleep(100)
}

No matter how many times you will run that test, the result will be always the same. The messages are printing in order of launching new coroutines.

And finally we may proceed to the last test. We will call suspension method delay inside our coroutine. We will also print the name of currently used thread before and after calling delay.

@Test
fun testSimpleCoroutineWithDelay() {
   repeat(10) {
      GlobalScope.launch {
         println("Before delay $it: ${Thread.currentThread().name}")
         delay(10)
         println("After delay $it: ${Thread.currentThread().name}")
      }
   }
   Thread.sleep(200)
}

You can easily verify that after suspension the rest of job inside coroutine has been processed in different thread than before.

Summary

In this short article I was trying to explain, using simple words, what exactly is Kotlin coroutine. I hope it helps you to understand the most important differences between threads and Kotlin coroutines.

The post Kotlin Coroutines vs Java Threads appeared first on Piotr's TechBlog.

Example

The source code with sample applications is as usual available on GitHub in the repository: https://github.com/piomin/sample-spring-microservices-transactions.git.

Architecture

First, we need to add a new component to our system. It is responsible just for managing distributed transactions across microservices. That element is described as transaction-server on the diagram below. We also use another popular component in microservices-based architecture discovery-server. There are three applications: order-service, account-service and product-service. The application order-service is communicating with account-service and product-service. All these applications are using Postgres database as a backend store. Just for simplification I have run a single database with multiple tables. In a normal situation we would have a single database per each microservice.

Now, we will consider the following situation (it is visualized on the diagram below). The application order-service is creating an order, storing it in the database, and then starting a new distributed transaction (1). After that, it is communicating with application product-service to update the current number of stored products and get their price (2). At the same time product-service is sending information to transaction-server that it is participating in the transaction (3). Then order-service is trying to withdraw the required funds from the customer account and transfer them into another account related to a seller (4). Finally, we are rolling back the transaction by throwing an exception inside the transaction method from order-service (6). This rollback should cause a rollback of the whole distributed transaction.

Building transaction server

We are starting implementation from transaction-server. A transaction server is responsible for managing distributed transactions across all microservices in our sample system. It exposes REST API available for all other microservices for adding new transactions and updating their status. It also sends asynchronous broadcast events after receiving transaction confirmation or rollback from a source microservice. It uses RabbitMQ message broker for sending events to other microservices via topic exchange. All other microservices are listening for incoming events, and after receiving them they are committing or rolling back transactions. We can avoid using a message broker for exchanging events and use communication over HTTP endpoints, but that makes sense only if we have a single instance of every microservice. Here’s the picture that illustrates the currently described architecture.

Let’s take a look on the list of required dependencies. It would be pretty the same for other applications. We need spring-boot-starter-amqp for integration with RabbitMQ, spring-boot-starter-web for exposing REST API over HTTP, spring-cloud-starter-netflix-eureka-client for integration with Eureka discovery server and some basic Kotlin libraries.

<dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
   <groupId>com.fasterxml.jackson.module</groupId>
   <artifactId>jackson-module-kotlin</artifactId>
</dependency>
<dependency>
   <groupId>org.jetbrains.kotlin</groupId>
   <artifactId>kotlin-reflect</artifactId>
</dependency>
<dependency>
   <groupId>org.jetbrains.kotlin</groupId>
   <artifactId>kotlin-stdlib</artifactId>
</dependency>
<dependency>
   <groupId>org.springframework.cloud</groupId>
   <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
</dependency>

In the main class we are defining a topic exchange for events sent to microservices. The name of exchange is trx-events, and it is automatically created on RabbitMQ after application startup.

@SpringBootApplication
class TransactionServerApp {

    @Bean
    fun topic(): TopicExchange = TopicExchange("trx-events")
}

fun main(args: Array) {
    runApplication(*args)
}

Here are domain model classes used by a transaction server. The same classes are used by the microservices during communication with transaction-server.

data class DistributedTransaction(var id: String? = null,var status: DistributedTransactionStatus,
                                  val participants: MutableList<DistributedTransactionParticipant> = mutableListOf())
                          
class DistributedTransactionParticipant(val serviceId: String, var status: DistributedTransactionStatus)

enum class DistributedTransactionStatus {
    NEW, CONFIRMED, ROLLBACK, TO_ROLLBACK
}   

Here’s the controller class. It is using a simple in-memory implementation of repository and RabbitTemplate for sending events to RabbitMQ. The HTTP API provides methods for adding new transaction, finishing existing transaction with a given status (CONFIRM or ROLLBACK), searching transaction by id and adding participants (new services) into a transaction.

@RestController
@RequestMapping("/transactions")
class TransactionController(val repository: TransactionRepository,
                            val template: RabbitTemplate) {

    @PostMapping
    fun add(@RequestBody transaction: DistributedTransaction): DistributedTransaction =
            repository.save(transaction)

    @GetMapping("/{id}")
    fun findById(@PathVariable id: String): DistributedTransaction? = repository.findById(id)

    @PutMapping("/{id}/finish/{status}")
    fun finish(@PathVariable id: String, @PathVariable status: DistributedTransactionStatus) {
        val transaction: DistributedTransaction? = repository.findById(id)
        if (transaction != null) {
            transaction.status = status
            repository.update(transaction)
            template.convertAndSend("trx-events", DistributedTransaction(id, status))
        }
    }

    @PutMapping("/{id}/participants")
    fun addParticipant(@PathVariable id: String,
                       @RequestBody participant: DistributedTransactionParticipant) =
            repository.findById(id)?.participants?.add(participant)

    @PutMapping("/{id}/participants/{serviceId}/status/{status}")
    fun updateParticipant(@PathVariable id: String,
                          @PathVariable serviceId: String,
                          @PathVariable status: DistributedTransactionStatus) {
        val transaction: DistributedTransaction? = repository.findById(id)
        if (transaction != null) {
            val index = transaction.participants.indexOfFirst { it.serviceId == serviceId }
            if (index != -1) {
                transaction.participants[index].status = status
                template.convertAndSend("trx-events", DistributedTransaction(id, status))
            }
        }
    }

}   

Handling transactions in downstream services

Let’s analyze how our microservices are handling transactions on the example of account. Here’s the implementation of AccountService that is called by the controller for transfering funds from/to account. All methods here are @Transactional and here we need an attention – @Async. It means that each method is running in a new thread and is processing asynchronously. Why? That’s a key concept here. We will block the transaction in order to wait for confirmation from transaction-server, but the main thread used by the controller will not be blocked. It returns the response with the current state of Account immediately.

@Service
@Transactional
@Async
class AccountService(val repository: AccountRepository,
                     var applicationEventPublisher: ApplicationEventPublisher) {
    
    fun payment(id: Int, amount: Int, transactionId: String) =
            transfer(id, amount, transactionId)

    fun withdrawal(id: Int, amount: Int, transactionId: String) =
            transfer(id, (-1) * amount, transactionId)

    private fun transfer(id: Int, amount: Int, transactionId: String) {
        val accountOpt: Optional<Account> = repository.findById(id)
        if (accountOpt.isPresent) {
            val account: Account = accountOpt.get()
            account.balance += amount
            applicationEventPublisher.publishEvent(AccountTransactionEvent(transactionId, account))
            repository.save(account)
        }
    }

}

Here’s the implementation of @Controller class. As you see it is calling methods from AccountService, that are being processed asynchronously. The returned Account object is taken from EventBus bean. This bean is responsible for exchanging asynchronous events within the application scope. En event is sent by the AccountTransactionListener bean responsible for handling Spring transaction events.

@RestController
@RequestMapping("/accounts")
class AccountController(val repository: AccountRepository,
                        val service: AccountService,
                        val eventBus: EventBus) {

    @PostMapping
    fun add(@RequestBody account: Account): Account = repository.save(account)

    @GetMapping("/customer/{customerId}")
    fun findByCustomerId(@PathVariable customerId: Int): List<Account> =
            repository.findByCustomerId(customerId)

    @PutMapping("/{id}/payment/{amount}")
    fun payment(@PathVariable id: Int, @PathVariable amount: Int,
                @RequestHeader("X-Transaction-ID") transactionId: String): Account {
        service.payment(id, amount, transactionId)
        return eventBus.receiveEvent(transactionId)!!.account
    }

    @PutMapping("/{id}/withdrawal/{amount}")
    fun withdrawal(@PathVariable id: Int, @PathVariable amount: Int,
                   @RequestHeader("X-Transaction-ID") transactionId: String): Account {
        service.withdrawal(id, amount, transactionId)
        return eventBus.receiveEvent(transactionId)!!.account
    }

}

The event object exchanged between bean is very simple. It contains an id of transaction and the current Account object.

class AccountTransactionEvent(val transactionId: String, val account: Account)

Finally, let’s take a look at the implementation of AccountTransactionListener bean responsible for handling transactional events. We are using Spring @TransactionalEventListener for annotating methods that should handle incoming events. There are 4 possible event types to handle: BEFORE_COMMIT, AFTER_COMMIT, AFTER_ROLLBACK and AFTER_COMPLETION. There is one very important thing in @TransactionalEventListener, which may be not very intuitive. It is being processed in the same thread as the transaction. So if you would do something that should not block the thread with transaction you should annotate it with @Async. However, in our case this behaviour is required, since we need to block a transactional thread until we receive a confirmation or rollback from transaction-server for a given transaction. These events are sent by transaction-server through RabbitMQ, and they are also exchanged between beans using EventBus. If the status of the received event is different than CONFIRMED we are throwing the exception to rollback transaction.
The AccountTransactionListener is also listening on AFTER_ROLLBACK and AFTER_COMPLETION. After receiving such an event type it is changing the status of the transaction by calling endpoint exposed by transaction-server.

@Component
class AccountTransactionListener(val restTemplate: RestTemplate,
                                 val eventBus: EventBus) {

    @TransactionalEventListener(phase = TransactionPhase.BEFORE_COMMIT)
    @Throws(AccountProcessingException::class)
    fun handleEvent(event: AccountTransactionEvent) {
        eventBus.sendEvent(event)
        var transaction: DistributedTransaction? = null
        for (x in 0..100) {
            transaction = eventBus.receiveTransaction(event.transactionId)
            if (transaction == null)
                Thread.sleep(100)
            else break
        }
        if (transaction == null || transaction.status != DistributedTransactionStatus.CONFIRMED)
            throw AccountProcessingException()
    }

    @TransactionalEventListener(phase = TransactionPhase.AFTER_ROLLBACK)
    fun handleAfterRollback(event: AccountTransactionEvent) {
        restTemplate.put("http://transaction-server/transactions/transactionId/participants/{serviceId}/status/{status}",
                null, "account-service", "TO_ROLLBACK")
    }

    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMPLETION)
    fun handleAfterCompletion(event: AccountTransactionEvent) {
        restTemplate.put("http://transaction-server/transactions/transactionId/participants/{serviceId}/status/{status}",
                null, "account-service", "CONFIRM")
    }
    
}

Here’s the implementation of the bean responsible for receiving asynchronous events from a message broker. As you see after receiving such an event it is using EventBus to forward that event to other beans.

@Component
class DistributedTransactionEventListener(val eventBus: EventBus) {

    @RabbitListener(bindings = [
        QueueBinding(exchange = Exchange(type = ExchangeTypes.TOPIC, name = "trx-events"),
                value = Queue("trx-events-account"))
    ])
    fun onMessage(transaction: DistributedTransaction) {
        eventBus.sendTransaction(transaction)
    }

}

Integration with database

Of course our application is using Postgres as a backend store, so we need to provide integration. In fact, that is the simplest step of our implementation. First we need to add the following 2 dependencies. We will use Spring Data JPA for integration with Postgres.

<dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-data-jpa</artifactId>
</dependency>
<dependency>
   <groupId>org.postgresql</groupId>
   <artifactId>postgresql</artifactId>
   <scope>runtime</scope>
</dependency>

Our entity is very simple. Besides the id field it contains two fields: customerId and balance.

@Entity
data class Account(@Id @GeneratedValue(strategy = GenerationType.AUTO) val id: Int,
                   val customerId: Int,
                   var balance: Int)

We are using the well-known Spring Data repository pattern.

interface AccountRepository: CrudRepository<Account, Int> {

    fun findByCustomerId(id: Int): List<Account>

}

Here’s the suggested list of configuration settings.

spring:
  application:
    name: account-service
  datasource:
    url: jdbc:postgresql://postgresql:5432/trx
    username: trx
    password: trx
    hikari:
      connection-timeout: 2000
      initialization-fail-timeout: 0
  jpa:
    database-platform: org.hibernate.dialect.PostgreSQLDialect
    hibernate:
      ddl-auto: create
    show-sql: true
    properties:
      hibernate:
        format_sql: true
  rabbitmq:
    host: rabbitmq
    port: 5672
    connection-timeout: 2000

Building order-service

Ok, we have already finished the implementation of transaction-server, and two microservices account-service and product-service. Since the implementation of product-service is very similar to account-service, I have explained everything on the example of account-service. Now, we may proceed to the last part – the implementation of order-service. It is responsible for starting a new transaction and marking it as finished. It also may finish it with rollback.Of course, rollback events may be sent by another two applications as well.
The implementation of @Controller class is visible below. I’ll describe it step by step. We are starting a new distributed transaction by calling POST /transactions endpoint exposed by transaction-server (1). Then we are storing a new order in database (2). When we are calling a transactional method from downstream service we need to set HTTP header X-Transaction-ID. The first transactional method that is called here is PUT /products/{id}/count/{count}(3). It updates the number of products in the store and calculates a final price (4). In the step it is calling another transaction method – this time from account-service (5). It is responsible for withdrawing money from customer accounts. We are enabling Spring transaction events processing (6). In the last step we are generating a random number, and then basing on its value application is throwing an exception to rollback transaction (7).

@RestController
@RequestMapping("/orders")
class OrderController(val repository: OrderRepository,
                      val restTemplate: RestTemplate,
          var applicationEventPublisher: ApplicationEventPublisher) {

    @PostMapping
    @Transactional
    @Throws(OrderProcessingException::class)
    fun addAndRollback(@RequestBody order: Order) {
        var transaction  = restTemplate.postForObject("http://transaction-server/transactions",
                DistributedTransaction(), DistributedTransaction::class.java) // (1)
        val orderSaved = repository.save(order) // (2)
        val product = updateProduct(transaction!!.id!!, order) // (3)
        val totalPrice = product.price * product.count // (4)
        val accounts = restTemplate.getForObject("http://account-service/accounts/customer/{customerId}",
                Array<Account>::class.java, order.customerId)
        val account  = accounts!!.first { it.balance >= totalPrice}
        updateAccount(transaction.id!!, account.id, totalPrice) // (5)
        applicationEventPublisher.publishEvent(OrderTransactionEvent(transaction.id!!)) // (6)
        val r = Random.nextInt(100) // (7)
        if (r % 2 == 0)
            throw OrderProcessingException()
    }

    fun updateProduct(transactionId: String, order: Order): Product {
        val headers = HttpHeaders()
        headers.set("X-Transaction-ID", transactionId)
        val entity: HttpEntity<*> = HttpEntity<Any?>(headers)
        val product = restTemplate.exchange("http://product-service/products/{id}/count/{count}",
                HttpMethod.PUT, null, Product::class.java, order.id, order.count)
        return product.body!!
    }

    fun updateAccount(transactionId: String, accountId: Int, totalPrice: Int): Account {
        val headers = HttpHeaders()
        headers.set("X-Transaction-ID", transactionId)
        val entity: HttpEntity<*> = HttpEntity<Any?>(headers)
        val account = restTemplate.exchange("http://account-service/accounts/{id}/withdrawal/{amount}",
                HttpMethod.PUT, null, Account::class.java, accountId, totalPrice)
        return account.body!!
    }
}

Conclusion

Even a trivial implementation of distributed transactions in microservices, like the one, demonstrated in this article, can be complicated. As you see we need to add a new element to our architecture, transaction-server, responsible only for distributed transaction management. We also have to add a message broker in order to exchange events between our applications and transaction-server. However, many of you were asking me about distributed transactions in the microservices world, so I decided to build that simple demo. I’m waiting for your feedback and opinions.

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

If you are interested in more detailed explanation of Spring Cloud components used for inter-service communication you should refer to the third part of my online course Microservices With Spring Boot And Spring Cloud: Part 3 – Inter-service communication.

Example

You can find a source code snippets related to this article in my GitHub repository https://github.com/piomin/course-spring-microservices.git. That repository is also used for my online course, so I decided to extend it with the new examples. All the required changes were performed in directory inter-communication/inter-caller-service inside that repository. The code is written in Kotlin.
There are three applications, which are a part of our sample system: discovery-server (Spring Cloud Netflix Eureka), inter-callme-service (Spring Boot application that expose REST API), and finally inter-caller-service (Spring Boot application that calls endpoints exposed by inter-callme-service).

How to start with Spring Cloud Load Balancer

To enable Spring Cloud Load Balancer for our application we first need to include the following starter to Maven dependencies. That module may be also included together with some other Spring Cloud starters.

<dependency>
	<groupId>org.springframework.cloud</groupId>
	<artifactId>spring-cloud-starter-loadbalancer</artifactId>
</dependency>

Because Ribbon is still used as a default client-side load balancer for REST-based communication between applications we need to disable it in application properties. Here’s a fragment of application.yml file.

spring:
  application:
    name: inter-caller-service
  cloud:
    loadbalancer:
      ribbon:
        enabled: false

For discovery integration we also need to include spring-cloud-starter-netflix-eureka-client. To use RestTemplate with a client-side load balancer we should define the bean visible below and annotate it with @LoadBalanced. As you on the code below I’m also setting interceptor on RestTemplate, but more about it in the next section.

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

Adapt traffic to average response time

Spring Cloud Load Balancer provides a simple round robin rule for load balancing between multiple instances of a single service. Our goal here is to implement a rule, which measures each application response time and gives a weight according to that time. The longer the response time, the less weight it will get. The rule should randomly pick an instance where the possibility is determined by its weight. To record response time of each call we need to set an already mentioned interceptor that implements ClientHttpRequestInterceptor. Interceptor is executed on every request (1). Since the implementation is very typical, one line requires explanation (2). I’m getting the address of the target application from a thread scoped variable existing in Slf4J MDC. Of course I could also implement a simple thread scoped context based on ThreadLocal, but MDC is used here just for simplification.

class ResponseTimeInterceptor(private val responseTimeHistory: ResponseTimeHistory) : ClientHttpRequestInterceptor {

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

    override fun intercept(request: HttpRequest, array: ByteArray,
                           execution: ClientHttpRequestExecution): ClientHttpResponse {
        val startTime: Long = System.currentTimeMillis()
        val response: ClientHttpResponse = execution.execute(request, array) // 1
        val endTime: Long = System.currentTimeMillis()
        val responseTime: Long = endTime - startTime
        logger.info("Response time: instance->{}, time->{}", MDC.get("address"), responseTime)
        responseTimeHistory.addNewMeasure(MDC.get("address"), responseTime) // 2
        return response
    }
}

Of course, counting an average response time is just a part of our job. The most important is the implementation of our custom load balancer, which is visible below. It should implement interface ReactorServiceInstanceLoadBalancer. It need to inject ServiceInstanceListSupplier bean to fetch a list of available instances of a given service in overridden method choose. While choosing the right instance we are analyzing the average response time for each instance saved in ResponseTimeHistory by ResponseTimeInterceptor. In the beginning our load balancer acts like a simple round robin.

class WeightedTimeResponseLoadBalancer(
        private val serviceInstanceListSupplierProvider: ObjectProvider<ServiceInstanceListSupplier>,
        private val serviceId: String,
        private val responseTimeHistory: ResponseTimeHistory) : ReactorServiceInstanceLoadBalancer {

    private val logger: Logger = LoggerFactory.getLogger(WeightedTimeResponseLoadBalancer::class.java)
    private val position: AtomicInteger = AtomicInteger()

    override fun choose(request: Request<*>?): Mono<Response<ServiceInstance>> {
        val supplier: ServiceInstanceListSupplier = serviceInstanceListSupplierProvider
                .getIfAvailable { NoopServiceInstanceListSupplier() }
        return supplier.get().next()
                .map { serviceInstances: List<ServiceInstance> -> getInstanceResponse(serviceInstances) }
    }

    private fun getInstanceResponse(instances: List<ServiceInstance>): Response<ServiceInstance> {
        return if (instances.isEmpty()) {
            EmptyResponse()
        } else {
            val address: String? = responseTimeHistory.getAddress(instances.size)
            val pos: Int = position.incrementAndGet()
            var instance: ServiceInstance = instances[pos % instances.size]
            if (address != null) {
                val found: ServiceInstance? = instances.find { "${it.host}:${it.port}" == address }
                if (found != null)
                    instance = found
            }
            logger.info("Current instance: [address->{}:{}, stats->{}ms]", instance.host, instance.port,
                    responseTimeHistory.stats["${instance.host}:${instance.port}"])
            MDC.put("address", "${instance.host}:${instance.port}")
            DefaultResponse(instance)
        }
    }
}

Here’s the implementation of ResponseTimeHistory bean, which is responsible for storing measures and selecting the instance of service based on computed weight.

class ResponseTimeHistory(private val history: MutableMap<String, Queue<Long>> = mutableMapOf(),
                          val stats: MutableMap<String, Long> = mutableMapOf()) {

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

    fun addNewMeasure(address: String, measure: Long) {
        var list: Queue<Long>? = history[address]
        if (list == null) {
            history[address] = LinkedList<Long>()
            list = history[address]
        }
        logger.info("Adding new measure for->{}, measure->{}", address, measure)
        if (measure == 0L)
            list!!.add(1L)
        else list!!.add(measure)
        if (list.size > 9)
            list.remove()
        stats[address] = countAvg(address)
        logger.info("Counting avg for->{}, stat->{}", address, stats[address])
    }

    private fun countAvg(address: String): Long {
        val list: Queue<Long>? = history[address]
        return list?.sum()?.div(list.size) ?: 0
    }

    fun getAddress(numberOfInstances: Int): String? {
        if (stats.size < numberOfInstances)
            return null
        var sum: Long = 0
        stats.forEach { sum += it.value }
        var r: Long = Random.nextLong(100)
        var current: Long = 0
        stats.forEach {
            val weight: Long = (sum - it.value)*100 / sum
            logger.info("Weight for->{}, value->{}, random->{}", it.key, weight, r)
            current += weight
            if (r <= current)
                return it.key
        }
        return null
    }

}

Customizing Spring Cloud Load Balancer

The implementation of our mechanism for weighted response time rule is ready, so the last step is to apply it to Spring Cloud Load Balancer. To do that we need to create a dedicated configuration class with ReactorLoadBalancer bean declaration as shown below.

class CustomCallmeClientLoadBalancerConfiguration(private val responseTimeHistory: ResponseTimeHistory) {

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

The custom configuration may be passed to a load balancer using annotation @LoadBalancerClient. The name of client should be the same as registered in discovery. This part of code is currently commented out in the GitHub repository, so if you would like to enable it for testing just uncomment it.

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

    @Bean
    fun responseTimeHistory(): ResponseTimeHistory = ResponseTimeHistory()

    @Bean
    fun responseTimeInterceptor(): ResponseTimeInterceptor = ResponseTimeInterceptor(responseTimeHistory())

    // THE REST OF IMPLEMENTATION...
}

Customizing instance list supplier

Currently Spring Cloud Load Balancer does not support a static list of instances set in configuration properties (unlike Netflix Ribbon). We can easily add such a mechanism. The static list of instances for every service will be defined as shown below.

spring:
  application:
    name: inter-caller-service
  cloud:
    loadbalancer:
      ribbon:
        enabled: false
      instances:
        - name: inter-callme-service
          servers: localhost:59600, localhost:59800

As the first step, we should define a class that implements interface ServiceInstanceListSupplier and overrides two methods: getServiceId() and get(). The following implementation of ServiceInstanceListSupplier takes the list of service addresses from application properties through @ConfigurationProperties.

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

    override fun getServiceId(): String = environment.getProperty("loadbalancer.client.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)
    }

}

Here’s the implementation of configuration class with properties.

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

    val instances: MutableList<ServiceConfig> = mutableListOf()

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

}

The same as for the previous sample we should also register our implementation of ServiceInstanceListSupplier as a bean inside custom configuration class.

class CustomCallmeClientLoadBalancerConfiguration) {

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

Testing Spring Cloud Load Balancer

To test the solution implemented for the purpose of this article you should:

1. Run the instance of discovery server (only if StaticServiceInstanceListSupplier is disabled)
2. Run two instances of inter-callme-service (for one selected instance activate random delay using VM parameter -Dspring.profiles.active=delay)
3. Run instance of inter-caller-service, which is available on port 8080
4. Send some test requests to inter-caller-service using command, for example curl -X POST http://localhost:8080/caller/random-send/12345

Our test scenario is visualized in the following picture.

Conclusion

Currently, Spring Cloud Load Balancer does not offer such many interesting features for inter-service communication as the Netflix Ribbon client. Of course, it is still being actively developed by the Spring Team. The good news is that we can easily customize Spring Cloud Load Balancer to add some custom features. In this article I demonstrated how to provide more advanced load balancing algorithms or create custom instances of list suppliers.

The post A Deep Dive Into Spring Cloud Load Balancer appeared first on Piotr's TechBlog.

Example

The source code repository used for this article is available on my GitHub in repository sample-kotlin-playground. It is available here: https://github.com/piomin/sample-kotlin-playground.git.

Apply scope function

Apply refer to the context object as a lambda receiver – by keyword this. It is the same as “you would be inside” the class of that object. You can even omit the keyword this. As a results it just returns the context object. All these features makes it ideal for changing value of object fields. Let’s take a look on the following example of code. Then we will try to implement the same functionality using other scope function.

@Test fun testApplyFunction() {
  val p: Person = Person("John", "Smith", 1)
    .apply {
      age = 20
      localization = "London"
    }
  Assert.assertEquals(20, p.age)
  Assert.assertEquals("London", p.localization)
}

Let

Let is able to access the context object as a lambda argument. If the argument name is not specified, the object is accessed by the implicit default name it. It is returning the lambda result. What means that if we want to do the same thing with person object as in the previous sample we need to return exactly that object.

@Test fun testLetFunction() {
  val p: Person = Person("John", "Smith", 1)
    .let {
      it.age = 20
      it.localization = "London"
      it
    }
  Assert.assertEquals(20, p.age)
  Assert.assertEquals("London", p.localization)
}

Run scope function

Run the same as apply function refers to the context object as a lambda receiver. But unlike apply it returns the lambda result instead of the context object. The test is very similar to the testApplyFunction – the only difference is that we have to return object using this keyword.

@Test fun testRunFunction() {
  val p: Person = Person("John", "Smith", 1)
    .run {
      age = 20
      localization = "London"
      this
    }
  Assert.assertEquals(20, p.age)
  Assert.assertEquals("London", p.localization)
}

Also scope function

We can use also function to do some additional work after previous changes. Let’s implement our test with apply in slightly different way. Function also returns the context object and takes the context object as a lambda argument.

@Test fun testAlsoFunction() {
  val p: Person = Person("John", "Smith", 1)
    .apply {
      age = 20
      localization = "London"
    }
    .also {
      Assert.assertEquals(20, it.age)
      Assert.assertEquals("London", it.localization)
    }
}

TakeIf scope function

TakeIf is not a scope function. It is in addition provided by the Kotlin standard library together with takeUnless. While takeIf returns this object if it matches the predicate, takeUnless returns the object if it doesn’t match the predicate and null if it does. Let’s take a look on the simple test that illustrates usage of this function.

@Test fun testTakeIfFunction() {
  var p: Person? = Person("John", "Smith", 1).takeIf { it.id > 1 }
  Assert.assertNull(p)
  p = Person("John", "Smith", 1).takeIf { it.id == 1 }
  Assert.assertNotNull(p)
}

The post Kotlin Scope Functions appeared first on Piotr's TechBlog.

Example

The sample Kotlin application as always is available on GitHub. Here’s the URL to the repository: https://github.com/piomin/sample-spring-kotlin-microservice.git.

1. Configure kubectl to connect GKE

Before the exercise, I have created a project My First Project and a single-node Kubernetes cluster on Google Cloud Platform. You can login to your using web console available under address https://console.cloud.google.com. Then you should navigate to your cluster available as Kubernetes Engine service and copy command responsible for configuring kubectl to connect with your cluster. After running gcloud command responsible for it you may check out a newly created kubectl context by running command kubectl config get-context. It displays all previously used kubectl contexts and marks the currently used context.
All the described steps are visible below.

2. Add Skaffold to the project

A perfect solution that simplifies building and deploying applications on Kubernetes is Skaffold. For JVM-based applications, you may use it together with Jib Maven Plugin. Both these tools are provided by Google. Skaffold is a command-line tool that facilitates continuous development for Kubernetes applications. Jib is designed to build optimized Docker and OCI images for your Java applications without a Docker daemon.
To enable Skaffold for a project we need to create skaffold.yaml file in the root directory. Skaffold has many configuration settings, but a standard scenario does not require much. Here’s the skaffold.yaml for our current project. It sets the name of a Docker image and activates a Jib plugin.

We also need to add Jib Plugin to our Maven or Gradle configuration file. Here’s pom.xml for our project.

If you put all the Kubernetes manifests in k8s directory you don’t need to provide any other configuration. Let’s take a look at the described operations.

3. Create Kotlin microservice

We are building a simple web application written in Kotlin. It is built on top of Spring Boot. It exposes HTTP API for managing Person objects, Swagger API documentation and Spring Boot Actuator endpoints. We are using the newest stable version of Spring Boot 2.2.6.RELEASE, and 1.3.70 version of Kotlin.

Here’s the @Controller implementation responsible for handling HTTP API requests.

Let’s take a look at more details.

4. Build/Deploy with Skaffold and Jib

Since we finished an implementation of the sample Kotlin application we may build and deploy it on Google Kubernetes Engine. Thanks to Skaffold we just need to run command skaffold dev --port-forward in the project root directory. Before running that command you should have configured credentials to your Docker Hub repository. Skaffold try to push the Docker image with the application before deploying it on GKE. The credentials are stored in file $HOME_DIR/.docker/config.json.

5. Verify on Google Kubernetes Engine

If Skaffold command finished successfully, we may verify the results. The following video illustrates that a new image piomin/sample-spring-kotlin-microservice with the application has been pushed to my Docker Hub repository, and new deployment and service has been created on GKE. Thanks to Skaffold --port-forward we may easily test locally application running on Google Cloud Platform. We may call Actuator endpoint http://localhost:8080/actuator/info.

Conclusion

Kotlin simplifies and speeds up building Spring Boot applications. Skaffold and Jib simplify building and running containers on Kubernetes, also when you are using a remote cluster available on Google Cloud. With that powerful stack of tools, you can easily develop and run microservices in the cloud environment.

The post Running Kotlin Microservice on Google Kubernetes Engine appeared first on Piotr's TechBlog.

1. Spring Boot Kotlin configuration and dependencies

To use Kotlin in your Maven project you have to include plugin kotlin-maven-plugin, and /src/main/kotlin, /src/test/kotlin directories to the build configuration. We will also set the -Xjsr305 compiler flag to strict. This option is responsible for checking support for JSR-305 annotations (for example @NotNull annotation).

<build>
   <sourceDirectory>${project.basedir}/src/main/kotlin</sourceDirectory>
   <testSourceDirectory>${project.basedir}/src/test/kotlin</testSourceDirectory>
   <plugins>
      <plugin>
         <groupId>org.jetbrains.kotlin</groupId>
         <artifactId>kotlin-maven-plugin</artifactId>
         <configuration>
            <args>
               <arg>-Xjsr305=strict</arg>
            </args>
            <compilerPlugins>
               <plugin>spring</plugin>
            </compilerPlugins>
         </configuration>
         <dependencies>
            <dependency>
               <groupId>org.jetbrains.kotlin</groupId>
               <artifactId>kotlin-maven-allopen</artifactId>
               <version>${kotlin.version}</version>
            </dependency>
         </dependencies>
      </plugin>
   </plugins>
</build>

We should also include some core Kotlin libraries like kotlin-stdlib-jdk8 and kotlin-reflect. They are provided by default for a Kotlin project on start.spring.io. For REST-based applications you will also need Jackson library used for JSON serialization/deserialization. Of course, we have to include Spring starters for Web application together with Actuator responsible for providing management endpoints.

<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>com.fasterxml.jackson.module</groupId>
   <artifactId>jackson-module-kotlin</artifactId>
</dependency>
<dependency>
   <groupId>org.jetbrains.kotlin</groupId>
   <artifactId>kotlin-reflect</artifactId>
</dependency>
<dependency>
   <groupId>org.jetbrains.kotlin</groupId>
   <artifactId>kotlin-stdlib-jdk8</artifactId>
</dependency>

We use the latest stable version of Spring Boot with Kotlin 1.2.71

<parent>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-parent</artifactId>
   <version>2.1.2.RELEASE</version>
</parent>
<properties>
   <java.version>1.8</java.version>
   <kotlin.version>1.2.71</kotlin.version>
</properties>

2. Building Spring Boot Kotlin microservice application

Let’s begin from the basics. If you are familiar with Spring Boot and Java, the biggest difference is in the main class declaration. You will call runApplication method outside the Spring Boot application class. The main class, the same as in Java, is annotated with @SpringBootApplication.

@SpringBootApplication
class SampleSpringKotlinMicroserviceApplication

fun main(args: Array<String>) {
    runApplication<SampleSpringKotlinMicroserviceApplication>(*args)
}

Our sample application is very simple. It exposes some REST endpoints providing CRUD operations for a model object. Even at this fragment of code illustrating controller implementation you can see some nice Kotlin features. We may use shortened function declaration with inferred return type. Annotation @PathVariable does not require any arguments. The input parameter name is considered to be the same as the variable name. Of course, we are using the same annotations as with Java. In Kotlin, every property declared as having a non-null type must be initialized in the constructor. So, if you are initializing it using dependency injection it has to be declared as lateinit. Here’s the implementation of PersonController.

@RestController
@RequestMapping("/persons")
class PersonController {

    @Autowired
    lateinit var repository: PersonRepository

    @GetMapping("/{id}")
    fun findById(@PathVariable id: Int): Person? = repository.findById(id)

    @GetMapping
    fun findAll(): List<Person> = repository.findAll()

    @PostMapping
    fun add(@RequestBody person: Person): Person = repository.save(person)

    @PutMapping
    fun update(@RequestBody person: Person): Person = repository.update(person)

    @DeleteMapping("/{id}")
    fun remove(@PathVariable id: Int): Boolean = repository.removeById(id)

}

Kotlin automatically generates getters and setters for class properties declared as var. Also if you declare a model as a data class it generates equals, hashCode, and toString methods. The declaration of our model class Person is very concise as shown below.

data class Person(var id: Int?, var name: String, var age: Int, var gender: Gender)

I have implemented my own in-memory repository class. I use Kotlin extensions for manipulating a list of elements. This built-in Kotlin feature is similar to Java streams, with the difference that you don’t have to perform any conversion between Collection and Stream.

@Repository
class PersonRepository {
    val persons: MutableList<Person> = ArrayList()

    fun findById(id: Int): Person? {
        return persons.singleOrNull { it.id == id }
    }

    fun findAll(): List<Person> {
        return persons
    }

    fun save(person: Person): Person {
        person.id = (persons.maxBy { it.id!! }?.id ?: 0) + 1
        persons.add(person)
        return person
    }

    fun update(person: Person): Person {
        val index = persons.indexOfFirst { it.id == person.id }
        if (index >= 0) {
            persons[index] = person
        }
        return person
    }

    fun removeById(id: Int): Boolean {
        return persons.removeIf { it.id == id }
    }

}

The sample application source code is available on GitHub in repository https://github.com/piomin/sample-spring-kotlin-microservice.git.

3. Enabling Spring Boot Actuator endpoints

Since we have already included Spring Boot starter with Actuator into the application code, we can take advantage of its production-ready features. Spring Boot Actuator gives you very powerful tools for monitoring and managing your apps. You can provide advanced healthchecks, info endpoints or send metrics to numerous monitoring systems like InfluxDB. After including Actuator artifacts the only thing we have to do is to enable all its endpoints for our application via HTTP.

management.endpoints.web.exposure.include: '*'

We can customize Actuator endpoints to provide more details about our app. A good practice is to expose information about version and git commit to info endpoint. As usual Spring Boot provides auto-configuration for such features, so the only thing we need to do is to include some Maven plugins to build configuration in pom.xml. The goal build-info set for spring-boot-maven-plugin forces it to generate a properties file with basic information about the version. The file is located in directory META-INF/build-info.properties. Plugin git-commit-id-plugin will generate git.properties file in the root directory.

<plugin>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-maven-plugin</artifactId>
   <executions>
      <execution>
         <goals>
            <goal>build-info</goal>
         </goals>
      </execution>
   </executions>
</plugin>
<plugin>
   <groupId>pl.project13.maven</groupId>
   <artifactId>git-commit-id-plugin</artifactId>
   <configuration>
      <failOnNoGitDirectory>false</failOnNoGitDirectory>
   </configuration>
</plugin>

Now you should just build your application using mvn clean install command and then run it.

$ java -jar target\sample-spring-kotlin-microservice-1.0-SNAPSHOT.jar

The info endpoint is available under address http://localhost:8080/actuator/info. It exposes all interesting information for us.

{
   "git":{
      "commit":{
         "time":"2019-01-14T16:20:31Z",
         "id":"f7cb437"
      },
      "branch":"master"
   },
   "build":{
      "version":"1.0-SNAPSHOT",
      "artifact":"sample-spring-kotlin-microservice",
      "name":"sample-spring-kotlin-microservice",
      "group":"pl.piomin.services",
      "time":"2019-01-15T09:18:48.836Z"
   }
}

4. Enabling Swagger API documentation

Build info and git properties may be easily injected into the application code. It can be useful in some cases. One of those cases is if you have enabled auto-generated API documentation. The most popular tool to use for it is Swagger. You can easily integrate Swagger2 with Spring Boot using the SpringFox Swagger project. First, you need to include the following dependencies to your pom.xml.

<dependency>
   <groupId>io.springfox</groupId>
   <artifactId>springfox-swagger2</artifactId>
   <version>2.9.2</version>
</dependency>
<dependency>
   <groupId>io.springfox</groupId>
   <artifactId>springfox-swagger-ui</artifactId>
   <version>2.9.2</version>
</dependency>

Then, you should enable Swagger by annotating configuration class with @EnableSwagger2. Required information is available inside beans BuildProperties and GitProperties. We just have to inject them into Swagger configuration class as shown below. We set them as optional to prevent application startup failure in case they are not present on classpath.

@Configuration
@EnableSwagger2
class SwaggerConfig {

    @Autowired
    lateinit var build: Optional<BuildProperties>
    @Autowired
    lateinit var git: Optional<GitProperties>

    @Bean
    fun api(): Docket {
        var version = "1.0"
        if (build.isPresent && git.isPresent) {
            var buildInfo = build.get()
            var gitInfo = git.get()
            version = "${buildInfo.version}-${gitInfo.shortCommitId}-${gitInfo.branch}"
        }
        return Docket(DocumentationType.SWAGGER_2)
                .apiInfo(apiInfo(version))
                .select()
                .apis(RequestHandlerSelectors.any())
                .paths{ it.equals("/persons")}
                .build()
                .useDefaultResponseMessages(false)
                .forCodeGeneration(true)
    }

    @Bean
    fun uiConfig(): UiConfiguration {
        return UiConfiguration(java.lang.Boolean.TRUE, java.lang.Boolean.FALSE, 1, 1, ModelRendering.MODEL, java.lang.Boolean.FALSE, DocExpansion.LIST, java.lang.Boolean.FALSE, null, OperationsSorter.ALPHA, java.lang.Boolean.FALSE, TagsSorter.ALPHA, UiConfiguration.Constants.DEFAULT_SUBMIT_METHODS, null)
    }

    private fun apiInfo(version: String): ApiInfo {
        return ApiInfoBuilder()
                .title("API - Person Service")
                .description("Persons Management")
                .version(version)
                .build()
    }

}

The documentation is available under context path /swagger-ui.html. Besides API documentation it displays the full information about application version, git commit id and branch name.

5. Choosing your app server

Spring Boot Web can be run on three different embedded servers: Tomcat, Jetty or Undertow. By default it uses Tomcat. To change the default server you just need include the suitable Spring Boot starter and exclude spring-boot-starter-tomcat. The good practice may be to enable switching between servers during application build. You can achieve it by declaring Maven profiles as shown below.

<profiles>
   <profile>
      <id>tomcat</id>
      <activation>
         <activeByDefault>true</activeByDefault>
      </activation>
      <dependencies>
         <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
         </dependency>
      </dependencies>
   </profile>
   <profile>
      <id>jetty</id>
      <dependencies>
         <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
            <exclusions>
               <exclusion>
                  <groupId>org.springframework.boot</groupId>
                  <artifactId>spring-boot-starter-tomcat</artifactId>
               </exclusion>
            </exclusions>
         </dependency>
         <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-jetty</artifactId>
         </dependency>
      </dependencies>
   </profile>
   <profile>
      <id>undertow</id>
      <dependencies>
         <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
            <exclusions>
               <exclusion>
                  <groupId>org.springframework.boot</groupId>
                  <artifactId>spring-boot-starter-tomcat</artifactId>
               </exclusion>
            </exclusions>
         </dependency>
         <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-undertow</artifactId>
         </dependency>
      </dependencies>
   </profile>
</profiles>

Now, if you would like to enable another server than Tomcat for your application you should activate the appropriate profile during Maven build.

$ mvn clean install -Pjetty

Conclusion

Development of microservices using Kotlin and Spring Boot is nice and simple. Based on the sample application I have introduced the main Spring Boot features for Kotlin. I also described some good practices you may apply to your microservices when building it using Spring Boot and Kotlin. You can compare described approach with some other micro-frameworks used with Kotlin, for example Ktor described in one of my previous articles Kotlin Microservices with Ktor.

The post Kotlin Microservice with Spring Boot appeared first on Piotr's TechBlog.

I have already described some aspects related to reactive programming with Spring based on Spring WebFlux and Spring Data JDBC projects in the following articles:

• Reactive Microservices with Spring WebFlux and Spring Cloud
• Introduction to Reactive APIs with Postgres, R2DBC, Spring Data JDBC and Spring WebFlux

Spring Framework supports reactive programming since version 5. That support is built on top of Project Reactor – https://projectreactor.io. Reactor is a fourth-generation Reactive programming library for building non-blocking applications on the JVM based on the Reactive Streams Specification. Working with this library can be difficult at first, especially if you don’t have any experience with reactive streams. Reactive Core gives us two data types that enable us to produce a stream of data: Mono and Flux. With Flux we can emit 0..nelements, while with Mono we can create a stream of 0..1elements. Both those types implement Publisher interface. Both these types are lazy, which means they won’t be executed until you consume it. Therefore, when building reactive APIs it is important not to block the stream. Spring WebFlux doesn’t allow that.

Introduction

The sample project is available on GitHub in repository reactive-playground https://github.com/piomin/reactive-playground.git. It is written in Kotlin. In addition to some Kotlin libraries, only a single dependency that needs to be added in order to use Project Reactor is reactor-core.

<dependency>
   <groupId>io.projectreactor</groupId>
   <artifactId>reactor-core</artifactId>
   <version>3.2.1.RELEASE</version>
</dependency>

I would not like to show you the features of Project Reactor based on simple String objects like in many other articles. Therefore, I have created the following class hierarchy for our tests, that allows us to simulate APIs built for three different domain objects.

Class Organization contains a list of Employee and Department. Each department contains a list of Employee assigned only to the given department inside the organization. Class Employee has properties: organizationId that assigns it to the organization and departmentId that assigns it to the department.

data class Employee(var id: Int, var name: String, var salary: Int) {
    var organizationId: Int? = null
    var departmentId: Int? = null

    constructor(id: Int, name: String, salary: Int, organizationId: Int, departmentId: Int) : this(id, name, salary) {
        this.organizationId = organizationId
        this.departmentId = departmentId
    }

    constructor(id: Int, name: String, salary: Int, organizationId: Int) : this(id, name, salary) {
        this.organizationId = organizationId
    }
}

Here’s the implementation of Department class.

class Department(var id: Int, var name: String, var organizationId: Int) {
    var employees: MutableList<Employee> = ArrayList()

    constructor(id: Int, name: String, organizationId: Int, employees: MutableList<Employee>) : this(id, name, organizationId) {
        this.employees.addAll(employees)
    }

    fun addEmployees(employees: MutableList<Employee>) : Department {
        this.employees.addAll(employees)
        return this
    }

    fun addEmployee(employee: Employee) : Department {
        this.employees.add(employee)
        return this
    }

}

Here’s the implementation of Organization class.

class Organization(var id: Int, var name: String) {
    var employees: MutableList<Employee> = ArrayList()
    var departments: MutableList<Department> = ArrayList()

    constructor(id: Int, name: String, employees: MutableList<Employee>, departments: MutableList<Department>) : this(id, name){
        this.employees.addAll(employees)
        this.departments.addAll(departments)
    }

    constructor(id: Int, name: String, employees: MutableList<Employee>) : this(id, name){
        this.employees.addAll(employees)
    }
}

Scenario 1

We have API methods that return data streams. First of them return Flux emitting employees assigned to the given organization. Second of them just returns Mono with the current organization.

private fun getOrganizationByName(name: String) : Mono<Organization> {
   return Mono.just(Organization(1, name))
}

private fun getEmployeesByOrganization(id: Int) : Flux<Employee> {
   return Flux.just(Employee(1, "Employee1", 1000, id),
                Employee(2, "Employee2", 2000, id))
}

We would like to return the single stream emitting organization that contains a list of employees as shown below.

Here’s the solution. We use the zipWhen method that waits for the result from source Mono, and then calls the second Mono. Because we can zip only the same stream types (in that case these are Mono) we need to convert Flux<Employee> returned by getEmployeesByOrganization method into Mono<MutableList<Employee>> using collectList function. Thanks to zipWhen we can then combine two Mono streams and create new objects inside map function.

@Test
fun testScenario1() {
   val organization : Mono<Organization> = getOrganizationByName("test")
      .zipWhen { organization ->
         getEmployeesByOrganization(organization.id!!).collectList()
      }
      .map { tuple -> 
         Organization(tuple.t1.id, tuple.t1.name, tuple.t2)
      }
}

Scenario 2

Let’s consider another scenario. Now, we have Flux streams that emit employees and departments. Every employee has property departmentId responsible for assignment to the department.

private fun getDepartments() : Flux<Department> {
    return Flux.just(Department(1, "X", 1),
                     Department(2, "Y", 1))
}

private fun getEmployees() : Flux<Employee> {
    return Flux.just(Employee(1, "Employee1", 1000, 1, 1),
            Employee(2, "Employee2", 2000, 1, 1),
            Employee(3, "Employee3", 1000, 1, 2),
            Employee(4, "Employee4", 2000, 1, 2))
}

The goal is to merge those two streams and return the single Flux stream emitting departments that contains all employees assigned to the given department. Here’s the picture that illustrates the transformation described above.

We can do that in two ways as shown below. First calls flatMap function on stream with departments. Inside flatMap we zip every single Department with a stream of employees. That stream is then filtered by departmentId and converted into Mono type. Finally, we are creating a Mono type using map function that emits a department containing a list of employees.
The second way groups Flux with employees by departmentId. Then it invokes zipping and mapping functions similar to the previous way.

@Test
fun testScenario2() {
   val departments: Flux<Department> = getDepartments()
      .flatMap { department ->
         Mono.just(department)
            .zipWith(getEmployees().filter { it.departmentId == department.id }.collectList())
            .map { t -> t.t1.addEmployees(t.t2) }
      }

   val departments2: Flux<Department> = getEmployees()
      .groupBy { it.departmentId }
      .flatMap { t -> getDepartments().filter { it.id == t.key() }.elementAt(0)
         .zipWith(t.collectList())
         .map { it.t1.addEmployees(it.t2) }
      }
}

Scenario 3

This scenario is simpler than two previous scenarios. We have two API methods that emit Flux with the same object types. First of them contains list of employees having id, name, salary properties, while the second id, organizationId, departmentId properties.

private fun getEmployeesBasic() : Flux<Employee> {
   return Flux.just(Employee(1, "AA", 1000),
                        Employee(2, "BB", 2000))
}

private fun getEmployeesRelationships() : Flux<Employee> {
   return Flux.just(Employee(1, 1, 1),
              Employee(2, 1, 2))
}

We want to convert it into a single stream emitting employees with a full set of properties. The following picture illustrates the described transformation.

In that case the solution is pretty simple. We are zipping two Flux streams using zipWith function, and then map two zipped objects into a single containing the full set of properties.

@Test
fun testScenario3() {
   val employees : Flux<Employee> = getEmployeesBasic()
      .zipWith(getEmployeesRelationships())
      .map { t -> Employee(t.t1.id, t.t1.name, t.t1.salary, t.t2.organizationId!!, t.t2.departmentId!!) }
}

Scenario 4

In this scenario we have two independent Flux streams that emit the same type of objects – Employee.

private fun getEmployeesFirstPart() : Flux<Employee> {
   return Flux.just(Employee(1, "AA", 1000), Employee(3, "BB", 3000))
}

private fun getEmployeesSecondPart() : Flux<Employee> {
   return Flux.just(Employee(2, "CC", 2000), Employee(4, "DD", 4000))
}

We would like to merge those two streams into a single stream ordered by id. The following picture shows that transformation.

Here’s the solution. We use mergeOrderedWith function with a comparator that compares id. Then we can perform some transformations on every object, but it is only an option that shows the usage on map function.

@Test
fun testScenario4() {
   val persons: Flux<Employee> = getEmployeesFirstPart()
      .mergeOrderedWith(getEmployeesSecondPart(), Comparator { o1, o2 -> o1.id.compareTo(o2.id) })
      .map {
         Employee(it.id, it.name, it.salary, 1, 1)
      }
}

Scenario 5

And the last scenario in this article. We have a single input stream Mono with Organization that contains a list of departments. Each of department inside that list also contains the list of all employees assigned to the given department. Here’s our API method implementation.

private fun getDepartmentsByOrganization(id: Int) : Flux<Department> {
   val dep1 = Department(1, "A", id, mutableListOf(
         Employee(1, "Employee1", 1000, id, 1),
         Employee(2, "Employee2", 2000, id, 1)
      )
   )
   val dep2 = Department(2, "B", id, mutableListOf(
         Employee(3, "Employee3", 1000, id, 2),
         Employee(4, "Employee4", 2000, id, 2)
      )
   )
   return Flux.just(dep1, dep2)
}

The goal is to convert the stream to the same stream Flux with Department, but containing a list of all employees in the department. The following picture visualizes the described transformation.

Here’s the solution. We invoke flatMapIterable function that converts Flux with Department> into Flux with Employees by returning List of Employee. Then we convert it to Mono and add to the newly created Organization object inside map function.

@Test
fun testScenario5() {
   var organization: Mono<Organization> = getDepartmentsByOrganization(1)
      .flatMapIterable { department -> department.employees }
      .collectList()
      .map { t -> Organization(1, "X", t) }
}

The post Reactive programming with Project Reactor appeared first on Piotr's TechBlog.

1. Setting up an environment

We can use Gradle or Maven for setting up our build environment. The first goal is to add some dedicated Maven repositories, because Ktor dependencies are not available in central repo.

<repositories>
    <repository>
        <id>ktor</id>
        <url>http://dl.bintray.com/kotlin/ktor</url>
    </repository>
    <repository>
        <id>kotlinx</id>
        <url>http://dl.bintray.com/kotlin/kotlinx</url>
    </repository>
    <repository>
        <id>jcenter</id>
        <url>http://jcenter.bintray.com</url>
    </repository>
</repositories>

The current version of Ktor framework is 0.9.5.

<properties>
    <ktor.version>0.9.5</ktor.version>
</properties>

Before adding any dependencies we should also configure kotlin-maven-plugin, and the compiler to avoid warnings when using Kotlin coroutines. They are still an experimental feature in Kotlin.

<plugin>
   <groupId>org.jetbrains.kotlin</groupId>
   <artifactId>kotlin-maven-plugin</artifactId>
   <version>${kotlin.version}</version>
   <executions>
      <execution>
         <id>compile</id>
         <phase>compile</phase>
         <goals>
            <goal>compile</goal>
         </goals>
      </execution>
      <execution>
         <id>test-compile</id>
         <phase>test-compile</phase>
         <goals>
            <goal>test-compile</goal>
         </goals>
      </execution>
   </executions>
   <configuration>
      <jvmTarget>1.8</jvmTarget>
      <args>
         <arg>-Xcoroutines=enable</arg>
      </args>
   </configuration>
</plugin>

2. Sample applications

We will create two microservices built on top of Ktor Framework: account-service and customer-service. The application customer-service calls endpoint exposed by account-service for searching all accounts of a given customer. Both microservices expose some other endpoints for adding or finding objects. There is also a third element in our architecture: discovery server. We will use HashiCorp’s Consul for that. Every instance of account-service would register itself in the discovery server, while customer-service would fetch the list of register instances to obtain their addresses and ports. In summary, that is a typical example of communication between microservices presented in some of my previous articles on this blog. The sample system architecture is visualized on the following diagram.

3. The Basics

First, we will add some dependencies required for running Ktor on the server side. Our application will be launched on Netty server, and will use Jackson library for JSON serialization.

<dependency>
   <groupId>io.ktor</groupId>
   <artifactId>ktor-server-core</artifactId>
   <version>${ktor.version}</version>
</dependency>
<dependency>
   <groupId>io.ktor</groupId>
   <artifactId>ktor-server-netty</artifactId>
   <version>${ktor.version}</version>
</dependency>
<dependency>
   <groupId>io.ktor</groupId>
   <artifactId>ktor-jackson</artifactId>
   <version>${ktor.version}</version>
</dependency>

After that we can create the main method that initializes the embedded server based on Netty. Because we would like to set the server listen port dynamically during application startup, we use a method that takes command line arguments as a parameter. Thanks to that we will use parameter -port=TARGET_PORT, when starting every single instance of application.

fun main(args: Array<String>) {
   val server = embeddedServer(Netty, commandLineEnvironment(args))
   server.start(wait = true)
}

It is possible to implement different configuration strategies with Ktor, but the recommended way is through a configuration file, called a HOCON file. Here’s the application.conf file for customer-service.

ktor {
  deployment {
    port: 8095
  }
  application {
    modules = [ pl.piomin.services.CustomerApplicationModuleKt.main ]
  }
}

Each application takes a list of required Ktor modules as a parameter. In fact, modules are the vital part of your application. These are user-defined functions receiving the Application class that is in charge of configuring the server pipeline, install features, registering routes, handling requests, etc. For the configuration visible above, the method main that defines our module is available inside file CustomerApplicationModule.kt located inside package pl.piomin.services.
Inside that method you can find the declarations of features provided by the Ktor framework and used by the application. This is the most important part of our code, because it contains almost all the logic implemented by the application. The most commonly used feature is Routing. It defines all the HTTP API endpoints exposed by our application. In the following fragment of code I have defined four endpoints: a single POST method for adding a new account, and three GET endpoints providing different find methods.

package pl.piomin.services
fun Application.main() {
   val repository = AccountRepository()
   //...
   routing {
      get("/accounts") {
         call.respond(message = repository.accounts)
      }
      get("/accounts/{id}") {
         val id: String? = call.parameters["id"]
         if (id != null)
            call.respond(message = repository.accounts.filter { it.id == id.toInt() })
      }
      get("/accounts/customer/{customerId}") {
         val customerId: String? = call.parameters["customerId"]
         if (customerId != null)
            call.respond(message = repository.accounts.filter { it.customerId == customerId.toInt() })
      }
      post("/accounts") {
         var account: Account = call.receive()
         account.id = repository.accounts.size + 1
         repository.addAccount(account)
         log.info("$account")
         call.respond(message = account)
      }
   }
}

4. Using built-in features

A Ktor application typically consists of a series of features. You can think of features as functionality that is injected into the request and response pipeline. Usually, an application would have a series of features such as DefaultHeaders which add headers to every outgoing response or Routing which allows us to define routes to handle requests, etc. We can create our own custom features, but there is also a set of built-in features implemented as Ktor modules. We can install any of the built-in features just by passing the class name inside the install statement. Here’s the list of features I used in the sample applications:

• ContentNegotation – it provides automatic content conversion according to Content-Type and Accept headers. I have used Jackson library for converting between objects and JSON content
• Metrics – it provides implementation for generating metrics with useful information about the server and the requests. We can use different exporters by selecting from modules provided by Dropwizard Metrics library. I decided to use Slf4jReporter that prints metric values to the log destination. But you could also use, for example InfluxReporter for sending metrics to InfluxDB
• CallLogging – it is used for logging the incoming client requests. It leverages the ApplicationEnvironment.log that uses slf4j, so you can easily configure the output
• CallId – this feature allows identifying a request/call by generating request id and can work along the CallLogging feature. The CallLogging feature add generated value to MDC by calling method callIdMdc with MDC field name as a parameter

install(ContentNegotiation) {
   jackson {
   }
}
install(Metrics) {
   Slf4jReporter.forRegistry(registry).outputTo(log).build().start(10, TimeUnit.SECONDS)
}
install(CallLogging) {
   level = Level.TRACE
   callIdMdc("X-Request-ID")
}
install(CallId) {
   generate(10)
}

To be able to work with the Metrics feature we first need to include the following dependency to pom.xml file.

<dependency>
   <groupId>io.ktor</groupId>
   <artifactId>ktor-metrics</artifactId>
   <version>${ktor.version}</version>
</dependency>

5. Configure logging

The logging configuration for the Ktor framework is based on SLF4J. You just need to include dependency of the logging provider to pom.xml. It can be Logback.

<dependency>
    <groupId>ch.qos.logback</groupId>
    <artifactId>logback-classic</artifactId>
    <version>1.2.3</version>
</dependency>

Then you need to create logback.xml configuration file, and place it in src/main/resources directory. The following configuration forces SLF4J to print logs only to the console. It includes the MDC field X-Request-ID generated using CallId feature.

<configuration>
    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
        <encoder>
            <pattern>%d{YYYY-MM-dd HH:mm:ss.SSS} [%thread] [%X{X-Request-ID}] %-5level %logger{36} - %msg%n</pattern>
        </encoder>
    </appender>

    <root level="trace">
        <appender-ref ref="STDOUT"/>
    </root>

    <logger name="org.eclipse.jetty" level="INFO"/>
    <logger name="io.netty" level="INFO"/>
</configuration>

Now, you are able to use a logger in your application just by calling io.ktor.application.log instance.

import io.ktor.application.log
// ...
post("/accounts") {
   var account: Account = call.receive()
   account.id = repository.accounts.size + 1
   repository.addAccount(account)
   log.info("$account")
   call.respond(message = account)
}

6. Running Consul server

The implementation of our applications is almost finished. We just need to add the communication between two sample microservices. To achieve it we first need to run a discovery server. Running Consul on the local machine is pretty easily using a Docker container. Here’s the command that runs Consul in standalone mode, and exposes API on port 8500.

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

Consul provides a web interface, which is available under address http://192.168.99.100:8500.

7. Custom Ktor feature for service discovery with Consul

Ktor Framework does not provide any components that implement typical microservice patterns like service discovery or distributed configuration. This library is in the early stage of development (still the current version is below 1.0), so such features will probably be implemented in the future. However, with Ktor we may easily implement a custom feature that is able to communicate with Consul. There are two types of features available for applications that use Ktor: server-side and client-side features. In that case we will implement a feature on the client side. It is responsible for intercepting an event of sending a message by HTTP client, and including communication with Consul to the pipeline.
Before starting any implementation we need to include a set of Ktor client libraries and also a library providing methods for calling Consul API.

<dependency>
   <groupId>io.ktor</groupId>
   <artifactId>ktor-client-apache</artifactId>
   <version>${ktor.version}</version>
</dependency>
<dependency>
   <groupId>io.ktor</groupId>
   <artifactId>ktor-client-json</artifactId>
   <version>${ktor.version}</version>
</dependency>
<dependency>
   <groupId>io.ktor</groupId>
   <artifactId>ktor-client-jackson</artifactId>
   <version>${ktor.version}</version>
</dependency>
<dependency>
   <groupId>com.orbitz.consul</groupId>
   <artifactId>consul-client</artifactId>
   <version>1.2.3</version>
</dependency>

Let’s take a look on the implementation of ConsulFeature. It uses Consul client provided by consul-client. The default Consul address is http://localhost:8500, but it can be overridden on the calling side. I would not like to get into the implementation details of the Ktor client feature. The most important thing is the code inside install method. It is executed during Render phase, which is fired before Send phase. Our custom feature replaces an address provided as a hostname in the client’s URL by the address and port taken from Consul.

class ConsulFeature(var consulUrl: String) {

    class Config {
        var consulUrl: String = "http://localhost:8500"
        fun build(): ConsulFeature = ConsulFeature(consulUrl)
    }

    companion object Feature : HttpClientFeature<Config, ConsulFeature> {
        var currentNodeIndex: Int = 0

        override val key = AttributeKey<ConsulFeature>("ConsulFeature")

        override fun prepare(block: Config.() -> Unit): ConsulFeature = Config().apply(block).build()

        override fun install(feature: ConsulFeature, scope: HttpClient) {
            scope.requestPipeline.intercept(HttpRequestPipeline.Render) {
                var consulClient = Consul.builder().withUrl(feature.consulUrl).build()
                val nodes = consulClient.healthClient().getHealthyServiceInstances(context.url.host).response
                val selectedNode = nodes[currentNodeIndex]
                context.url.host = selectedNode.service.address
                context.url.port = selectedNode.service.port
                currentNodeIndex = (currentNodeIndex + 1) % nodes.size
                println("Calling ${selectedNode.service.id}: ${context.url.buildString()}")
            }
        }
    }
}

How it works? Everything should be clear for you after looking at that fragment of code containing HTTP client declaration, and usage of that client inside customer-service endpoint implementation. Instead of setting an IP address in the client’s URL I used the name of the calling service – in that case account-service. This name is then replaced by ConsulFeature with an address and port taken for Consul server. If there is more than one instance of account-service registered in Consul server, the feature performs load balancing using the typical Round Robin method.

val client = HttpClient(Apache) {
   install(ConsulFeature) {
      consulUrl = "http://192.168.99.100:8500"
   }
   install(JsonFeature)
}
// ...
routing {
   get("/customers/{id}") {
      val id: String? = call.parameters["id"]
      if (id != null) {
         val accounts = client.get<Accounts>("http://account-service/accounts/customer/$id")
         val customerRet = customer.copy(id = customer.id, name = customer.name)
         customerRet.accounts.addAll(accounts)
         call.respond(message = customerRet)
      }
   }
}

And the last thing. The application needs to register itself in Consul after startup. Here’s the main function of account-service. It uses the register method of Consul client. An unique id is automatically generated using application listen port number.

fun main(args: Array<String>) {
    val server = embeddedServer(Netty, commandLineEnvironment(args))
    val consulClient = Consul.builder().withUrl("http://192.168.99.100:8500").build()
    val service = ImmutableRegistration.builder()
            .id("account-${server.environment.connectors[0].port}")
            .name("account-service")
            .address("localhost")
            .port(server.environment.connectors[0].port)
            .build()
    consulClient.agentClient().register(service)

    server.start(wait = true)
}

8. How it works?

Ok, let’s run two instances of account-service and a single instance of customer-service. When running two instances of account-service we need to override default port number by declaring an application running parameter -port=PORT_NUMBER.

Every instance of microservice should be registered in Consul after startup.

We are running instances of account-service. Here are the details.

Now, let’s add some test data by calling POST endpoints exposed by our microservices. As you can see I added new accounts only on the first instance of account-service. Because all the microservices store data in-memory, those objects will be stored only by instance running on port 8090.

$ curl -d '{"name":"John Smith"}' -H "Content-Type: application/json" -X POST http://localhost:8095/customers
$ curl -d '{"number":"1234567890", "balance":5000, "customerId":1}' -H "Content-Type: application/json" -X POST http://localhost:8090/accounts
$ curl -d '{"number":"1234567891", "balance":10000, "customerId":1}' -H "Content-Type: application/json" -X POST http://localhost:8090/accounts

If you call endpoint http://localhost:8095/customers/1 it tries to connect with running instances of account-service. Once it returns response {"id":1,"name":"John Smith","accounts":[{"id":1,"balance":5000,"number":"1234567890","customerId":1},{"id":2,"balance":10000,"number":"1234567891","customerId":1}]}, while the second time {"id":1,"name":"John Smith","accounts":[]}. That is the expected result. Because ConsulFeature load balances between two instances of account-service, when only the first instance stores data. Here’s the fragment of application logs.

Conclusion

In this article I presented how to use the Ktor framework for building microservices architecture with Consul server. We have used some basic Ktor features like mechanisms for routing, logging, metrics, and some more advanced solutions for building our own feature that interacts with Consul during inter-service communication. Ktor seems to be a very interesting framework. I will definitely follow the progress in development of this framework.

The post Kotlin Microservices with Ktor appeared first on Piotr's TechBlog.