Apache Cassandra 2019 and WebFlux and WebMVC

Apache Cassandra 2019 and WebFlux and WebMVC

Create the Table in Cassandra
> CREATE KEYSPACE jobsmonitor WITH REPLICATION = { 'class' : 'SimpleStrategy', 'replication_factor' : 2 };

> use jobsmonitor;

> CREATE TABLE jobcounthistory(
               ... source_id text,
               ... record_date TIMESTAMP,
               ... new_active_count int,
               ... new_admin_count int,
               ... old_active_count int,
               ... old_admin_count int,
               ... PRIMARY KEY ( source_id, record_date )
               ... ) WITH CLUSTERING ORDER BY (record_date DESC);

The query will be
> select * from jobcounthistory where source_id = 'id1';
> select * from jobcounthistory where source_id = 'id1' and record_date > '2019-06-01';

Create the first example of WebMVC
Add pom.xml
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-cassandra</artifactId>
</dependency>

The POJO Domain Class
package com.sillycat.mvclatency.domain;

import java.util.Date;

import org.springframework.data.cassandra.core.cql.Ordering;
import org.springframework.data.cassandra.core.cql.PrimaryKeyType;
import org.springframework.data.cassandra.core.mapping.Column;
import org.springframework.data.cassandra.core.mapping.PrimaryKeyColumn;
import org.springframework.data.cassandra.core.mapping.Table;

import lombok.Getter;
import lombok.Setter;
import lombok.ToString;

@Setter
@Getter
@ToString
@Table("jobcounthistory")
public class JobCountHistory {

    @PrimaryKeyColumn(name = "source_id", ordinal = 0, type = PrimaryKeyType.PARTITIONED)
    private String sourceID;

    @PrimaryKeyColumn(name = "record_date", ordinal = 1, type = PrimaryKeyType.CLUSTERED, ordering = Ordering.DESCENDING)
    private Date recordDate;

    @Column("new_active_count")
    private Integer newActiveCount;

    @Column("new_admin_count")
    private Integer newAdminCount;

    @Column("old_active_count")
    private Integer oldActiveCount;

    @Column("old_admin_count")
    private Integer oldAdminCount;

}

The Repository Interface Class
package com.sillycat.mvclatency.repository;

import java.util.Date;
import java.util.List;

import org.springframework.data.cassandra.repository.Query;
import org.springframework.data.repository.CrudRepository;
import org.springframework.stereotype.Repository;

import com.sillycat.mvclatency.domain.JobCountHistory;

@Repository
public interface JobCountHistoryRepositoryCassandra extends CrudRepository<JobCountHistory, String> {

    @Query("select * from jobcounthistory where source_id=?0")
    public List<JobCountHistory> findBySourceID(String sourceID);

    @Query("select * from jobcounthistory where source_id=?0 and record_date>?1")
    public List<JobCountHistory> findBySourceIDNearTime(String sourceID, Date recordDate);

    @Query("delete from jobcounthistory where source_id=?0")
    public void deleteBySourceID(String sourceID);

}

Configuration to add the Cassandra connection application.xml
spring:
data:
  cassandra:
   keyspace-name: jobsmonitor
   contact-points: ubuntu-master, ubuntu-dev2

Unit Class
package com.sillycat.mvclatency.repository;

import java.util.Date;
import java.util.List;
import java.util.Random;

import org.junit.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import com.sillycat.mvclatency.domain.JobCountHistory;

import lombok.extern.slf4j.Slf4j;

@Slf4j
@RunWith(SpringRunner.class)
@SpringBootTest
public class JobCountHistoryCassandraTest {

    @Autowired
    JobCountHistoryRepositoryCassandra jobCountHistoryRepositoryCassandra;

    @Test
    public void testBeanNotNull() {
        log.info("init the bean jobCountHistoryRepositoryCassandra");
        Assert.assertNotNull(jobCountHistoryRepositoryCassandra);
    }

    @Test
    public void save() throws Exception {
        JobCountHistory item1 = new JobCountHistory();
        item1.setNewActiveCount(new Random().nextInt(10));
        item1.setNewAdminCount(new Random().nextInt(100));
        item1.setOldActiveCount(new Random().nextInt(10));
        item1.setOldAdminCount(new Random().nextInt(100));
        item1.setSourceID("9527");
        item1.setRecordDate(new Date());
        jobCountHistoryRepositoryCassandra.save(item1);

        List<JobCountHistory> result = jobCountHistoryRepositoryCassandra.findBySourceID("9527");
        Assert.assertNotNull(result);
    }

}

Add the Controller to test the performance
package com.sillycat.mvclatency.web;

import java.util.List;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;

import com.sillycat.mvclatency.domain.JobCountHistory;
import com.sillycat.mvclatency.repository.JobCountHistoryRepositoryCassandra;

import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;

@Api(value = "/jobcounthistory/")
@RestController
public class JobCountHistoryController {

    @ApiOperation(value = "fetch job by sourceID", response = String.class)
    @GetMapping("/jobcounthistory/{sourceID}")
    public List<JobCountHistory> findBySourceID(@PathVariable String sourceID) {
        List<JobCountHistory> results = jobCountHistoryRepositoryCassandra.findBySourceID(sourceID);
        return results;
    }

    @Autowired
    JobCountHistoryRepositoryCassandra jobCountHistoryRepositoryCassandra;

}


Install WRK if needed on Ubuntu
> sudo apt-get install build-essential libssl-dev git -y
> git clone https://github.com/wg/wrk.git wrk
> cd wrk/
> make
> sudo cp wrk /usr/local/bin

Check the Performance Differences
> wrk -t4 -c1000 -d60s --timeout 10s http://ubuntu-master:8081/jobcounthistory/9527 --latency

The strange thing is that I get better performance in WebMVC, close but worse performance in WebFlux.

Here are some changes in WebFlux
Package dependencies pom.xml
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-cassandra</artifactId>
</dependency>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-test</artifactId>
<scope>test</scope>
</dependency>

The JobCountHistory POJO are the same as previous project.

The repository interface class is different
package com.sillycat.webfluxlatency.repository;

import java.util.Date;

import org.springframework.data.cassandra.repository.Query;
import org.springframework.data.cassandra.repository.ReactiveCassandraRepository;
import org.springframework.stereotype.Repository;

import com.sillycat.webfluxlatency.domain.JobCountHistory;

import reactor.core.publisher.Flux;

@Repository
public interface JobCountHistoryRepositoryCassandra extends ReactiveCassandraRepository<JobCountHistory, String>{

    @Query("select * from jobcounthistory where source_id=?0")
    public Flux<JobCountHistory> findBySourceID(String sourceID);
   
    @Query("select * from jobcounthistory where source_id=?0 and record_date>=?1")
    public Flux<JobCountHistory> findBySourceIDNearTime(String sourceID, Date recordDate);
   
}

The WebFlux Controller is as follow:
package com.sillycat.webfluxlatency.web;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;

import com.sillycat.webfluxlatency.domain.JobCountHistory;
import com.sillycat.webfluxlatency.repository.JobCountHistoryRepositoryCassandra;

import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import reactor.core.publisher.Flux;

@Api(value = "/jobcounthistory/")
@RestController
public class JobCountHistoryController {

    @ApiOperation(value = "fetch job by sourceID", response = JobCountHistory.class)
    @GetMapping("/jobcounthistory/{sourceID}")
    public Flux<JobCountHistory> findBySourceID(@PathVariable String sourceID) {
        Flux<JobCountHistory> results = jobCountHistoryRepositoryCassandra.findBySourceID(sourceID);
        return results;
    }

    @Autowired
    JobCountHistoryRepositoryCassandra jobCountHistoryRepositoryCassandra;

}

The configuration are similar, but the unit test is different.
package com.sillycat.webfluxlatency.repository;

import java.util.Date;
import java.util.Random;

import org.junit.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import com.sillycat.webfluxlatency.domain.JobCountHistory;

import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;

@Slf4j
@RunWith(SpringRunner.class)
@SpringBootTest
public class JobCountHistoryCassandraTest {

    @Autowired
    JobCountHistoryRepositoryCassandra jobCountHistoryRepositoryCassandra;

    @Test
    public void testBeanNotNull() {
        log.info("init the bean jobCountHistoryRepositoryCassandra");
        Assert.assertNotNull(jobCountHistoryRepositoryCassandra);
    }

    @Test
    public void save() throws Exception {
        JobCountHistory item1 = new JobCountHistory();
        Date now = new Date();
        item1.setNewActiveCount(new Random().nextInt(10));
        item1.setNewAdminCount(new Random().nextInt(100));
        item1.setOldActiveCount(new Random().nextInt(10));
        item1.setOldAdminCount(new Random().nextInt(100));
        item1.setSourceID("9527");
        item1.setRecordDate(now);
        item1.setRecordDate(new Date());
        Mono<JobCountHistory> result1 = jobCountHistoryRepositoryCassandra.save(item1);
        StepVerifier.create(result1).expectNext(item1).verifyComplete();

        Flux<JobCountHistory> result2 = jobCountHistoryRepositoryCassandra.findBySourceIDNearTime("9527", now);
        StepVerifier.create(result2).expectNextCount(1l).verifyComplete();

    }

}

I used wrk to compare the performance, it seems that in this situation WebMVC is faster than WebFlux. I need more testing on that to figure out.

References:
https://www.baeldung.com/spring-data-cassandra-reactive
https://github.com/eugenp/tutorials/tree/master/persistence-modules/spring-data-cassandra
https://github.com/eugenp/tutorials/tree/master/persistence-modules/spring-data-cassandra-reactive

WebMVC with Cassandra
https://www.baeldung.com/spring-data-cassandra-tutorial
https://www.baeldung.com/spring-data-cassandratemplate-cqltemplate