DEV Community: Ishan Soni

Postgres Partitioning and Spring Data JPA

Ishan Soni — Wed, 20 Mar 2024 13:52:12 +0000

But first — Inheritance

Inheritance in Postgres allows you to create a table that inherits from another table. This feature can be used to model hierarchical data within the database. A child table inherits all columns from the parent table but can also have additional columns.

Example: Build a data model for cities and population data. Each country has many cities but only 1 capital. A common access pattern that your model should be able to support is — Get all capitals and their population data!

One way to solve this problem:

A common approach one could take is the following (add an is_capital column to your schema).



create table cities (
 name varchar(100),
 country_iso_code varchar(10),
 population bigserial,
 is_capital boolean,
 primary key (name)
);

insert into cities (name, country_iso_code, population, is_capital) values
('New Delhi', 'IN', 10927986, true),
('Pune', 'IN', 3124458, false),
('Bangkok', 'TH', 5104476, true),
('Almaty', 'KZ', 1977011, false),
('Stockholm', 'SE', 1515017, true),
('Washington, D.C.', 'US', 689545, true);

select * from cities where is_capital = true;

Another way to solve this problem (using inheritance)



create table cities (
 name varchar(100),
 country_iso_code varchar(10),
 population bigserial,
 primary key (name)
);

create table capitals (
 -- You can define additional columns here if you want to
) inherits (cities);

insert into cities (name, country_iso_code, population) values
('Pune', 'IN', 3124458),
('Almaty', 'KZ', 1977011);


insert into capitals (name, country_iso_code, population) values
('New Delhi', 'IN', 10927986),
('Bangkok', 'TH', 5104476),
('Stockholm', 'SE', 1515017),
('Washington, D.C.', 'US', 689545);

A select * from cities returns data from all tables (base + inherited).

The explain command shows that it does a scan on both the tables.

A select * from capitals returns data in the capitals table only.

The explain command shows that it does a scan on the capitals table only, potentially improving query performance.

Partitioning

Partitioning in postgres is based on/backed by inheritance under the hood. Partitioning in PostgreSQL involves splitting a large table into smaller, more manageable tables, while still treating them as a single table. This can significantly improve performance for large datasets by reducing index size, enhancing query performance, and making bulk operations like loads and deletes more efficient. This can also become the base for your archival process.

The idea is to partition/split a table by a partition key and while querying, supply the partition key in the where clause such that postgres only queries the required partition thereby improving query performance.

Example: We have a portfolio management application. It stores EOD stock prices for a large number of securities. Over time, this table will become bulky and this can deteriorate performance. Also, how do you create an archiving strategy to move/delete old data from this table?



create table security_prices (
 id varchar(255),
 security_id varchar(255),
 price numeric,
 business_date date
) partition by range (business_date);

-- Create partitioned tables. In postgres, this has to be done manually :(
-- Range (Inclusive, Exclusive)
create table security_prices_march_2024 partition of security_prices for values from ('2024-03-01') to ('2024-04-01');
create table security_prices_april_2024 partition of security_prices for values from ('2024-04-01') to ('2024-05-01');
create table security_prices_may_2024 partition of security_prices for values from ('2024-05-01') to ('2024-06-01');

insert into security_prices (id, security_id, price, business_date) values 
('1', 'IBM', 191.42, '2024-03-19'),
('2', 'MSFT', 419.27, '2024-03-19'),
('3', 'AAPL', 174.85, '2024-03-19'),
('4', 'AMZN', 175.29, '2024-03-19'),
('5', 'NVDA', 1000, '2024-04-01'),
('6', 'NVDA', 1050, '2024-05-01'),
('7', 'AMD', 200, '2024-04-15'),
('8', 'AMD', 215, '2024-05-15');

A select * from security_prices returns data from all partitioned tables.

The explain command shows that it does a scan on all the tables.

But, what if you mention the partition key in the where clause?

A select * from security_prices where business_date = ‘2024–03–19’ returns the correct data:

And, the explain command shows that it queried the march table only!

The example above is an example of Range Partitioning — Partitioning your tables by ranges.

Now that you have separate tables, you could drop or move data safely from old tables (Archiving!)

Note — In the above example, I have not defined a primary key. In PostgreSQL, if you do not explicitly define a PK for a table, then the table will not have a PK (lookup ctid in postgres to get more information on this). If you choose to add a PK to a partitioned table (which I suggest you should), the PK must include the partition key. In our case, since business_date is the partition key, if you were to add a PK, it would need to include business_date along with other columns to ensure uniqueness across all partitions. For instance, you could define a composite primary key that includes both id and business_date: CREATE TABLE security_prices (…, PRIMARY KEY (id, business_date);. In this case, you can create a composite key (which will include both Id and BusinessDate) using @ Embeddable and embed it in your entity using @ EmbeddedId (see — Composite keys and JPA)

Now, let’s map this entity to our spring data jpa application:



spring.application.name=security-service

spring.datasource.url=jdbc:postgresql://localhost:5432/security-service
spring.datasource.username=postgres
spring.datasource.password=password
spring.jpa.hibernate.ddl-auto=validate



@Entity(name = "security_prices")
@Data
@EqualsAndHashCode(of = {"id"})
public class SecurityPrice {

    @Id
    /*
    IMP - How do you map an entity with no primary key in spring data jpa? 
    JPA expects every entity to have an @Id column!
    You cannot use the internal ctid postgres column here.
    But, you can use any business key as an Id in JPA even though it's not a primary key in the database
     */
    private String id = UUID.randomUUID().toString();
    private String securityId;
    private BigDecimal price;
    private LocalDate businessDate;

}



@Repository
public interface SecurityPriceJpaRepository extends JpaRepository<SecurityPrice, String> {
    Optional<SecurityPrice> findBySecurityIdAndBusinessDate(String securityId, LocalDate businessDate);
}



@Service
public class SecurityPriceService {

    private final SecurityPriceJpaRepository securityPriceJpaRepository;

    public SecurityPriceService(SecurityPriceJpaRepository securityPriceJpaRepository) {
        this.securityPriceJpaRepository = securityPriceJpaRepository;
    }

    @Transactional
    public String createSecurityPrice(SecurityPriceRequest securityPriceRequest) {
        String securityId = securityPriceRequest.securityId();
        LocalDate businessDate = securityPriceRequest.businessDate();

        Optional<SecurityPrice> existingPriceOptional =
                securityPriceJpaRepository.findBySecurityIdAndBusinessDate(securityId, businessDate);

        SecurityPrice securityPrice = new SecurityPrice();
        securityPrice.setSecurityId(securityId);
        securityPrice.setPrice(securityPriceRequest.price());
        securityPrice.setBusinessDate(businessDate);

        existingPriceOptional.ifPresent(price -> securityPrice.setId(price.getId()));

        securityPriceJpaRepository.save(securityPrice);
        return securityPrice.getId();
    }

}



public record SecurityPriceRequest(String securityId, BigDecimal price, LocalDate businessDate) { }



@RestController
@RequestMapping("/v1/securities")
public class SecurityPriceController {

    private final SecurityPriceService securityPriceService;

    public SecurityPriceController(SecurityPriceService securityPriceService) {
        this.securityPriceService = securityPriceService;
    }

    @PostMapping("/prices")
    public ResponseEntity<String> createPriceEntry(@RequestBody SecurityPriceRequest securityPriceRequest) {
        return ResponseEntity.ok(securityPriceService.createSecurityPrice(securityPriceRequest));
    }

}

Running the explain select * from security_prices with a business date of 2024–03–19 shows that it queries the march table only!

List Partitioning

Another type of partitioning that is commonly used is List partitioning.

Example: You have an ecommerce application that generates a lot of orders. An order transitions through the following states — (CREATED, SHIPPED, DELIVERED, CANCELLED, ARCHIVED). A cron runs every night that marks DELIVERED (delivered > a month ago) and CANCELLED orders as ARCHIVED. Archived orders are only needed for audit/reporting purposes. Keeping these orders in the orders table will impact the performance of our application.

Let’s come up with a partitioning scheme for this problem:



create table orders (
 order_id varchar(255),
 user_id varchar(255),
 total numeric,
 order_items jsonb,
 status varchar(255)
) partition by list (status);

create table active_orders partition of orders for values in ('CREATED', 'SHIPPED', 'DELIVERED', 'CANCELLED');

create table archived_orders partition of orders for values in ('ARCHIVED');

Let’s add a new order



insert into orders values ('1', '1', 100, null, 'CREATED');
select * from orders;



explain select * from orders;



explain select * from orders where status in ('CREATED', 'SHIPPED', 'DELIVERED', 'CANCELLED');



update orders set status = 'ARCHIVED' where order_id = '1';

The order will automatically move into the archived_orders table. In your application, always specify the status in the where clause. This way, your application will almost always access the active orders table which will have less numbers of orders at any given point of time boosting performance!

Introduction to Spring Cloud Function and Spring Cloud Stream

Ishan Soni — Thu, 08 Feb 2024 17:21:20 +0000

What is Streaming and Stream processing?

Event Stream Processing (ESP) is the practice of taking action on a series of data points/events (stream) that originate from a system that continuously creates data.

Stream processing can typically be represented using a DAG:

This is a very high level abstraction. Can we model it using something we already have in Java? Yes — Java8 Functions!

Spring Cloud Function

Spring Cloud Function is a framework that promotes the implementation of business logic as Java Functions further instrumenting them to execute in context of different runtimes (HTTP, Messaging, etc.)!

Example — Let’s create a function that turns a string into uppercase:

public Function<String, String> toUpper() {
    return (str) -> {
        System.out.println("Original: " + str);
        return str.toUpperCase();
    };
}

If you are using spring initializer, and simply add the Spring Cloud Function dependency, it will add the spring-cloud-function-context dependency. If you have the Spring Web dependency in your project and add the Spring Cloud Function dependency, it will add the spring-cloud-function-web dependency i.e. we will be able to expose the above function as a rest endpoint (i.e this function is automatically bound to an HTTP context). Let’s see it in action!

//Returns a Function (A Processor node -> has both Input and Output!)
@Bean
public Function<String, String> toUpper() {
    return (str) -> {
        System.out.println("Original: " + str);
        return str.toUpperCase();
    };
}

Start your application and invoke this function at:

localhost:/toUpper/ishan

Output:

With Spring Cloud Function, we were successfully able to bind this function to an HTTP context!

Spring Cloud Stream

A framework that instruments your Spring Cloud Functions as event handlers bound to destinations using binders specific to a particular messaging system. Also see — why do you need a messaging system?

Eg — You want your Spring Cloud Function to become a message handler for a RabbitMQ destination(queue)!

Setup

Add the spring-cloud-stream-binder-rabbit dependency.

Bring up RabbitMQ using docker-compose

---
version: “3.8”
services:
  rabbitmq:
    image: rabbitmq:3.11-management-alpine
    container_name: rabbitmq
    ports:
      - 5672:5672
      - 15672:15672
    environment:
      RABBITMQ_DEFAULT_USER: guest
      RABBITMQ_DEFAULT_PASS: guest
      RABBITMQ_DEFAULT_VHOST: /

Connect RabbitMQ to your spring application (application.properties/yaml)

rabbitmq.host=localhost
rabbitmq.port=5672
rabbitmq.username=guest
rabbitmq.password=guest

When you bring the application up, you’ll see two new exchanges!

If you drop a message in toUpper-in-0 exchange, your toUpper function will get invoked and the output will be pushed to toUpper-out-0 exchange! Note — The REST access to this function will still work if you also have the spring-cloud-function-web dependency!

Note — A Function is a processor. That is why it created 2 exchanges! It listens to toUpper-in-0 and sends the output to toUpper-out-0

Let’s create a consumer for toUpper-out-0 exchange:

@Bean
public Consumer<String> consumeUpper() {
    return (upper) -> {
        System.out.println("Consumed: " + upper);
    };
}

# Spring Cloud Functions defined in your application
# Define multiple functions definitions by using ;
spring.cloud.function.definition=toUpper;consumeUpper

# Based on the Function definitions, Spring Cloud Stream will create bindings
# which connects the Function's input and/or output to destinations! 
# The toUpper-in-0 (the toUpper function listens to this)
# and toUpper-out-0 (the toUpper function produces results to this) were 
# automatically created

# If you do not do any configuration for the consumeUpper function,
# it will automatically create a consumeUpper-in-0 exchange and listen 
# to that. But we want this function to listen to toUpper-out-0
# What we are saying below is ->
# Bind consumeUpper-in-0 binding to listen to the toUpper-out-0 destination
# Since this is a Consumer function, the binding that is created is consumeUpper-in-0
spring.cloud.stream.bindings.consumeUpper-in-0.destination=toUpper-out-0
spring.cloud.stream.bindings.consumeUpper-in-0.group=school-service

The concept of groups is similar to consumer groups in Kafka. But more on that in another article.

Let’s drop the message “Ishan” into the exchange “toUpper-in-0” using the RabbitMQ admin console. You’ll see the following in your application logs:

Producer

You can directly produce data to a destination using the StreamBridge class:

@Autowired
private StreamBridge streamBridge;

...

streamBridge.send(Destination, data);

or you can use a Supplier

Supplier is a little different that a Function or a Consumer. Functions/Consumers are triggered whenever there is an input. But a Supplier doesn’t have a trigger. In Spring Cloud Function, the Supplier has an automated polling mechanism which polls the supplier every second (which you can override)

@Bean
public Supplier<String> createData() {
    return () -> {
        System.out.println("Creating some data");
        return "Ishan-" +
                ThreadLocalRandom.current().nextInt(0, 10000) +
                "-" +
                LocalDateTime.now();
    };
}

spring.cloud.function.definition=toUpper;consumeUpper;createData

...

# Modify the poll interval of the supplier
spring.cloud.stream.poller.fixed-delay=10000

# The binding destination of the supplier function.
# If we had provided no configuration, spring cloud stream would have
# automatically created a createData-out-0 exchange and the data this
# supplier creates will be pushed to that exchange. But we want this
# supplier to push data to the toUpper-in-0 exchange
spring.cloud.stream.bindings.createData-out-0.destination=toUpper-in-0

Restart your application again:

Sending and Consuming data based on specific binding and routing keys

If you are unfamiliar with RabbitMQ routing and binding keys, see this article first.

Example — You are using StreamBridge to send messages to a students exchange. How do you set the routing key of the message?

spring.cloud.stream.rabbit.bindings.students.producer.routingKeyExpression=headers['type']

What you are saying is, treat the type header as a routing key

import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;

...

Message<StudentCreated> message = MessageBuilder.withPayload(studentCreated)
        .setHeader("type", "student.created").build();
streamBridge.send("students", message);

Now, how do you bind the below consumer and the queue it listens to a specific binding key on an exchange? (i.e this consumer only wants to listen to StudentCreated events from the students exchange)

@Bean
public Consumer<StudentCreated> studentCreated() {
  ...
}

spring.cloud.function.definition=studentCreated
spring.cloud.stream.bindings.studentCreated-in-0.destination=students
spring.cloud.stream.bindings.studentCreated-in-0.group=school-service
#IMP
spring.cloud.stream.rabbit.bindings.studentCreated-in-0.consumer.bindingRoutingKey=student.created
spring.cloud.stream.rabbit.bindings.studentCreated-in-0.consumer.bindingRoutingKeyDelimiter=,

What is RabbitMQ and why do you need a Message Broker?

Ishan Soni — Thu, 08 Feb 2024 14:31:32 +0000

Asynchronous communication and Message Brokers

In traditional distributed applications, communication was usually done in a synchronous fashion. Synchronous communication, though simple has its own challenges.

Let's take an example of an e-commerce application:

Let's assume our checkout service wants to interact with the inventory service to do a task. The following problems can occur if you are using synchronous communication:

There could be a network issue causing the RPC call to fail.
The inventory service could be down.
What if too many requests come in making the inventory service a bottleneck?

While you can create resiliency mechanisms for synchronous communication like retrying in case of network failures, introducing a circuit breaker to gracefully shut down functions in case the target service is down, spawning additional pods (horizontal scaling) and putting them behind a load balancer to distribute requests in case load on the system increases, there are still things that can go wrong.

This is where asynchronous communication and message queues/brokers come into the picture. They sit in between distributed components.

The checkout service will add an event/message to the message queue and move onto the next task. The inventory service, whenever it's ready (and at it’s own pace) will process these messages.

RabbitMQ

RabbitMQ is a message broker and implements the AMQP (Advanced Messaging Queuing Protocol).

Producers, instead of submitting messages directly to a queue, submit messages to an Exchange. The message can have a Routing Key.

Queues are connected to exchanges using queue Bindings.

Consumers listen to a queue.

An Exchange routes an incoming message to queue(s) using the routing key of the incoming message and the queue bindings. How messages move through the system depends on the type of exchange.

Types of Exchanges

Fanout Exchange

A fanout exchange routes messages to all queues that are bound to it and ignore the routing key.

Topic Exchange

Topic exchanges route messages to queues based on wildcard matches between the routing key and the queue binding.

The routing key must be a list of words, delimited by a period (.). Examples are agreements.us and agreements.eu.stockholm which in this case identifies agreements that are set up for a company with offices in lots of different locations. The routing patterns may contain an asterisk (“”) to match a word in a specific position of the routing key (e.g., a routing pattern of “agreements...b.” only match routing keys where the first word is “agreements” and the fourth word is “b”). A pound symbol (“#”) indicates a match of zero or more words (e.g., a routing pattern of “agreements.eu.berlin.#” matches any routing keys beginning with “agreements.eu.berlin”).

Direct Exchange

With the topic exchange, you can do a partial match between the routing key and queue bindings, while with a direct exchange you can only do a direct/full match.

Spring Data MongoDB — CRUD, Aggregations, Views and Materialized Views

Ishan Soni — Tue, 30 Jan 2024 17:11:00 +0000

Basic Setup

Setup a single node MongoDB instance using docker-compose.yaml and bring it up using the docker-compose up -d command



version: '3.9'

services:
  mongo:
    image: mongo
    container_name: mongodb
    ports:
      - 27017:27017
    volumes:
      - ./mongo:/data/db
    environment:
      - MONGO_INITDB_ROOT_USERNAME=mongo
      - MONGO_INITDB_ROOT_PASSWORD=root

I’d recommand using the MongoDB Compass application as the GUI tool.

Add the following MongoDB dependencies in your pom.xml or use spring initializer and select the Spring Data MongoDB dependency:



<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-mongodb</artifactId>
</dependency>

Add the required data source configuration details to your application.properties/yaml files:



spring.data.mongodb.authentication-database=admin #Pre-existing DB. For auth related stuff
spring.data.mongodb.username=mongo #See docker-compose.yml file
spring.data.mongodb.password=root  #See docker-compose.yml file
spring.data.mongodb.database=school-service #Name of your database
spring.data.mongodb.host=localhost
spring.data.mongodb.port=27017

Create your Entities and Repositories (Let’s create a school management project)

@ Document annotation simply marks a class as being a domain object that needs to be persisted to the database, along with allowing us to choose the name of the collection to be used.

@ Id marks this attribute as the primary key. If the value of the @ Id field is not null, it’s stored in the database as-is; otherwise, the converter will assume we want to store an ObjectId in the database (either ObjectId, String or BigInteger work).



@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
public class Address {
    //...Other fields
    private String city;
    private String state;
}



@Document(value = "institutes")
@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
public class Institute {
    @Id
    private String id;
    private String name;
    private Address address;
}



@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
public class User {
    private String firstName;
    private String lastName;
    private Gender gender; //Enum - MALE, FEMALE
    private LocalDate dob;
    //Incremented using a cron that runs every day!
    private int age;
    private String email;
    private String phone;
    private Address address;
}



@Data
@NoArgsConstructor
@AllArgsConstructor
public class Hobby {
    private String hobby;
    //0 - 10
    private short experience;
}



@Document("students")
@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
public class Student {
    @Id
    private String id;
    private User userDetails;
    private String instituteId;
    private List<Hobby> hobbies;
    //...Other fields
}



@Document(value = "teachers")
@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
public class Teacher {
    @Id
    private String id;
    private User userDetails;
    private String instituteId;
}



@Document(value = "classes")
@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
public class InstituteClass {
    @Id
    private String id;
    private String instituteId;
    private String classId;
    private String section;
    private short year;
    private List<String> studentIds;
    private String classTeacherId; //Homeroom Teacher
}



@Repository
public interface InstituteRepository extends MongoRepository<Institute, String> {
    //A normal spring data query method.
    Optional<List<Institute>> findByName(String name);
}
//+ Other repositories

Add @ EnableMongoRepositories to your application configuration. Spring will provide you with proxy repository implementations using which you can do almost all CRUD related stuff. Simply autowire these repositories and you are good to go:

User defined queries

You can create user defined queries (and specify prototype — the where clause and projection — the select clause)

Using the @Query annotation:



@Repository
public interface StudentRepository extends MongoRepository<Student, String> {

    // ?0 is a Positional parameter - will be replaced by id!
    // value -> Prototype and fields -> Projection!
    // Will only populate the id and hobbies fields in the student document!
    @Query(value = "{'id' : ?0}", fields = "{hobbies : 1}")
    Student findHobbiesById(String id);

    @Query(value = "{'userDetails.gender': ?0, 'userDetails.age': {'$lt': ?1}}")
    List<Student> findAllByGenderWithAgeLessThan(Gender gender, int age);

    @Query(value = "{'hobbies': {'$elemMatch': {'hobby': ?0, 'experience': {'$gt': ?1}}}}")
    List<Student> getStudentsWithHobbyExperienceGreaterThan(String hobby, int experience);

}

Using MongoTemplate:

Let’s try to re-create the above methods using MongoTemplate. You can simply auto-wire MongoTemplate in your repository classes and start using it:



import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.stereotype.Repository;
...

@Repository
@Slf4j
public class CustomStudentRepository {

    private final MongoTemplate mongoTemplate;

    public CustomStudentRepository(MongoTemplate mongoTemplate) {
        log.info("Initialized mongodb template " + mongoTemplate);
        this.mongoTemplate = mongoTemplate;
    }

    public List<Student> findAllByGenderWithAgeLessThan(Gender gender, int age) {
        Query query = new Query();
        //query.allowSecondaryReads();
        //query.withReadPreference(...
        //query.withReadConcern(...

        Criteria criteria = Criteria.where("userDetails.gender").is(gender.name()).and("userDetails.age").lt(10);

        query.addCriteria(criteria);

        //mongoTemplate.find(Query, Entity.class);
        return mongoTemplate.find(query, Student.class);
    }


    public List<Student> getStudentsWithHobbyExperienceGreaterThan(String hobby, int experience) {
        Query query = new Query();
        Criteria criteria = Criteria.where("hobbies").elemMatch(
                Criteria.where("hobby").is(hobby)
                        .and("experience").gt(experience)
        );
        query.addCriteria(criteria);

        return mongoTemplate.find(query, Student.class);
    }

}

Updates

Let’s assume you have a cron that runs at midnight and finds out all of the students whose birthday is today. We now need to increment their age by 1. Here is how you can do it using MongoTemplate:



//Todo - Instead, use in and pass-in a list of student ids!
public void incrementAgeBy1(String studentId) {
    Query query = new Query();
    query.addCriteria(Criteria.where("id").is(studentId));

    Update update = new Update();
    //update.set()
    //update.unset()
    //update.addToSet()
    //update.push()
    //and more...
    update.inc("userDetails.age", 1);

    //mongoTemplate.updateFirst(Query, UpdateDefinition, Entity.class);
    mongoTemplate.updateFirst(query, update, Student.class);
}

Aggregations

Let’s try to do the following aggregations — Find gender count for an institute and Find gender count for all institutes:

Using the @Aggregation annotation:



import org.springframework.data.mongodb.repository.Aggregation;
...

@Repository
public interface StudentRepository extends MongoRepository<Student, String> {
    //...Other methods

    record StudentsByGenderForInstitute(String gender, long total) {}

    @Aggregation(pipeline = {
            "{'$match': {'instituteId': ?0}}",
            "{'$group': {'_id': '$userDetails.gender', 'total': {'$count':{}}}}",
            "{'$project': {'_id': 0, 'gender': '$_id', 'total': 1}}"
    })
    List<StudentsByGenderForInstitute> getStudentsByGenderForInstitute(String instituteId);

    @Data
    class StudentsByGender {
        private String instituteId;
        private String gender;
        private long total;
    }

    @Aggregation(pipeline = {
            "{'$group': {'_id': {'instituteId: '$instituteId', 'gender': '$userDetails.gender'}, 'total': {'$count': {}}}}",
            "{'$project': {'_id': 0, 'instituteId': '$_id.instituteId', 'gender': '$_id.gender', 'total': 1}}"
    })
    //Not working for some reason - throws a stack overflow error
    List<StudentsByGender> getStudentsByGender();

}

Let’s try to do the same aggregations using MongoTemplate:



import org.springframework.data.mongodb.core.aggregation.*;
...

@Repository
@Slf4j
public class CustomStudentRepository {

    private final MongoTemplate mongoTemplate;

    public CustomStudentRepository(MongoTemplate mongoTemplate) {
        log.info("Initialized mongodb template " + mongoTemplate);
        this.mongoTemplate = mongoTemplate;
    }
    //...Other methods    

    record StudentsByGenderForInstitute(String gender, long total) {}

    public List<StudentsByGenderForInstitute> getStudentsByGenderForInstitute(String instituteId) {
        MatchOperation match = Aggregation.match(Criteria.where("instituteId").is(instituteId));
        GroupOperation group = Aggregation.group("userDetails.gender").count().as("total");
        ProjectionOperation project = Aggregation.project()
                .andExclude("_id")
                .and("_id").as("gender")
                .andInclude("total");

        Aggregation aggregation = Aggregation.newAggregation(match, group, project);

        AggregationResults<StudentsByGenderForInstitute> results =
                mongoTemplate.aggregate(aggregation, Student.class, StudentsByGenderForInstitute.class);
        return results.getMappedResults();
    }

    record StudentsByGender(String instituteId, String gender, long total) {}

    public List<StudentsByGender> getStudentsByGender() {
        GroupOperation group = Aggregation.group("instituteId", "userDetails.gender").count().as("total");

        ProjectionOperation project = Aggregation.project().andExclude("_id")
                .and("_id.instituteId").as("instituteId")
                .and("_id.gender").as("gender")
                .andInclude("total");

        Aggregation aggregation = Aggregation.newAggregation(group, project);
        AggregationResults<StudentsByGender> results =
                mongoTemplate.aggregate(aggregation, Student.class, StudentsByGender.class);
        return results.getMappedResults().stream()
                .sorted(Comparator.comparing(result -> result.instituteId)).collect(Collectors.toList());
    }

}

Views

A MongoDB view is a read-only queryable object whose contents are defined by an aggregation pipeline on other collections or views. MongoDB does not persist the view contents to disk. A view’s content is computed on-demand when a client queries the view.

Standard views use the indexes of the underlying collection. As a result, you cannot create, drop or re-build indexes on a standard view directly, nor get a list of indexes on the view.

A view definition pipeline cannot include the $out or the $merge stage. This restriction also applies to embedded pipelines, such as pipelines used in $lookup or $facet stages.

You can create a view/materialised view on any aggregation pipeline, but it’ll make the most sense if you are able to join multiple collections! — $lookup.

Lookup

$lookup performs a left outer join to a collection in the same database to filter in documents from the “joined” collection for processing. The $lookup stage adds a new array field to each input document. The new array field contains the matching documents from the “joined” collection. The $lookup stage passes these reshaped documents to the next stage. You’ll probably end up using $unwind (Kind of like flatMap) with $lookup since the joined collections is added as an array field!

Example: Create a View of Students with Institute Details

Let’s start with something:



//Aggregation.lookup(from, localfield, foreignfield, as);   
LookupOperation lookup = Aggregation.lookup("institutes", "instituteId", "_id", "institute");    
AggregationResults<Object> results = mongoTemplate.aggregate(Aggregation.newAggregation(lookup), Student.class, Object.class);

The results are like this (notice the array):

Let’s add the $unwind stage



LookupOperation lookup = Aggregation.lookup("institutes", "instituteId", "_id", "institute");
UnwindOperation unwind = Aggregation.unwind("institute");
AggregationResults<Object> results = mongoTemplate.aggregate(Aggregation.newAggregation(lookup, unwind), Student.class, Object.class);

Let’s combine these operations and create our view:



record StudentWithInstituteDetails(String _id, User userDetails, List<Hobby> hobbies, Institute institute) {}

@PostConstruct
public void createStudentsWithInstituteDetailsView() {
    LookupOperation lookup = Aggregation.lookup("institutes", "instituteId", "_id", "institute");
    UnwindOperation unwind = Aggregation.unwind("institute");
    AggregationOperation[] aggregationOperations = {lookup, unwind};
    mongoTemplate.createView("students-with-institute-details", Student.class, aggregationOperations);
}

public List<StudentWithInstituteDetails> getStudentsWithInstituteDetails() {
    return mongoTemplate.find(new Query(), StudentWithInstituteDetails.class, "students-with-institute-details");
}

Materialized Views

An on-demand materialized view is a pre-computed aggregation pipeline result that is stored on and read from disk. On-demand materialized views are typically the results of a $merge or $out stage. Materialized views are therefore much more performant than standard views.

$merge is preferred but make sure you think about how your materialised view will get refreshed when data updates happen. You don’t want your entire data to be refreshed every time a single data update happens!

Example — Create a materialised view for the following aggregation — Group students by school and gender

Your initial run should cover all schools and all students, but when a student enrols or leaves a particular school, you want to rerun the pipeline for only that combination

Example:



Materialised View =
{_id: {school: A, gender: Female}, count: 50}
{_id: {school: A, gender: Male}, count: 55}
{_id: {school: B, gender: Female}, count: 60}
{_id: {school: B, gender: Male}, count: 40}
...

Eg. a new female student joins school A
You only want the first document to be updated! (_id!)



@PostConstruct
public void createSchoolGenderView() throws Exception {
    LookupOperation lookup = Aggregation.lookup("institutes", "instituteId", "_id", "institute");
    UnwindOperation unwind = Aggregation.unwind("institute");
    GroupOperation group = Aggregation.group("institute.name", "userDetails.gender").count().as("count");
    MergeOperation merge = Aggregation.merge().intoCollection("school_gender_view").whenMatched(MergeOperation.WhenDocumentsMatch.replaceDocument()).build();

    mongoTemplate.aggregate(
            Aggregation.newAggregation(lookup, unwind, group, merge),
            Student.class,
            Object.class
    );
}

But how do I refresh this? You could create something like this:



public void createSchoolGenderView(String instituteId, String gender) {
    MatchOperation match = null;
    if (StringUtils.hasLength(instituteId) && StringUtils.hasLength(gender)) {
        match = Aggregation.match(
                Criteria.where("instituteId").is(instituteId)
                        .and("userDetails.gender").is(gender)
        );
    }

    LookupOperation lookup = Aggregation.lookup("institutes", "instituteId", "_id", "institute");
    UnwindOperation unwind = Aggregation.unwind("institute");
    GroupOperation group = Aggregation.group("institute.name", "userDetails.gender").count().as("count");
    MergeOperation merge = Aggregation.merge().intoCollection("school_gender_view").whenMatched(MergeOperation.WhenDocumentsMatch.replaceDocument()).build();
    Aggregation aggregation = null;
    if (Objects.nonNull(match)) {
        aggregation = Aggregation.newAggregation(match, lookup, unwind, group, merge);
    } else {
        aggregation = Aggregation.newAggregation(lookup, unwind, group, merge);
    }

    mongoTemplate.aggregate(
            aggregation,
            Student.class,
            Object.class
    );

}

Java Multithreading — Thread states and introduction to thread profiling

Ishan Soni — Thu, 04 Jan 2024 12:41:20 +0000

Thread states

A Thread can be in any one of the following states at any point in time:

NEW

This is the state of a thread when it is created but not yet started. For example, when you create a new Thread object using the constructor, the thread is in the new state. You can only call the start() method on a new thread; otherwise, an IllegalThreadStateException will be thrown.

RUNNABLE

This is the state of a thread when it is ready to run and waiting for CPU time. For example, when you call the start() method on a new thread, or when a blocked or waiting thread becomes eligible to run again. The thread scheduler selects a thread from the runnable state for execution, based on its priority and other factors.

RUNNING

This is the state of a thread when it is executing on the CPU. For example, when the thread scheduler picks a thread from the runnable state and assigns it to a CPU core, the thread is in the running state. The thread executes its run() method and performs its tasks. A thread can enter the running state only from the runnable state.

BLOCKED

This is the state of a thread when it is waiting to acquire a monitor lock that is held by another thread. For example, when a thread tries to enter a synchronized block or method that is already occupied by another thread, the thread is blocked until the lock is released. A thread can enter the blocked state from the running state, and can return to the runnable state when it acquires the lock.

WAITING

This is the state of a thread when it is waiting indefinitely for another thread to perform a certain action. For example, when a thread calls the wait() method on an object, or the join() method on another thread, or the park() method of LockSupport class, the thread is in the waiting state. A thread can enter the waiting state from the running state, and can return to the runnable state when it receives a notification or an interruption.

TIMED WAITING

This is the state of a thread when it is waiting for a specified amount of time for another thread to perform a certain action. For example, when a thread calls the sleep() method, or the wait() method with a timeout parameter, or the join() method with a timeout parameter, or the parkNanos() or parkUntil() methods of LockSupport class, the thread is in the timed waiting state. A thread can enter the timed waiting state from the running state, and can return to the runnable state when the timeout expires or when it receives a notification or an interruption.

TERMINATED

This is the state of a thread when it has completed its execution or has been aborted due to an exception or an error. For example, when a thread finishes its run() method normally or abnormally, or when it is stopped by calling the stop() method (which is deprecated), the thread is in the terminated state. A thread can enter the terminated state only from the running state, and cannot return to any other state.

What is the state of a thread when it is waiting for an I/O operation to complete? — It’s either Blocked or Timed Waiting, depending on the type of I/O operation and the API used. When a thread performs a traditional blocking I/O operation, such as reading a file or a socket, it enters the blocked state and waits for the I/O operation to finish. However, some I/O operations may have a timeout parameter that specifies how long the thread should wait for the operation to complete. In this case, the thread enters the timed waiting state and waits for either the operation to finish or the timeout to expire.

Can the OS put a thread executing a synchronised block into the blocked state mid way and schedule a blocked thread that is waiting for the lock? — No. This is because the synchronized block ensures that only one thread can enter the block at a time, and the thread that enters the block acquires the monitor lock of the object that is used for synchronization. The OS cannot interrupt or preempt a thread that holds a monitor lock, unless the thread voluntarily releases the lock by exiting the block or by calling wait() on the object.

Thread Profiling

Thread Profiling can help us in understanding what states threads are in and why. We can also see if the threads are able to run in parallel or not (i.e see the concurrency level of threads). It also tells us how much time threads spend in different states. We can also find and analyse cases of high lock contention. We’ll use Java Flight Recorder (JFR) to profile threads.
What is Java Flight Recorder and how to enable it and do a flight recording? — JFR

When we profile our application using JFR, it’ll show different thread events. Let’s try to map these thread events to the states we discussed above:

Java Thread Parked

This event occurs when a thread calls the LockSupport.park() method or its variants, which causes the thread to suspend its execution until it is unparked by another thread or a timeout expires. This is often used by concurrency utilities, such as java.util.concurrent.locks, to implement efficient blocking mechanisms. This event corresponds to the Waiting or Timed Waiting state, depending on whether the park() method was called with a timeout parameter or not.

Java Thread Sleep

This event occurs when a thread calls the Thread.sleep() method or its variants, which causes the thread to suspend its execution for a specified amount of time. This is often used to introduce delays or pauses in the execution flow. This event corresponds to the Timed Waiting state.

Java Monitor Wait

This event occurs when a thread calls the Object.wait() method or its variants, which causes the thread to release the monitor (lock) of an object and wait until it is notified by another thread or a timeout expires. This is often used to implement inter-thread communication and coordination. This event corresponds to the Waiting or Timed Waiting state, depending on whether the wait() method was called with a timeout parameter or not.

Java Monitor Blocked

This event occurs when a thread tries to acquire the monitor lock (inside a synchronized method/block) of an object that is already owned by another thread, which causes the thread to block until the monitor is released. This is often used to implement mutual exclusion and synchronization. This event corresponds to the Blocked state.

Socket Read

This event occurs when a thread calls the SocketInputStream.read() method or its variants, which causes the thread to read data from a socket. This is often used to implement network communication. This event corresponds to the Blocked state.

Socket Write

This event occurs when a thread calls the SocketOutputStream.write() method or its variants, which causes the thread to write data to a socket. This is also often used to implement network communication. This event corresponds to the Blocked state.

File Read

This event means that the thread is reading data from a file by using the read() method of the File class or its subclasses. This method can block the thread until some data is available or an exception occurs. This event corresponds to the Blocked state.

File Write

This event means that the thread is writing data to a file by using the write() method of the File class or its subclasses. This method can block the thread until the data is written or an exception occurs. This event corresponds to the Blocked state.

Profiling examples:

Example 1

Let’s assume we have a SpringBoot (embedded tomcat, default thread pool size = 200) service that exposes a GET /portfolio/{userId} endpoint to download a users portfolio. The controller forwards the request to the PortfolioDownloader class. This class allows only 5 threads to execute the critical section in parallel at any point in time using a Semaphore (i.e only 5 concurrent download requests). If we fire 200 concurrent download requests, what would be the state of the rest of the threads? Let’s profile and find out:



public enum PortfolioDownloader {

  INSTANCE;

  private static final Semaphore lock = new Semaphore(5);

  public String downloadPortfolio(String userId) {
    try {
      lock.acquire();
    } catch (InterruptedException e) {
      throw new RuntimeException(e);
    }
    System.out.println("Thread " + Thread.currentThread() + " downloading portfolio for user " + userId);
    //Mock download
    long aMockVariable = 0L;
    long downloadStartTime = System.currentTimeMillis();
    while (System.currentTimeMillis() - downloadStartTime < 2000) {
      aMockVariable++;
    }

    lock.release();
    return "s3://portfolioapp-" + aMockVariable + ".portfolio-" + userId;
  }

}

i.e Thread 5, 4, 7, 1, and 2 acquire the permit first and go about their task while Thread 23 (at the end) gets the permit after some time. This will also reflect in the Thread profiling page:

See how Thread 23 was parked for some time until it got the permit

Note — Threads call the Semaphore.accquire() method to get a permit. If a permit is not available, the method internally uses the LockSupport.park() to suspend threads until a permit is available!

Example 2

Let’s assume we have the following SpringBoot application:



@RestController
@RequestMapping("/v1/counter")
public class CounterResource {

  @PostMapping
  public ResponseEntity<Integer> increment() {
    return ResponseEntity.ok(Counter.increment());
  }

}



public final class Counter {

  private static int COUNT = 0;

  public synchronized static int increment() {
    try {
      Thread.sleep(50);
    } catch (InterruptedException e) {
      throw new RuntimeException(e);
    }
    ++COUNT;
    System.out.println("Thread " + Thread.currentThread() + " called. Current count = " + COUNT);
    return COUNT;
  }

}

We’ll fire 2087 concurrent requests to this API

Thread 14 acquires the monitor on the object first. See how Thread 68 goes into the Blocked State (Java Monitor Blocked) before it is able to acquire the monitor.

Try to do thread profiling of a simple consumer producer (using wait() and notify()) application yourself using JFR and note the thread states!

Introduction to unit testing in Java

Ishan Soni — Thu, 28 Dec 2023 14:52:29 +0000

The point of writing tests is not only to ensure that your code works now, but to ensure that the code will continue to work in the future.

What does a typical test run involve?

Prepration
Provide test input
Running the test case
Provide expected output
Verify output (expected vs actual)
Do something if the test fails (expected and output do not match)

3, 5, 6 are common tasks while 1, 2, 4 change based on your test. Instead of doing everything yourself, offload the common tasks to a framework — JUnit is the defacto unit testing framework in Java. The current JUnit framework release is 5.

Problems with JUnit4

Outdated. > 10 years old
No support for newer testing patterns
Was playing catch up with newer java features
Had a monolothic architecture (one jar that had it all)

JUnit5 Architecture

Setup

Let’s test the deposit and withdraw functionality of the Account aggregate:

public class Transaction {

    public enum Type {
        CREDIT, DEBIT, HOLD
    };

    public Transaction(String accountId, Type type, BigDecimal amount, String reference) {
        this.id = UUID.randomUUID().toString();
        this.transactionTime = System.currentTimeMillis();
        this.accountId = accountId;
        this.type = type;
        this.amount = amount;
        this.reference = reference;
    }

    private final String id;

    private final String accountId;

    private final long transactionTime;

    private final Type type;

    private final BigDecimal amount;

    private final String reference;

    //Getters

}

public class Account {

    private String id;

    private String userId;

    private BigDecimal balance = BigDecimal.ZERO;

    Transaction deposit(BigDecimal amount, String reference) {
        balance = balance.add(amount);
        return new Transaction(id, Transaction.Type.CREDIT, amount, reference);
    }

    Transaction withdraw(BigDecimal amount, String reference) {
        balance = balance.subtract(amount);
        return new Transaction(id, Transaction.Type.DEBIT, amount, reference);
    }

    //Getters and Setters

}

Add the jupiter-engine and jupiter-api dependency.

<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>org.junit</groupId>
            <artifactId>junit-bom</artifactId>
            <version>5.8.2</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>  
    </dependencies>
</dependencyManagement>

<dependencies>
    <dependency>
        <groupId>org.junit.jupiter</groupId>
        <artifactId>junit-jupiter-api</artifactId>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>org.junit.jupiter</groupId>
        <artifactId>junit-jupiter-engine</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

Create a test class in the src/java/test directory using the same package convention (so that you get access to all default properties of the class you want to test) and annotate your methods with @test (from the jupiter jar!) or use your IDE to create a test:

class AccountTest {

    @Test
    public void testDeposit() {
        Account account = new Account();
        Transaction transaction1 = account.deposit(BigDecimal.TEN, "FY 23-24 qtr 2 interest");
        Transaction transaction2 = account.deposit(BigDecimal.ONE, "Transaction XYZ Surcharge waiver");

        assertEquals(BigDecimal.valueOf(11), account.getBalance());
        assertEquals(Transaction.Type.CREDIT, transaction1.getType());
        assertEquals(BigDecimal.TEN, transaction1.getAmount());

    }

}

Running the test from the IDE:

Assertions (Leverage JUnit’s way of reporting an error!)

import static org.junit.jupiter.api.Assertions.*; //static import - contains all the assert methods!

assertEquals(expected, actual) //Example

Using assertions you can assert/verify if your code is producing the desired results or not. Your test will fail if these assertions fail. Example, let’s change our deposit method and introduce a bug:

Transaction deposit(BigDecimal amount, String reference) {
    //We are not using the return value of the add method. 
    //Note: BigDecimal is an immutable object! 
    balance.add(amount);
    return new Transaction(id, Transaction.Type.CREDIT, amount, reference);
}

If we run our test case now, it will fail:

Be a little more descriptive when your test fails:

assertEquals(BigDecimal.valueOf(11), account.getBalance(), "The account balance after two deposits of 10 and 1 should be 11");

Asserting Exceptions using assertThrows

public class MathUtils {

    //Other methods omitted

    public int divide(int a, int b) {
        return a / b;
  }

}

class MathUtilsTest {

    @Test
    void testDivideBy0() {
        MathUtils mathUtils = new MathUtils();
        /*
        assertThrows(Exception class, Executable)
        public interface Executable {
            void execute() throws Throwable;
        }
       */
       assertThrows(ArithmeticException.class, () -> mathUtils.divide(1, 0),
    "Divide by 0 should raise an arithmetic exception");
    }

}

Running tests using maven instead of your IDE

If you try to run mvn test, the tests are not executed!

Use the maven surefire plugin. This will allow you to run tests using the mvn test command (The plugin binds a goal to execute all test when the test phase is executed). You can then potentially automate test execution in your CI/CD pipeline!

<build>
    <plugins>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-surefire-plugin</artifactId>
            <version>3.0.0-M7</version>
        </plugin>
    </plugins>
</build>

The Test Life Cycle and Hooks (BeforeAll, BeforeEach, AfterEach, AfterAll)

class AccountTest

    private Account account;

    @BeforeAll
    public static void beforeAll() {
        System.out.println("In before all");
    }

    @BeforeEach
    public void beforeEach() {
        System.out.println("In before each");
        account = new Account();
    }

    @AfterEach
    public void afterEach() {
        System.out.println("In after each");
    }

    @AfterAll
    public static void afterAll() {
        System.out.println("In after all");
    }

    @Test
    public void testDeposit() {
        Transaction transaction1 = account.deposit(BigDecimal.TEN, "FY 23-24 qtr 2 interest");
        Transaction transaction2 = account.deposit(BigDecimal.ONE, "Transaction XYZ Surcharge waiver");

        assertEquals(BigDecimal.valueOf(11), account.getBalance(), "The account balance after two deposits of 10 and 1 should be 11");
        assertEquals(Transaction.Type.CREDIT, transaction1.getType());
        assertEquals(BigDecimal.TEN, transaction1.getAmount());
    }

    @Test
    public void testWithdraw() {
        Transaction transaction = account.withdraw(BigDecimal.TEN, "ATM withdrawal");

        assertEquals(BigDecimal.valueOf(-10), account.getBalance(), "The account balance after a withdrawal of 10 should be -10");
        assertEquals(Transaction.Type.DEBIT, transaction.getType());
        assertEquals(BigDecimal.TEN, transaction.getAmount());
    }

}

Important — The @BeforeAll and @AfterAll are called before the instance of the test class is created/after the instance of the test class is destroyed. Therefore, they should be static methods!

Paramaterised Tests

Allows you to execute a single test method multiple times with different parameters

Also allows you to specify the source of your tests

In order to use parameterised test, add the junit-jupiter-params artifact

<dependency>
    <groupId>org.junit.jupiter</groupId>
    <artifactId>junit-jupiter-params</artifactId>
    <scope>test</scope>
</dependency>

Then annotate your test methods with @ParameterizedTest + Specify a Source. Example:

public class MathUtils {

  public double areaOfCircle(double radius) {
    return Math.PI * radius * radius;
  }

  public int add(int a, int b) {
    return a + b;
  }

  public int divide(int a, int b) {
    return a / b;
  }

}

@ParameterizedTest
@ValueSource(doubles = {1d, 2d, 3d, 4d, 5d})
void testCircleArea(double radius) {
  MathUtils math = new MathUtils();
  double expected = Math.PI * radius * radius;
  double actual = math.areaOfCircle(radius);

  assertEquals(expected, actual, "Area of circle is PI * r * r");
}

As you can see, the single test method was run 5 times with values 1, 2, 3, 4, and 5 passed to the double radius parameter respectively.

With the @ValueSource you can pass a single literal value (ints, strings, floats, doubles, etc. to your test case)

Use @NullSource, @EmptySource, @NullAndEmpty source to pass null/empty sources

Use @EnumSource to pass enums

Use @CsvSource to pass in multiple literal arguments

@ParameterizedTest
@CsvSource({"1, 1, 2", "2, 2, 4", "5, 5, 10", "15, 15, 30", "50, 50, 100"})
void testAdd(int a, int b, int expected) {
  MathUtils math = new MathUtils();
  int actual = math.add(a, b);
  assertEquals(expected, actual, "adding " + a + ", " + b + " should result in " + expected);
}

Use @MethodSource to provide more complex arguments

@ParameterizedTest
@MethodSource("divideSource")
void testDivideBy0(int a, int b) {
  /*
  assertThrows(Exception class, Executable)
  public interface Executable {
     void execute() throws Throwable;
  }
   */
  assertThrows(ArithmeticException.class, () -> this.math.divide(a, b),
      "Divide by 0 should raise an arithmetic exception");
}

//Should be static and return a stream/collection of Arguments
//Each Arguments item will initiate a test execution using said arguments!
//Can be in a separate class as well. You'll then need to use fully qualified name in the @MethodSource annotation
//Arguments.of(...Object) - Can pass complex arguments
static Stream<Arguments> divideSource() {
  return Stream.of(
      Arguments.of(1, 0),
      Arguments.of(2, 0),
      Arguments.of(3, 0)
  );
}

Mockito

Mockito is a Java-based mocking framework that is used for unit testing of Java applications. It allows you to create mock objects that simulate the behavior of real objects, and verify their interactions and expectations. Mockito can be used with other testing frameworks, such as JUnit and is useful for testing components that depend on other components, but are not yet available or are difficult to access. By using mock objects, you can isolate the component under test and focus on its logic and functionality.

Add the mockito-core and the mockito-junit-jupiter (A 3rd party extension library for junit and mockito integration) dependency

<!-- https://mvnrepository.com/artifact/org.mockito/mockito-core -->
<dependency>
  <groupId>org.mockito</groupId>
  <artifactId>mockito-core</artifactId>
  <version>4.9.0</version>
  <scope>test</scope>
</dependency>
<!-- https://mvnrepository.com/artifact/org.mockito/mockito-junit-jupiter -->
<dependency>
  <groupId>org.mockito</groupId>
  <artifactId>mockito-junit-jupiter</artifactId>
  <version>4.9.0</version>
  <scope>test</scope>
</dependency>

Let’s extend our Accounts functionality and add a service. This example assumes you have a SpringBoot project (with Spring Data). The Account and Transaction Repository are Spring Data repositories.

@Service
public class AccountApplicationService {

    @Autowired
    private AccountRepository accountRepository;

    @Autowired
    private TransactionRepository transactionRepository;

    @Transactional
    public void deposit(String accountId, BigDecimal amount, String reference) {
        Account account = accountRepository.getAccountById(accountId).orElseThrow();
        Transaction transaction = account.deposit(amount, reference);
        transactionRepository.save(transaction);
        accountRepository.save(account);
    }

}

Let’s try to test AccountApplicationService. To test it, I want to Mock the AccountRepository and TransactionRepository.

@ExtendWith(MockitoExtension.class)
class AccountApplicationServiceTest {

    private Account account;

    //Create a mock for this dependency
    @Mock
    private AccountRepository accountRepository;

    //Create a mock for this dependency
    @Mock
    private TransactionRepository transactionRepository;

    //@InjectMocks creates an instance of this class and injects the mocks that are created with the @Mock (or @Spy) annotations into this instance
    @InjectMocks
    private AccountApplicationService accountApplicationService;

    @BeforeEach
    public void beforeEach() {
        this.account = new Account();
        //Whenever the accountRepository.getAccountById method is called,
        //return the given account object
        Mockito.when(accountRepository.getAccountById(Mockito.anyString())).thenReturn(Optional.of(account));
    }

    @Test
    public void testDeposit() {
        accountApplicationService.deposit("temp", BigDecimal.TEN, "FY 23-24 qtr 2 interest");
        assertEquals(BigDecimal.TEN, account.getBalance());
        //Verify if the deposit call resulted in a call to the save method of the transactionRepository
        Mockito.verify(transactionRepository).save(Mockito.any());
    }

}

Saga, CDC with Transactional Inbox/Outbox

Ishan Soni — Fri, 22 Dec 2023 11:22:07 +0000

In a distributed system, some business transactions may span multiple services. How to implement transactions that span multiple services? — The Saga Pattern

And not just a business transaction, what are some other strategies to keep a distributed system eventually consistent? i.e. when there are multiple systems that store related data across them, what are some common strategies to keeping them consistent? — another such strategy is CDC (Change Data Capture)

Don’t confuse Saga and CDC

Saga is a substitute for a distributed transaction. An example would be: Let’s assume that our business rule (invariant) says that when a user places an order, it will only be fulfilled if the order total is within the user’s credit limit or balance (and deduct the order total from the credit limit). If the credit limit check/deduction fails, the order will not be fulfilled. In this case, the transaction spans over the order service (which manages the order) and the customer service (which manages the customer credit). You can use Saga in this case.

An example for CDC would be: Let’s assume you have a customer service that manages customer data (user-handle, email, name, etc.) and a notification service that sends out emails. The notification service caches some user data so that it does not have to query the customer service every time it has to send out an email to the customer. Now, what if the customer changes their name? That change has to be propagated to the notification service as well. Note that a transaction in this case finishes as soon as the change is committed in the customer service. But since some customer-related data is stored by the notification service as well, we need to make it consistent. You can use CDC in this case.

But there are still some problems we need to tackle in both Saga and CDC:

Guaranteed event/message delivery

Suppose you have a microservice (A) that needs to update it’s own database and publish an event to another microservice (B, using a message broker) as part of a saga step. It will probably do the following:

Update the concerned aggregate/entity in a transaction
Create and send an event to the message broker

Now, what if the event publishing to the messaging broker failed (let’s say because of a network issue?)

Retry capabilities

Let’s say you are using CDC. Sometimes, you may want retry capabilities in your application. What if microservice B is not able to consume an event due to a data/code issue? You spot it and quickly do a hotfix. But, you may also want to retry the events that were sent by microservice A which microservice B was not able to consume before the hotfix!

Use the Transactional Inbox/Outbox Pattern along with Saga/CDC

Transactional Inbox/Outbox pattern is a technique used in microservices architecture to ensure reliable and consistent data exchange between different services. It works by using a database table as an intermediary between the service that produces the data and the service that consumes it. The producer service inserts the data into the outbox table as part of the original business transaction (which ensures atomicity of the database update and the event insertion), and then a separate process or component reads the data from the outbox table and publishes it to a message broker or another service. The consumer service can then receive the data from the message broker, and store it in its own inbox table. This way, the data is always available for the consumer service, even if the producer service or the message broker is down or unreachable. The Transactional Inbox/Outbox pattern can help with some common challenges in microservices communication, such as atomicity, durability, scalability, and performance.

i.e. whenever you want to publish an event, you first put it into an outbox table (in the same transaction as whatever generated the event in the first place). Some event emitter polls those and sends them, giving you at-least once delivery (solves the guaranteed delivery problem). If the business trasaction generating the event gets rolled back, the event dosen’t get saved; if the emitter fails to publish, the event still sits in the DB. You can add a status to each event in the DB (PENDING, PUBLISHED). The poller will pick up PENDING events and mark them as PUBLISHED after publishing them to a message broker. Setting the status of an event back to PENDING will allow you to resend the message if needed (solves the retry capability problem). You may be tempted to skip the inbox table but it may come in handy to create idempotent consumers. In case you want to retry an event, your consumers should have the capability to discard duplicate events. If your domain model doesn’t have this capability, the inbox table will be useful here — If the event-id is already present in the inbox table, you can skip this event!

Views, Materialized Views, and Spring Data JPA

Ishan Soni — Thu, 21 Dec 2023 08:20:11 +0000

Views

A view in an RDBMS is a virtual table that is defined by a query. It can combine data from two or more tables, or just contain a subset of information from a single table. A view does not store data, but rather displays the data that is stored in the base tables. You can query a view like you can a table, but you cannot insert, update, or delete data from a view unless certain conditions (not discussed here) are met!

i.e store complex queries under a name (it’s almost like a variable for a query)!

Let’s assume we have the following tables: (We are using Postgres in this example)

The Department table

The Employee table

The Employee-Department table

Now, individually these tables do not mean much. I’ll have to run the following query to get some meaningful data:

select e.id, e.name, e.email, d.name, ed.start_time,  ed.end_time
from employee e join employee_department ed on e.id = ed.employee_id
join department d on d.id = ed.department_id

Let’s say you want to group by department name and count the number of employees:

select d.name, count(e.id) 
from employee e join employee_department ed on e.id = ed.employee_id
join department d on d.id = ed.department_id
group by d.name;

This join query “from employee e join employee_department ed on e.id = ed.employee_id join department d on d.id = ed.department_id” is duplicated every time you want to work with employees and departments!

I can instead, create a view off of this query and give it a name and treat it as if it were a table!

-- create view <view-name> AS <your query>

create view employee_with_department_details AS

select e.id as employee_id, e.name as employee_name, e.email, d.name as department_name, ed.start_time, ed.end_time
from employee e join employee_department ed on e.id = ed.employee_id
join department d on d.id = ed.department_id;

Now, I can directly query this view as if it were a table

select * from employee_with_department_details;

select department_name, count(employee_id)
from employee_with_department_details
group by department_name;

Important — Whenever you execute a view, the query behind the view is executed every-time!

Query views directly in a Spring Data Jpa application using Interface Projections

interface EmployeeWithDepartmentDetails {
    @Value("#{target.employee_id}")
    Long getEmployeeId();
    @Value("#{target.employee_name}")
    String getEmployeeName();
    String getEmail();
    @Value("#{target.department_name}")
    String getDepartmentName();
    @Value("#{target.start_time}")
    Long getStartTime();
    @Value("#{target.end_time}")
    Long getEndTime();
}

@Repository
public interface DepartmentJpaRepository extends JpaRepository<Department, Long> {

    @Query(
            value = "select * from employee_with_department_details;",
            nativeQuery = true
    )
    List<EmployeeWithDepartmentDetails> getEmployeeWithDepartmentDetails();

}

Query the view:

Materialized Views

The main difference between a view and a materialized view in an RDBMS is that a view does not store any data, while a materialized view does. A view is a virtual table that is defined by a query, and it displays the data that is stored in the base tables. A materialized view is a view that stores the result set of the query in a physical table, and it can be queried like a regular table. However, a materialized view needs to be refreshed periodically to reflect the changes in the base tables, while a view is always up to date.

Therefore, a materialised view does 2 things:

Store the query that is used to create this materialised view (also done by a view)
Store the data returned by this query!

Unlike a normal view, a materialised view does not execute the query every time you fetch it, but rather returns the data stored by the view thus improving performance. That is why the data in a materialised view can become stale and require a refresh.

-- create materialized view <view-name> AS <your query>

create materialized view employee_with_department_details  AS

select e.id as employee_id, e.name as employee_name, e.email, d.name as department_name, ed.start_time, ed.end_time
from employee e join employee_department ed on e.id = ed.employee_id
join department d on d.id = ed.department_id;

select * from employee_with_department_details;

To refresh a materialised view:

refresh materialized view employee_with_department_details;

Query and update materialized views in a Spring Data Jpa application

Example use case: We want to refresh the materialized view whenever a new employee is added to a department so that the data returned by the materialized view is always current

@Transactional
public AddEmployeeToDepartment addEmployeeToDepartment(Long employeeId, Long departmentId) {
    final Department department = departmentJpaRepository.findById(departmentId).orElseThrow();
    final Employee employee = employeeJpaRepository.findById(employeeId).orElseThrow();
    final EmployeeDepartment employeeDepartment = new EmployeeDepartment(employeeId, departmentId);
    employeeDepartmentJpaRepository.save(
        employeeDepartment
    );

    //Refresh the materialised view since it will contain stale data otherwise!
    employeeDepartmentJpaRepository
        .refreshEmployeeWithDepartmentDetails();

    return new AddEmployeeToDepartment(true, null);
}

@Repository
public interface EmployeeDepartmentJpaRepository extends JpaRepository<EmployeeDepartment, Long> {

    @Query(
            value = "select * from employee_with_department_details;",
            nativeQuery = true
    )
    List<EmployeeWithDepartmentDetails> getEmployeeWithDepartmentDetails();

    /*
    The refresh materialized view statement does not return any result set,
    but Spring Data JPA expects one. To fix this, you need to annotate your method with @Modifying,
    which tells Spring Data JPA that the query is a DML statement and does not return any result.
    You can also use the clearAutomatically attribute to clear the persistence context after the query execution,
    which might be useful if you want to query the refreshed view afterwards.
    */
    @Modifying(clearAutomatically = true)
    @Transactional
    //use @Async instead!
    @Query(
            value = "refresh materialized view employee_with_department_details;",
            nativeQuery = true
    )
    void refreshEmployeeWithDepartmentDetails();

}

Query the materialised view:

Let’s try to add a new employee (Mr Robot) and tag them to the IT department and see if the materialized view is refreshed or not:

Ideally, you should not use materialized views for data that is updated very frequently. The data is basically static until you refresh it, at which time it’s flushed and the data is replaced by the result of the query at the new run time. They’re particulary good when the performance to run the query is poor but the data doesn’t have to be exact or up to the last second. For example, if you wanted to run a query that generates a report for the previous day, you could create a materialized view to get the data from yesterday and run it on a schedule after midnight. Then the user can query the materialized view with a select * in the morning and get quick results without waiting on the query to execute against the base data. Sometimes, it makes sense to have the materialized view to contain most of the result set and then some optimized query to just pull data from the current day, hour, etc. and union the results together.

Using Spring Application Events within Transactional Contexts

Ishan Soni — Wed, 20 Dec 2023 13:05:31 +0000

This post is inspired by this talk given by Bartłomiej Słota. I highly recommend checking it out.

Application events are a great way of communicating between beans defined in the same application or bounded context.

It is a great way of implementing the Observer Design Pattern. "Hey, a customer is created. If anybody is interested, here is the event. Do whatever you want to do with it!"

In Spring, you can publish events using the ApplicationEventPublisher, which is the super interface for ApplicationContext and is the abstraction around the Spring event bus. In the latest versions of Spring, you can simply use POJOs as events.

Example: Create a customer and then generate a verification token that will be sent to their email ID using which they can verify their account. These two should be independent processes, i.e., creating a customer and generating a verfication token should happen in two separate transactions!

The Customer entity and the CustomerJpaRepository

@Entity
public class Customer {

    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
    private String name;
    private String email;
    private String verificationToken;
    private boolean verified;

    public Customer(String name, String email) {
        this.name = name;
        this.email = email;
    }

    public void setVerificationToken(String generatedToken) {
        this.verificationToken = generatedToken;
    }

    public boolean verifyCustomer(String providedToken) {
        if (StringUtils.hasLength(this.verificationToken) && this.verificationToken.equals(providedToken)) {
            this.verified = true;
        }
        return this.verified;
    }
    //Getters and Setters
}

@Repository
public interface CustomerJpaRepository extends JpaRepository<Customer, Long> {
}

The CustomerService. The createCustomer method will create a customer and emit a CustomerCreated event using Spring’s ApplicationEventPublisher

@Service
public class CustomerService {

    private final CustomerJpaRepository customerJpaRepository;

    private final ApplicationEventPublisher applicationEventPublisher;

    public CustomerService(CustomerJpaRepository customerJpaRepository, ApplicationEventPublisher applicationEventPublisher) {
        this.customerJpaRepository = customerJpaRepository;
        this.applicationEventPublisher = applicationEventPublisher;
    }

    @Transactional
    public Customer createCustomer(String name, String email) {
        final Customer customer = customerJpaRepository.save(new Customer(name, email));
        applicationEventPublisher.publishEvent(new CustomerCreated(customer.getId()));
        return customer;
   }

}

To create subscribers, you can use Spring’s @EventListener

@Component
public class VerificationTokenGenerator {

    private final CustomerJpaRepository customerJpaRepository;

    public VerificationTokenGenerator(CustomerJpaRepository customerJpaRepository) {
        this.customerJpaRepository = customerJpaRepository;
    }

    @EventListener
    public void generateVerificationToken(CustomerCreated customerCreated) {
        final Long customerId = customerCreated.getId();
        final Customer customer = customerJpaRepository.findById(customerId).get();
        customer.setVerificationToken(
            String.valueOf(
                Objects.hashCode(customer)
            )
        ); //Dummy

        customerJpaRepository.save(customer);
    }

}

Is there a separation of concerns here? i.e does the token generation process happen in a separate transaction? — No!

Why? — By default, Spring events are synchronous i.e the producer won’t proceed unless all listeners have been executed i.e if the producer is executing inside a transaction, the transaction is propagated to all listeners i.e in this case the generateVerificationToken() is executed within the same transaction as createCustomer()

See above — The createCustomer() has executed, and we are currently in the generateVerificationToken(), still a “select * from customer” query returns no result! Once the generateVerificationToken method is successful, the transaction is committed!

But, how does customerJpaRepository.findById() in the generateVerificationToken() returns back a customer if the transaction is not committed? — Hibernate’s First Level Cache!

But I want the generateVerificationToken() to be executed in a separate transaction. How do we get around this? — use @TransactionalEventListener. i.e using @TransactionalEventListener, you can collaborate with surrounding transactions using Phases!

By default the phase is AFTER_COMMIT, i.e execute this listener only after the enclosing transaction has committed

If we change our code to this:

@TransactionalEventListener
public void generateVerificationToken(CustomerCreated customerCreated) {
    final Long customerId = customerCreated.getId();
    final Customer customer =   customerJpaRepository.findById(customerId).get();
    customer.setVerificationToken(
        String.valueOf(
            Objects.hashCode(customer)
        )
    ); //Dummy

    customerJpaRepository.save(customer);
}

The event listener receives the event only after the transaction has committed. But the verification token is not saved. It’s like the second save didn’t work at all!

Think about it, the listener is executed only after the original transaction commits. A transaction is all or nothing. Once the transaction has committed, you cannot do anything more!

Use this to do something after the transaction has successfully committed. Things like sending an email or to publish a message to a messaging broker, but if you want to do additional DB work, it won’t work! Instead you need to start another transaction!

    @TransactionalEventListener
    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public void generateVerificationToken(CustomerCreated customerCreated) {
        final Long customerId = customerCreated.getId();
        final Customer customer = customerJpaRepository.findById(customerId).get();
        customer.setVerificationToken(
            String.valueOf(
                Objects.hashCode(customer)
            )
        ); //Dummy

        customerJpaRepository.save(customer);
    }

}

i.e execute this listener after the original transaction (AFTER_COMMIT), and open a new Transaction (REQUIRES NEW)!

Rule of Thumb

Use @EventListener if you want this work to be a part of the enclosing/original transaction
Use @TransactionalEventListener (default phase = AFTER_COMMIT) if you want this work to be done after the enclosing transaction is committed. But remember, you won’t be able to do any DB work since the original transaction has already committed. Use this to do other work — send emails, send events to a messaging queue, etc.
Use @TransactionalEventListener (default phase = AFTER_COMMIT) along with @Transactional(propagation=REQUIRES_NEW) to do DB work in a new transaction after the enclosing/original transaction is committed!

The Observer design pattern

Ishan Soni — Wed, 20 Dec 2023 11:56:53 +0000

The Observer design pattern is also called Publish-Subscribe pattern and is the core principle of the Reactive paradigm (PUSH in favor of PULL).

Design Considerations

There are many observers (subscribers that are interested in some data) that are observing a publisher (which maintains that data). Observers register themselves to the publisher. The publisher can then push data to these observers, since it maintains a list of subscribed observers (PUSH!), whenever there is a change in the publisher’s state.

This pattern defines a one-to-many dependency between objects such that when the state of one object changes, all its dependents are notified and updated automatically.

Example: You are working on the user microservice. When a user is created, the following activities have to be performed:

Send out a welcome email to the user.
Generate and send a verification token to this user using which they can verify their account.

One way of structuring our code would be:

public void createUser(CreateUserCommand command) {
    //omitted - code to create and save a user
    sendWelcomeEmail();
    generateAndSendVerificationToken();
}

This violates both the S and O of the SOLID principles. Your UserService is doing too much work and is not extensible. Let's assume, as part of the user creation process, we now need to create an Avatar for the user as well. Your code would then become:

public void createUser(CreateUserCommand command) {
    //omitted - code to create and save a user
    sendWelcomeEmail();
    generateAndSendVerificationToken();
    generateAvatar();
}

Instead, let's try to use the Observer design pattern

The data

public record UserCreated(String userId, String email, String firstName, String lastName) {}

The Publisher interface

public interface Publisher<T> {
    void subscribe(Subscriber<T> subscriber);
    void unsubscribe(Subscriber<T> subscriber);
    void publish(T data);
}

The Subscriber interface

public interface Subscriber<T> {
    void next(T data);
}

Let's create the concrete Publisher and Subscribers

public class UserPublisher implements Publisher<UserCreated> {

    private final List<Subscriber<UserCreated>> subscribers = new ArrayList<>();

    @Override
    public void subscribe(Subscriber<UserCreated> subscriber) {
        subscribers.add(subscriber);
    }

    @Override
    public void unsubscribe(Subscriber<UserCreated> subscriber) {
        subscribers.remove(subscriber);
    }

    @Override
    public void publish(UserCreated data) {
        subscribers.forEach(s -> s.next(data));
    }

}

public class UserCreatedNotifier implements Subscriber<UserCreated>{

    @Override
    public void next(UserCreated data) {
        System.out.println("Sending email to " + data.email());
    }

}

public class VerificationTokenGenerator implements Subscriber<UserCreated> {

    @Override
    public void next(UserCreated data) {
        System.out.println("Generating verification token and sending to " + data.email());
    }
}

Let's test it out

public void createUser(CreateUserCommand command) {
    //... User creation code

    UserPublisher userPublisher = new UserPublisher();

    UserCreatedNotifier userCreatedNotifier = new UserCreatedNotifier();
    VerificationTokenGenerator verificationTokenGenerator = new VerificationTokenGenerator();

    userPublisher.subscribe(userCreatedNotifier);
    userPublisher.subscribe(verificationTokenGenerator);

    userPublisher.publish(new UserCreated("user-1", "user1@gmail.com", "User", "1"));
}

Now, let's work on the Avatar generation process and see how extensible this design pattern is:

public class AvatarGenerator implements Subscriber<UserCreated> {
    @Override
    public void next(UserCreated data) {
        System.out.println("Generating avatar for user " + data.email());
    }

}

//same code as before    
AvatarGenerator avatarGenerator = new AvatarGenerator();
userPublisher.subscribe(avatarGenerator);

Advantages

Loose Coupling: When two objects are loosely coupled, they can interact but have a very little knowledge of each other. The observer design pattern provides an object design where the publisher and subscribers are loosely coupled:
The only thing a publisher knows about a subscriber is that it implements the subscriber interface. We can add new/remove subscribers at any time since the only thing the publisher depends on is a list of objects that implement the subscriber interface.

Changes to either publisher or subscribers will not affect each other.