Kafka to Postgres test

Author: qtips

README.MD

@@ -1,8 +1,8 @@
-# Prerequisite
-Install and run zookeeper and kafka server https://kafka.apache.org/quickstart
+This project involves Kafka consumer and producer, and Postgresql database and a rest-service
 
-# Run
-Go to http://localhost:9011/kafka/publish?message="hello" to produce and consume message
+The tests demonstrate:
+1. Calling a rest service that adds a message to a (embedded) Kafka topic, where the message consumer writes to a postgresql database
+2. Multithreaded usage of postgresql with Spring Data and JPA and locking of rows. 
 
-
-Based on https://www.confluent.io/blog/apache-kafka-spring-boot-application/ 
+# TODO: 
+Simulate Kafka calls that result in errors in multithreaded environment to test different kinds of kafka settings and guarentees (e.g. exactly once vs at least once processing)

src/main/java/no/ainiq/kafkademo/app/HobbyNotificationService.java

@@ -20,12 +20,26 @@ public HobbyNotificationService(HobbyUserRepository repository) {
     }
 
     @Transactional
-    public void notify(String name) {
+    public void notifyUser(String name) {
         HobbyUser user = repository.findByNameAndStatus(name, HobbyUser.NotificationStatus.READY.name());
         if (user == null) {
             LOGGER.info("User with name {} already sent or failed", name);
             return;
         }
+        notify(user);
+    }
+
+    @Transactional
+    public void notifyNextUser() {
+        HobbyUser user = repository.findReady();
+        if (user == null) {
+            LOGGER.info("no users left to process");
+            return;
+        }
+        notify(user);
+    }
+
+    private void notify(HobbyUser user) {
         try {
             sendNotification(user);
             user.success();

src/main/java/no/ainiq/kafkademo/app/HobbyUser.java

@@ -59,6 +59,17 @@ public void failed() {
 
     }
 
+    @Override
+    public String toString() {
+        return "HobbyUser{" +
+                "id=" + id +
+                ", name='" + name + '\'' +
+                ", hobbies=" + hobbies +
+                ", status=" + status +
+                ", tries=" + tries +
+                '}';
+    }
+
     public enum NotificationStatus {
         READY, SENT, FAILED
     }

src/main/java/no/ainiq/kafkademo/app/repository/HobbyUserRepository.java

@@ -4,10 +4,16 @@
 import org.springframework.data.jpa.repository.Query;
 import org.springframework.data.repository.CrudRepository;
 
+import java.util.List;
+
 public interface HobbyUserRepository extends CrudRepository<HobbyUser, Long> {
     HobbyUser findByName(String name);
 
-    //@Lock(LockModeType.PESSIMISTIC_READ)
     @Query(value = "select * from hobbyuser where name=?1 and status = ?2 for update", nativeQuery = true)
     HobbyUser findByNameAndStatus(String name, String status);
+
+    @Query(value = "select * from hobbyuser where status = 'READY' for update skip locked limit 1", nativeQuery = true)
+    HobbyUser findReady();
+
+    List<HobbyUser> findByStatus(HobbyUser.NotificationStatus status);
 }

src/test/java/no/ainiq/kafkademo/ConcurrentPostgresTest.java

@@ -4,6 +4,7 @@
 import no.ainiq.kafkademo.app.HobbyNotificationService;
 import no.ainiq.kafkademo.app.HobbyUser;
 import no.ainiq.kafkademo.app.repository.HobbyUserRepository;
+import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootTest;
@@ -16,9 +17,8 @@
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
 
-import static org.awaitility.Awaitility.await;
+import static org.assertj.core.api.Assertions.assertThat;
 
 @SpringBootTest(value = {
         "kafka-enabled=false",
@@ -44,27 +44,44 @@ public class ConcurrentPostgresTest {
     @Autowired
     HobbyNotificationService service;
 
-    @Test
-    void notification() throws InterruptedException {
+    @Test()
+    @DisplayName("4 threads try to notify each user simultaneously")
+    void test_for_update() throws InterruptedException {
         List<String> names = names(10);
         names.forEach(n -> repository.save(HobbyUser.newInstance(n)));
         ExecutorService executorService = Executors.newFixedThreadPool(4);
 
         List<Callable<Object>> callables = new ArrayList<>();
-        callables.add(Executors.callable(() -> names.forEach(service::notify)));
-        callables.add(Executors.callable(() -> names.forEach(service::notify)));
-        callables.add(Executors.callable(() -> names.forEach(service::notify)));
-        callables.add(Executors.callable(() -> names.forEach(service::notify)));
+        callables.add(Executors.callable(() -> names.forEach(service::notifyUser)));
+        callables.add(Executors.callable(() -> names.forEach(service::notifyUser)));
+        callables.add(Executors.callable(() -> names.forEach(service::notifyUser)));
+        callables.add(Executors.callable(() -> names.forEach(service::notifyUser)));
 
         executorService.invokeAll(callables);
-        
-        System.out.println("STARTED!!!");
-        await().atMost(10, TimeUnit.SECONDS)
-                .pollInterval(3, TimeUnit.SECONDS)
-                .until(() -> {
-                    HobbyUser lastUser = repository.findByName(names.get(names.size() - 1));
-                    return lastUser.getTries() == 3 || lastUser.getStatus() == HobbyUser.NotificationStatus.SENT;
-                });
+
+        System.out.println("RESULTS:");
+        repository.findAll().forEach(System.out::println);
+
+        assertThat(repository.findByStatus(HobbyUser.NotificationStatus.READY)).isEmpty();
+    }
+
+    @Test
+    @DisplayName("4 threads notify 1 ready user at a time - skipping locked users by other threads")
+    void test_skip_locked() throws InterruptedException {
+        List<String> names = names(10);
+        names.forEach(n -> repository.save(HobbyUser.newInstance(n)));
+        ExecutorService executorService = Executors.newFixedThreadPool(4);
+
+        List<Callable<Object>> callables = new ArrayList<>();
+        for (int i = 0; i < 40; i++) {
+            callables.add(Executors.callable(() -> service.notifyNextUser()));
+        }
+        executorService.invokeAll(callables);
+
+        System.out.println("RESULTS:");
+        repository.findAll().forEach(System.out::println);
+
+        assertThat(repository.findByStatus(HobbyUser.NotificationStatus.READY)).isEmpty();
 
     }