Add example of batch reading from Kafka with context propagation (#311)

* add task functions to kotlin module

* add span column, change unique constraint

* правки по батчевой отправке и тесты

* rearrange imports

* Fix problem with propagator

* review fixes

* fix changed java version

* fix after review

* rewrite asserts without warnings

* add test for KafkaReadingService

* add new tracing test for kafka consumer

* Fix properties

* Refactor tests

---------

Co-authored-by: m.zharinova <m.zharinova@tbank.ru>
Co-authored-by: Ivan Vakhrushev <mfvanek@gmail.com>

Author: 3 people

build.gradle.kts

@@ -12,8 +12,8 @@ allprojects {
     version = "0.5.0"
 
     repositories {
-        mavenLocal()
         mavenCentral()
+        mavenLocal()
     }
 }
 

buildSrc/src/main/kotlin/sb-ot-demo.java-compile.gradle.kts

@@ -21,6 +21,7 @@ dependencies {
     if (osdetector.arch == "aarch_64") {
         testImplementation("io.netty:netty-all:4.1.104.Final")
     }
+    testImplementation("io.opentelemetry:opentelemetry-sdk-testing")
 }
 
 java {

db-migrations/build.gradle.kts

@@ -1,4 +1,15 @@
 plugins {
-    id("java")
+    id("java-library")
     id("sb-ot-demo.java-conventions")
+    id("io.freefair.lombok")
+}
+
+dependencies {
+    implementation(platform(project(":common-internal-bom")))
+    implementation(platform(libs.spring.boot.v3.dependencies))
+
+    implementation("io.micrometer:micrometer-tracing")
+    implementation("org.apache.kafka:kafka-clients")
+    implementation("org.slf4j:slf4j-api")
+    implementation("org.springframework:spring-jdbc")
 }

db-migrations/src/main/java/io/github/mfvanek/db/migrations/common/saver/DbSaver.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright (c) 2020-2025. Ivan Vakhrushev and others.
+ * https://github.com/mfvanek/spring-boot-open-telemetry-demo
+ *
+ * Licensed under the Apache License 2.0
+ */
+
+package io.github.mfvanek.db.migrations.common.saver;
+
+import io.micrometer.tracing.Span;
+import io.micrometer.tracing.Tracer;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.slf4j.MDC;
+import org.springframework.jdbc.core.simple.JdbcClient;
+
+import java.time.Clock;
+import java.time.LocalDateTime;
+import java.util.UUID;
+
+@Slf4j
+@RequiredArgsConstructor
+public class DbSaver {
+
+    private final String tenantName;
+    private final Tracer tracer;
+    private final Clock clock;
+    private final JdbcClient jdbcClient;
+
+    public void processSingleRecord(ConsumerRecord<UUID, String> record) {
+        try (MDC.MDCCloseable ignored = MDC.putCloseable("tenant.name", tenantName)) {
+            final Span currentSpan = tracer.currentSpan();
+            final String traceId = currentSpan != null ? currentSpan.context().traceId() : "";
+            final String spanId = currentSpan != null ? currentSpan.context().spanId() : "";
+            log.info("Received record: {} with traceId {} spanId {}", record.value(), traceId, spanId);
+            jdbcClient.sql("""
+                    insert into otel_demo.storage(message, trace_id, span_id, created_at)
+                    values(:msg, :traceId, :currentSpan, :createdAt);""")
+                .param("msg", record.value())
+                .param("traceId", traceId)
+                .param("currentSpan", spanId)
+                .param("createdAt", LocalDateTime.now(clock))
+                .update();
+        }
+    }
+}

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

@@ -3,3 +3,7 @@ databaseChangeLog:
         file: db/changelog/sql/schema.sql
     - include:
         file: db/changelog/sql/storage.sql
+    -   include:
+            file: db/changelog/sql/add_span_column.sql
+    -   include:
+            file: db/changelog/sql/set_span_and_trace_unique.sql

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

@@ -0,0 +1,7 @@
+--liquibase formatted sql
+
+--changeset marina.zharinova:2025.08.31:add span column
+alter table otel_demo.storage add column span_id text;
+
+--changeset marina.zharinova:2025.08.31:comment on span_id
+comment on column otel_demo.storage.span_id is 'SpanId of operation';

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

@@ -0,0 +1,7 @@
+--liquibase formatted sql
+
+--changeset ivan.vakhrushev:2025.08.31:remove unique from trace_id
+alter table otel_demo.storage drop constraint storage_trace_id_key;
+
+--changeset marina.zharinova:2025.08.31:add constraint on trace_id with span_id
+alter table otel_demo.storage add constraint trace_span_unique unique(trace_id, span_id);

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

@@ -5,7 +5,7 @@ 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,
+    trace_id text not null unique,
     created_at timestamptz not null
 );
 

spring-boot-3-demo-app-kotlin/src/main/kotlin/io/github/mfvanek/spring/boot3/kotlin/test/config/DbConfig.kt

@@ -0,0 +1,30 @@
+/*
+ * Copyright (c) 2020-2025. Ivan Vakhrushev and others.
+ * https://github.com/mfvanek/spring-boot-open-telemetry-demo
+ *
+ * Licensed under the Apache License 2.0
+ */
+
+package io.github.mfvanek.spring.boot3.kotlin.test.config
+
+import io.github.mfvanek.db.migrations.common.saver.DbSaver
+import io.micrometer.tracing.Tracer
+import org.springframework.beans.factory.annotation.Value
+import org.springframework.context.annotation.Bean
+import org.springframework.context.annotation.Configuration
+import org.springframework.jdbc.core.simple.JdbcClient
+import java.time.Clock
+
+@Configuration(proxyBeanMethods = false)
+class DbConfig {
+
+    @Bean
+    fun dbSaver(
+        @Value("\${app.tenant.name}") tenantName: String,
+        tracer: Tracer,
+        clock: Clock,
+        jdbcClient: JdbcClient
+    ): DbSaver {
+        return DbSaver(tenantName, tracer, clock, jdbcClient)
+    }
+}

spring-boot-3-demo-app-kotlin/src/main/kotlin/io/github/mfvanek/spring/boot3/kotlin/test/controllers/TimeController.kt

@@ -25,17 +25,22 @@ class TimeController(
     private val kafkaSendingService: KafkaSendingService,
     private val publicApiService: PublicApiService
 ) {
-
+    // http://localhost:8090/current-time
     @GetMapping(path = ["/current-time"])
     fun getNow(): LocalDateTime {
         logger.trace { "tracer $tracer" }
         val traceId = tracer.currentSpan()?.context()?.traceId()
         logger.info { "Called method getNow. TraceId = $traceId" }
         val nowFromRemote = publicApiService.getZonedTime()
         val now = nowFromRemote ?: LocalDateTime.now(clock)
-        kafkaSendingService.sendNotification("Current time = $now")
+        val message = "Current time = $now"
+        kafkaSendingService.sendNotification(message)
             .thenRun { logger.info { "Awaiting acknowledgement from Kafka" } }
             .get()
+
+        kafkaSendingService.sendNotificationToOtherTopic(message)
+            .thenRun { logger.info { "Awaiting acknowledgement from Kafka with batch" } }
+            .get()
         return now
     }
 }