Read from Kafka and write to PostgreSQL (#149)

* Add KafkaReadingService

* Add PostgreSQL

* Add test

* Fix compose file

* Add Spring Boot 2 example

Author: mfvanek

.editorconfig

@@ -4,12 +4,12 @@ end_of_line = crlf
 indent_size = 4
 indent_style = space
 insert_final_newline = true
-max_line_length = 130
+max_line_length = 199
 tab_width = 4
-ij_continuation_indent_size = 8
+ij_continuation_indent_size = 4
 
 [*.java]
-ij_java_imports_layout = *,|,javax.**,java.**,|,$*
+ij_java_imports_layout = *,|,java.**,javax.**,|,$*
 ij_java_align_multiline_parameters = true
 ij_java_blank_lines_after_class_header = 1
 ij_java_blank_lines_after_imports = 1
@@ -32,7 +32,7 @@ ij_java_keep_simple_lambdas_in_one_line = true
 ij_java_keep_simple_methods_in_one_line = true
 ij_java_names_count_to_use_import_on_demand = 101
 ij_java_class_count_to_use_import_on_demand = 101
-ij_java_packages_to_use_import_on_demand = javax.swing.*
+ij_java_packages_to_use_import_on_demand = java.awt.*, javax.swing.*
 ij_java_prefer_longer_names = true
 ij_java_space_after_closing_angle_bracket_in_type_argument = false
 ij_java_space_after_colon = true

common-internal-bom/build.gradle.kts

@@ -12,4 +12,11 @@ dependencies {
     api(platform("org.assertj:assertj-bom:3.26.3"))
     api(platform("org.testcontainers:testcontainers-bom:1.20.0"))
     api(platform("org.junit:junit-bom:5.10.3"))
+    api(platform("io.github.mfvanek:pg-index-health-bom:0.13.0"))
+
+    constraints {
+        api("org.liquibase:liquibase-core:4.28.0")
+        api("com.github.blagerweij:liquibase-sessionlock:1.6.9")
+        api("org.awaitility:awaitility:4.2.1")
+    }
 }

db-migrations/build.gradle.kts

@@ -0,0 +1,4 @@
+plugins {
+    id("java")
+    id("sb-ot-demo.java-conventions")
+}

db-migrations/src/main/resources/db/changelog/db.changelog-master.yaml

@@ -0,0 +1,5 @@
+databaseChangeLog:
+    - include:
+        file: db/changelog/sql/schema.sql
+    - include:
+        file: db/changelog/sql/storage.sql

db-migrations/src/main/resources/db/changelog/sql/schema.sql

@@ -0,0 +1,4 @@
+--liquibase formatted sql
+
+--changeset ivan.vakhrushev:2024.08.04:create.schema
+create schema if not exists otel_demo;

db-migrations/src/main/resources/db/changelog/sql/storage.sql

@@ -0,0 +1,17 @@
+--liquibase formatted sql
+
+--changeset ivan.vakhrushev:2024.08.04:storage.table
+create table if not exists otel_demo.storage
+(
+    id bigint generated always as identity,
+    message text not null,
+    trace_id varchar(64) not null unique,
+    created_at timestamptz not null
+);
+
+--changeset ivan.vakhrushev:2024.08.04:storage.comments.on.table.and.columns
+comment on table otel_demo.storage is 'Information about messages from Kafka';
+comment on column otel_demo.storage.id is 'Unique identifier of the record in the current table';
+comment on column otel_demo.storage.message is 'Message from Kafka';
+comment on column otel_demo.storage.trace_id is 'Unique traceId of operation';
+comment on column otel_demo.storage.created_at is 'Date and time of operation';

docker/docker-compose-base.yml

@@ -84,5 +84,24 @@ services:
         networks:
             - jaeger-example
 
+    postgres:
+        container_name: postgres
+        image: postgres:16.2
+        shm_size: "2gb"
+        environment:
+            POSTGRES_DB: "otel_demo_db"
+            POSTGRES_USER: "otel_demo_user"
+            POSTGRES_PASSWORD: "otel_demo_password"
+            PGDATA: "/var/lib/postgresql/data/pgdata"
+        volumes:
+            - otel_demo_db_data:/var/lib/postgresql/data
+        ports:
+            - "6432:5432"
+        networks:
+            - jaeger-example
+
 networks:
     jaeger-example:
+
+volumes:
+    otel_demo_db_data:

settings.gradle.kts

@@ -2,3 +2,4 @@ rootProject.name = "spring-boot-open-telemetry-demo"
 include("spring-boot-3-demo-app")
 include("common-internal-bom")
 include("spring-boot-2-demo-app")
+include("db-migrations")

spring-boot-2-demo-app/build.gradle.kts

@@ -24,10 +24,22 @@ dependencies {
     }
     implementation("org.springframework.cloud:spring-cloud-sleuth-otel-autoconfigure")
 
+    implementation("org.springframework.boot:spring-boot-starter-jdbc")
+    implementation("org.postgresql:postgresql")
+    implementation("com.zaxxer:HikariCP")
+    implementation(project(":db-migrations"))
+    implementation("org.liquibase:liquibase-core")
+    implementation("com.github.blagerweij:liquibase-sessionlock")
+    implementation("net.ttddyy:datasource-proxy:1.9") {
+        because("https://github.com/jdbc-observations/datasource-proxy/issues/111")
+    }
+
     testImplementation("org.springframework.boot:spring-boot-starter-test")
     testImplementation("org.springframework.boot:spring-boot-starter-webflux")
+    testImplementation("org.testcontainers:postgresql")
     testImplementation("org.testcontainers:kafka")
     testImplementation("org.springframework.kafka:spring-kafka-test")
+    testImplementation("org.awaitility:awaitility")
 }
 
 springBoot {

spring-boot-2-demo-app/src/main/java/io/github/mfvanek/spring/boot2/test/service/KafkaReadingService.java

@@ -0,0 +1,41 @@
+package io.github.mfvanek.spring.boot2.test.service;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.springframework.cloud.sleuth.Span;
+import org.springframework.cloud.sleuth.Tracer;
+import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.support.Acknowledgment;
+import org.springframework.stereotype.Service;
+
+import java.time.Clock;
+import java.time.LocalDateTime;
+import java.util.Map;
+import java.util.UUID;
+
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class KafkaReadingService {
+
+    private final Tracer tracer;
+    private final Clock clock;
+    private final NamedParameterJdbcTemplate jdbcTemplate;
+
+    @KafkaListener(topics = "${spring.kafka.template.default-topic}")
+    public void listen(ConsumerRecord<UUID, String> message, Acknowledgment ack) {
+        final Span currentSpan = tracer.currentSpan();
+        final String traceId = currentSpan != null ? currentSpan.context().traceId() : "";
+        log.info("Received record: {} with traceId {}", message.value(), traceId);
+        jdbcTemplate.update("insert into otel_demo.storage(message, trace_id, created_at) values(:msg, :traceId, :createdAt);",
+                Map.ofEntries(
+                        Map.entry("msg", message.value()),
+                        Map.entry("traceId", traceId),
+                        Map.entry("createdAt", LocalDateTime.now(clock))
+                )
+        );
+        ack.acknowledge();
+    }
+}