diff --git a/AGENTS.md b/AGENTS.md
new file mode 100644
index 0000000..c60e6e2
--- /dev/null
+++ b/AGENTS.md
@@ -0,0 +1,209 @@
+# AGENTS.md – Guidance for Automated Coding Agents
+
+---
+## Table of Contents
+
+### Workflow Guidelines
+- **Clarify Requirements First**: Before starting any change, ensure the requirement is fully understood. Ask the user follow‑up questions if anything is ambiguous. Multiple clarification questions are allowed.
+- **Obtain Acceptance of Action Plan**: Always draft a **detailed, verbose implementation plan** (steps, design decisions, test strategy) and present it to the user. Do **not** begin any code changes until the user explicitly approves the plan.
+- **Test‑Driven Development (TDD) Cycle**:
+  1. Write comprehensive unit/integration tests following project conventions.
+  2. Implement the minimum production code needed to satisfy the tests.
+  3. Refactor while keeping all tests green.
+  4. Repeat – add more tests, expand implementation, ensure clean, modular code.
+- **Never Cut Corners**: Do not add dummy implementations merely to make tests pass. Follow SOLID, Open‑Closed, KISS, and hexagonal architecture principles.
+- **Modular Design**: Keep new code modular, well‑encapsulated, and adhere to the existing layered architecture.
+- **Documentation & Review**: Update relevant documentation, comments, and ensure code quality matches a senior developer’s standards.
+
+
+1. [Project Overview](#project-overview)
+2. [Build, Lint & Test Commands](#build-lint--test-commands)
+3. [Running a Single Test](#running-a-single-test)
+4. [Code Style & Conventions](#code-style--conventions)
+   - Imports
+   - Formatting & Layout
+   - Types & Nullability
+   - Naming Conventions
+   - Error Handling
+   - Reactive & Coroutine Practices
+   - Architecture & Design Patterns
+5. **Testing Guidelines**
+6. **Performance & Back‑pressure**
+7. **Commit & PR Best Practices**
+8. **Additional Resources**
+---
+## Project Overview
+- **Preference**: Prefer Kotlin coroutines over Project Reactor for asynchronous handling. Agents should convert reactive streams to coroutines where feasible, while maintaining back‑pressure support.
+- **Language**: Kotlin 2.2.21 (JVM target)
+- **Framework**: Spring Boot 4.x (reactive stack)
+- **Database**: R2DBC (PostgreSQL) with Liquibase migrations
+- **Messaging**: Kafka
+- **Testing**: JUnit 5, Kotest, MockK, Spring Test, WebTestClient, Testcontainers
+- **Build Tool**: Gradle 8 (Groovy DSL)
+- **Containerisation**: Docker & Docker‑Compose (Postgres)
+
+The codebase follows a **hexagonal / clean‑architecture** flavour – core domain logic lives in `src/main/kotlin/com/softeno/template/app/...` and adapters (web, db, kafka) are separate packages.
+
+---
+## Build, Lint & Test Commands
+| Goal | Command | Description |
+|------|---------|-------------|
+| **Compile** | `./gradlew clean build` | Full clean compile, runs tests, creates JAR. |
+| **Run Application** | `./gradlew bootRun` | Starts the Spring Boot app (uses `application.properties`). |
+| **Package JAR** | `./gradlew bootJar` | Produces an executable JAR in `build/libs`. |
+| **Run Checks Only** | `./gradlew check` | Executes linting, static analysis, and tests without building the JAR. |
+| **Static Analysis** | `./gradlew detektMain detektTest` | Runs Detekt (Kotlin linter) on source and test code. |
+| **Formatting** | `./gradlew spotlessApply` | Applies Spotless formatting (Kotlin + Gradle). |
+| **Run All Tests** | `./gradlew test` | Executes unit + integration tests. |
+| **Run Integration Tests Only** | `./gradlew integrationTest` *(if defined)* | Use custom source set; otherwise filter with tags. |
+| **Run Tests with Coverage** | `./gradlew jacocoTestReport` | Generates a JaCoCo coverage report under `build/reports/jacoco`. |
+| **Run a Single Test** | See section below. |
+
+---
+## Running a Single Test
+Kotlin test classes are discovered by the `test` task. To run a single test method or class, use the Gradle `--tests` filter:
+
+```bash
+# Run a single test class
+./gradlew test --tests "com.softeno.template.app.permission.service.PermissionServiceTest"
+
+# Run a single test method (full qualified name)
+./gradlew test --tests "com.softeno.template.app.permission.service.PermissionServiceTest.shouldPersistAndRetrievePermission"
+```
+
+*Tip for agents*: When a failing test is identified, re‑run it with the same filter to verify the fix quickly.
+
+---
+## Code Style & Conventions
+The following rules are **non‑negotiable** for any automated agent. Violations will cause the CI lint step to fail.
+
+### 1. Imports
+- **Alphabetical order**, grouped by: `kotlin.*`, `java.*`, `org.*`, `com.*`.
+- One blank line between groups.
+- Use **single‑type imports** (`import java.util.UUID`) – avoid wildcard `*` imports.
+- Remove unused imports – Detekt’s `UnusedImports` rule enforces this.
+
+```kotlin
+import com.softeno.template.app.permission.api.PermissionController
+import com.softeno.template.app.permission.service.PermissionService
+import org.springframework.web.bind.annotation.GetMapping
+import org.springframework.web.bind.annotation.RequestMapping
+```
+
+### 2. Formatting & Layout
+- **Indentation**: 4 spaces, no tabs.
+- **Line length**: max 120 characters.
+- **Braces**: K&R style – opening brace on same line.
+- **Trailing commas** in multi‑line collections/parameter lists (Kotlin 1.9 supports it).
+- **Blank lines**: separate logical sections (e.g., imports, class header, properties, functions).
+- **Spotless** is the single source of truth – run `./gradlew spotlessApply` before committing.
+
+### 3. Types & Nullability
+- Prefer **non‑nullable** types. Use `?` only when the value can legitimately be absent.
+- When a nullable value is required, employ **`requireNotNull`** or **`?.let {}`** to avoid NPEs.
+- Leverage Kotlin’s **type‑aliases** for complex reactive types, e.g.: `typealias Flux<T> = reactor.core.publisher.Flux<T>` for readability.
+- Use **value classes** (`@JvmInline`) for simple wrappers (e.g., `PermissionId`).
+
+### 4. Naming Conventions
+| Element | Convention |
+|---------|-------------|
+| Packages | lower‑case, dot‑separated (`com.softeno.template.app.permission`) |
+| Classes / Interfaces | PascalCase (`PermissionService`, `PermissionRepository`) |
+| Enums | PascalCase, singular (`PermissionStatus`) |
+| Functions | camelCase, verb‑first (`createPermission`, `findById`) |
+| Properties | camelCase, descriptive (`permissionId`, `createdAt`) |
+| Constants | Upper‑snake (`MAX_RETRY_ATTEMPTS`) |
+| Test classes | `<ClassUnderTest>Test` (`PermissionServiceTest`) |
+| Test methods | descriptive, `should<Behavior>` (`shouldPersistAndRetrievePermission`) |
+
+### 5. Error Handling
+- **Never swallow exceptions**; re‑throw or map to a domain‑specific exception.
+- Use **Spring’s `ResponseStatusException`** for HTTP errors, with appropriate status codes.
+- For reactive pipelines, prefer **`onErrorMap`** / **`onErrorResume`** rather than `try/catch`.
+- Create a **`DomainException`** hierarchy for business‑logic errors.
+- Log errors at **WARN** or **ERROR** level with **structured logging** (`log.error("Failed to …", e)`).
+
+### 6. Reactive & Coroutine Practices
+- Prefer **Project Reactor** (`Mono`, `Flux`) for public APIs; use **Kotlin coroutines** (`suspend`, `Flow`) for internal services when it simplifies code.
+- **Back‑pressure**: never `block()` inside a reactive chain. Use `publishOn`/`subscribeOn` wisely.
+- Convert between Reactor and Coroutines with `reactor.kotlin.coroutines.awaitSingle`, `asFlux()`, etc.
+- Keep **side‑effects** at the end of the chain; avoid modifying shared state inside `map`.
+- **Timeouts**: apply `timeout(Duration)` on external calls (WebClient, R2DBC) to protect the system.
+
+### 7. Architecture & Design Patterns
+- Follow **Hexagonal (Ports & Adapters)**: core domain (`service`, `mapper`, `model`) has no Spring annotations; adapters (`controller`, `repository`, `kafka`) implement interfaces defined in the domain.
+- Use **Dependency Injection** (constructor injection only). No field injection.
+- Apply **SOLID**:
+  - **Single Responsibility** – each class does one thing.
+  - **Open/Closed** – extend behavior via interfaces, avoid modifying existing classes.
+  - **Liskov Substitution** – keep contracts consistent.
+  - **Interface Segregation** – small, focused interfaces.
+  - **Dependency Inversion** – depend on abstractions.
+- **KISS** – avoid over‑engineering; introduce a pattern only when the problem demands it.
+- Common OO patterns used:
+  - **Factory** (for creating domain objects from DTOs).
+  - **Adapter** (e.g., Kafka listener to service).
+  - **Decorator** (adding logging or metrics to repositories).
+- **Do NOT** introduce a full‑blown DDD layer unless a clear bounded context emerges.
+
+---
+## Testing Guidelines
+1. **Unit Tests**
+   - Focus on **pure business logic** – repository & controller layers are covered by integration tests.
+   - Use **MockK** for mocks/stubs; verify interactions only when they affect behavior.
+   - Avoid testing getters/setters or trivial data classes.
+   - Use **`@Nested`** classes (JUnit) for grouping scenarios.
+   - Keep test method names expressive (`shouldReturnPermissionWhenExists`).
+2. **Integration Tests**
+   - Spin up **Testcontainers** for PostgreSQL & Kafka where needed.
+   - Use **`@SpringBootTest`** with `webEnvironment = RANDOM_PORT` and `WebTestClient` for end‑to‑end HTTP flow.
+   - Test **error paths**, **validation**, **security** (401/403), and **back‑pressure** behaviour.
+   - Do **not** mock Spring beans; rely on the actual context to verify wiring.
+3. **Performance / Reactive Tests**
+   - Use **`StepVerifier`** with `virtualTime()` for time‑based operators.
+   - Verify **`request(n)`** behaviour to ensure back‑pressure compliance.
+4. **Coverage**
+   - Target **~80%** line coverage; focus on critical paths.
+   - Exclude generated code and third‑party libraries from coverage metrics.
+5. **Test Data**
+   - Use **Builder pattern** or Kotlin `apply {}` blocks for fixture creation.
+   - Keep fixtures **immutable**; share via companion objects when appropriate.
+6. **CI Recommendations**
+   - `./gradlew test` must succeed in CI.
+   - `./gradlew detektMain detektTest spotlessCheck` must pass.
+   - Failing tests abort the PR.
+
+---
+## Performance & Back‑pressure
+- **Reactive DB Calls**: always use **R2DBC** non‑blocking APIs; never call `blocking` JDBC.
+- **Kafka**: configure **`max.poll.records`** and **`reactor.kafka.receiver.KafkaReceiver`** to respect demand.
+- **WebClient**: enable **`connectionPool`** and **`readTimeout`**; reuse the same bean.
+- **Circuit Breaker**: use Resilience4j with sensible defaults (`failureRateThreshold=50`, `waitDurationInOpenState=30s`).
+- **Metric Collection**: expose **Micrometer** counters for request latency and throughput.
+- **Avoid heavy object allocation** in hot paths – reuse immutable data classes.
+- **Profiling**: run `./gradlew bootRun --args='--spring.profiles.active=dev'` and attach a profiler (YourKit, VisualVM) for bottleneck analysis.
+
+---
+## Commit & PR Best Practices
+1. **Atomic Commits** – one logical change per commit.
+2. **Commit Message** – `{type}(scope): short description`
+   - `type` = `feat`, `fix`, `refactor`, `test`, `chore`.
+   - Example: `feat(permission): add bulk‑create endpoint`.
+3. **PR Title** – concise, matches first line of commit.
+4. **PR Description** – include **What**, **Why**, **How** and **Testing steps**.
+5. **No Secrets** – never commit `.env`, API keys, passwords.
+6. **Never commit or push without explicit user approval** – agents must obtain clear user confirmation before executing any `git commit` or `git push` commands.
+7. **Rebase** – keep a linear history; avoid merge commits.
+8. **CI Checks** – ensure `./gradlew check` passes before merging.
+
+---
+## Additional Resources
+- **Kotlin Coding Conventions** – https://kotlinlang.org/docs/coding-conventions.html
+- **Spring Reactive Documentation** – https://docs.spring.io/spring-framework/reference/web/reactive.html
+- **Detekt Ruleset** – https://detekt.dev/docs/rules/
+- **Hexagonal Architecture** – https://alistair.cockburn.us/hexagonal-architecture/
+- **Resilience4j** – https://resilience4j.readme.io/
+- **Testcontainers** – https://www.testcontainers.org/
+- **Spotless Kotlin Plugin** – https://github.com/diffplug/spotless/tree/main/plugin-gradle#kotlin
+---
+*Generated by the AI assistant to guide automated agents working on the `spring-reactive-r2dbc-template` repository. Follow these rules strictly to keep the codebase clean, performant, and maintainable.*
diff --git a/build.gradle b/build.gradle
index ca00112..e24f14f 100644
--- a/build.gradle
+++ b/build.gradle
@@ -5,7 +5,7 @@ buildscript {
 }
 
 plugins {
-	id 'org.springframework.boot' version '3.5.10'
+	id 'org.springframework.boot' version '4.0.2'
 	id 'io.spring.dependency-management' version '1.1.7'
 	id 'org.jetbrains.kotlin.jvm' version '2.2.21'
 	id 'org.jetbrains.kotlin.plugin.spring' version '2.2.21'
@@ -14,6 +14,31 @@ plugins {
 	id 'groovy'
 }
 
+ext {
+	set('springCloudVersion', '2025.1.1')
+	set('testcontainersVersion', '2.0.3')
+	set('springBootVersion', '4.0.2')
+	set('wiremockVersion', '3.13.2')
+	set('mockkVersion', '1.14.9')
+	set('springmockkVersion', '4.0.2')
+	set('liquibaseVersion', '4.33.0')
+	set('postgresqlVersion', '42.7.10')
+	set('logbackVersion', '1.5.30')
+	set('picocliVersion', '4.7.7')
+	set('springdocVersion', '3.0.1')
+	set('logbackAppenderVersion', '2.25.0-alpha')
+	set('reactorKafkaVersion', '1.3.25')
+	set('r2dbcPostgresqlVersion', '1.1.1.RELEASE')
+	set('postgresqlVersion', '42.7.10')
+	set('junitPlatformLauncherVersion', '6.0.3')
+	set('micrometerRegistryPrometheusVersion', '1.16.3')
+	set('springSecurityTestVersion', '7.0.3')
+	set('springResilience4jVersion', '5.0.1')
+	set('micrometerTracingBridgeOtelVersion', '1.6.3')
+	set('micrometerContextPropagationVersion', '1.2.1')
+	set('nettyResolverDnsNativeMacosVersion', '4.2.3.Final')
+}
+
 group = 'com.softeno'
 version = '0.0.1-SNAPSHOT'
 
@@ -33,95 +58,96 @@ repositories {
 	mavenCentral()
 }
 
-ext {
-	set('springCloudVersion', '2025.0.1')
-	set('testcontainersVersion', '2.0.3')
-}
-
 dependencies {
-	implementation 'org.springframework.boot:spring-boot-starter-data-r2dbc'
-	implementation 'org.springframework.boot:spring-boot-starter-webflux'
-	implementation 'org.springframework:spring-jdbc'
-	runtimeOnly 'org.postgresql:r2dbc-postgresql'
-	runtimeOnly 'org.postgresql:postgresql'
+	implementation "org.springframework.boot:spring-boot-starter-data-r2dbc:${springBootVersion}"
+	implementation "org.springframework.boot:spring-boot-starter-webflux:${springBootVersion}"
+	implementation "org.springframework.boot:spring-boot-starter-webclient:${springBootVersion}"
+	implementation "org.springframework.boot:spring-boot-starter-jackson:${springBootVersion}"
+	implementation "org.springframework.boot:spring-boot-r2dbc:${springBootVersion}"
+	runtimeOnly "org.postgresql:r2dbc-postgresql:${r2dbcPostgresqlVersion}"
+	runtimeOnly "org.postgresql:postgresql:${postgresqlVersion}"
+	implementation "org.springframework.boot:spring-boot-starter-validation:${springBootVersion}"
+	developmentOnly "org.springframework.boot:spring-boot-devtools:${springBootVersion}"
+	annotationProcessor "org.springframework.boot:spring-boot-configuration-processor:${springBootVersion}"
 	implementation 'io.projectreactor.kotlin:reactor-kotlin-extensions'
 	implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-reactor'
 	implementation 'com.fasterxml.jackson.module:jackson-module-kotlin'
 	implementation 'org.jetbrains.kotlin:kotlin-reflect'
-	implementation 'org.jetbrains.kotlin:kotlin-stdlib-jdk8'
-	implementation 'org.springframework.boot:spring-boot-starter-validation'
-	developmentOnly 'org.springframework.boot:spring-boot-devtools'
 	implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-slf4j'
-	annotationProcessor 'org.springframework.boot:spring-boot-configuration-processor'
 
 	// test utils
-	testImplementation 'org.springframework.boot:spring-boot-starter-test'
+	testImplementation "org.springframework.boot:spring-boot-starter-test:${springBootVersion}"
+	testImplementation "org.springframework.boot:spring-boot-starter-webflux-test:${springBootVersion}"
+	testImplementation "org.springframework.boot:spring-boot-starter-jackson-test:${springBootVersion}"
+	testImplementation "org.springframework.boot:spring-boot-starter-r2dbc-test:${springBootVersion}"
+	testImplementation "org.springframework.boot:spring-boot-starter-data-r2dbc-test:${springBootVersion}"
+	testImplementation "org.springframework.boot:spring-boot-starter-liquibase-test:${springBootVersion}"
+	testImplementation "org.springframework.boot:spring-boot-webtestclient:${springBootVersion}"
+	testRuntimeOnly "org.junit.platform:junit-platform-launcher:${junitPlatformLauncherVersion}"
+	testImplementation "io.mockk:mockk:${mockkVersion}"
+	testImplementation "com.ninja-squad:springmockk:${springmockkVersion}"
 	testImplementation 'org.jetbrains.kotlinx:kotlinx-coroutines-test'
 	testImplementation 'org.jetbrains.kotlin:kotlin-test-junit5'
 	testImplementation 'io.projectreactor:reactor-test'
-	testImplementation 'io.mockk:mockk:1.14.9'
-	testImplementation 'com.ninja-squad:springmockk:4.0.2'
-	testRuntimeOnly 'org.junit.platform:junit-platform-launcher'
 
 	// wiremock
-	testImplementation 'org.wiremock:wiremock-standalone:3.13.2'
+	testImplementation "org.wiremock:wiremock-standalone:${wiremockVersion}"
 
 	// monitoring
-	runtimeOnly 'io.micrometer:micrometer-registry-prometheus'
-	implementation 'org.springframework.boot:spring-boot-starter-actuator'
-
-	// elk
-	implementation 'net.logstash.logback:logstash-logback-encoder:8.1'
+	runtimeOnly "io.micrometer:micrometer-registry-prometheus:${micrometerRegistryPrometheusVersion}"
+	implementation "org.springframework.boot:spring-boot-starter-actuator:${springBootVersion}"
 
 	// testcontainers
-	testImplementation 'org.springframework.boot:spring-boot-testcontainers'
-	testImplementation 'org.testcontainers:junit-jupiter'
-	testImplementation 'org.testcontainers:postgresql'
-	testImplementation 'org.testcontainers:r2dbc'
+	testImplementation "org.springframework.boot:spring-boot-testcontainers:${springBootVersion}"
+	testImplementation "org.testcontainers:testcontainers-junit-jupiter:${testcontainersVersion}"
+	testImplementation "org.testcontainers:testcontainers-postgresql:${testcontainersVersion}"
+	testImplementation "org.testcontainers:testcontainers-r2dbc:${testcontainersVersion}"
+	testImplementation "org.testcontainers:testcontainers-kafka:${testcontainersVersion}"
 
 	// liquibase
-	implementation ('org.liquibase:liquibase-core:4.33.0') {
+	implementation "org.springframework.boot:spring-boot-starter-liquibase:${springBootVersion}"
+	implementation ("org.liquibase:liquibase-core:${liquibaseVersion}") {
 		exclude group: 'javax.xml.bind', module: 'jaxb-receiver'
 		exclude group: 'org.liquibase.ext', module: 'liquibase-hibernate5'
 	}
+	liquibaseRuntime "org.liquibase.ext:liquibase-postgresql:${liquibaseVersion}"
+	liquibaseRuntime "org.postgresql:postgresql:${postgresqlVersion}"
+	liquibaseRuntime "ch.qos.logback:logback-core:${logbackVersion}"
+	liquibaseRuntime "ch.qos.logback:logback-classic:${logbackVersion}"
+	liquibaseRuntime "info.picocli:picocli:${picocliVersion}"
+	liquibaseRuntime "org.springframework.boot:spring-boot-starter-data-r2dbc:${springBootVersion}"
 	liquibaseRuntime 'org.jetbrains.kotlin:kotlin-stdlib-jdk8'
 	liquibaseRuntime 'org.jetbrains.kotlin:kotlin-reflect'
-	liquibaseRuntime 'org.liquibase.ext:liquibase-postgresql:4.33.0'
-	liquibaseRuntime 'org.postgresql:postgresql:42.7.10'
-	liquibaseRuntime 'ch.qos.logback:logback-core:1.5.30'
-	liquibaseRuntime 'ch.qos.logback:logback-classic:1.5.30'
-	liquibaseRuntime 'info.picocli:picocli:4.7.7'
-	liquibaseRuntime 'org.springframework.boot:spring-boot-starter-data-r2dbc:3.5.10'
 	liquibaseRuntime sourceSets.main.output
 
 	// spring security
-	implementation 'org.springframework.boot:spring-boot-starter-oauth2-client'
-	implementation 'org.springframework.boot:spring-boot-starter-oauth2-resource-server'
-	implementation 'org.springframework.boot:spring-boot-starter-security'
-	testImplementation 'org.springframework.security:spring-security-test'
+	implementation "org.springframework.boot:spring-boot-starter-oauth2-client:${springBootVersion}"
+	implementation "org.springframework.boot:spring-boot-starter-oauth2-resource-server:${springBootVersion}"
+	implementation "org.springframework.boot:spring-boot-starter-security:${springBootVersion}"
+	testImplementation "org.springframework.security:spring-security-test:${springSecurityTestVersion}"
 
 	// kafka
-	implementation 'org.springframework.kafka:spring-kafka'
-	implementation 'io.projectreactor.kafka:reactor-kafka'
-	testImplementation 'org.springframework.kafka:spring-kafka-test'
+	implementation "org.springframework.boot:spring-boot-starter-kafka:${springBootVersion}"
+	testImplementation "org.springframework.kafka:spring-kafka-test:${springBootVersion}"
 
 	// circuit breaker
-	implementation 'org.springframework.cloud:spring-cloud-starter-circuitbreaker-reactor-resilience4j'
+	implementation "org.springframework.cloud:spring-cloud-starter-circuitbreaker-reactor-resilience4j:${springResilience4jVersion}"
 
 	// macOs ARM only
 	if (osdetector.classifier == "osx-aarch_64") {
-		runtimeOnly("io.netty:netty-resolver-dns-native-macos:4.2.3.Final:${osdetector.classifier}")
+		runtimeOnly("io.netty:netty-resolver-dns-native-macos:${nettyResolverDnsNativeMacosVersion}:${osdetector.classifier}")
 	}
 
 	// springdoc
-	implementation 'org.springdoc:springdoc-openapi-starter-common:2.8.15'
-	implementation 'org.springdoc:springdoc-openapi-starter-webflux-ui:2.8.15'
-
-	// zipkin
-	implementation 'io.micrometer:micrometer-tracing-bridge-otel'
-	implementation 'io.opentelemetry:opentelemetry-exporter-zipkin'
-	implementation 'io.zipkin.reporter2:zipkin-sender-urlconnection'
-
+	implementation "org.springdoc:springdoc-openapi-starter-common:${springdocVersion}"
+	implementation "org.springdoc:springdoc-openapi-starter-webflux-ui:${springdocVersion}"
+
+	// opentelemetry
+	implementation "org.springframework.boot:spring-boot-starter-opentelemetry:${springBootVersion}"
+	testImplementation "org.springframework.boot:spring-boot-starter-opentelemetry-test:${springBootVersion}"
+	implementation "io.opentelemetry.instrumentation:opentelemetry-logback-appender-1.0:${logbackAppenderVersion}"
+	implementation "io.micrometer:micrometer-tracing-bridge-otel:${micrometerTracingBridgeOtelVersion}"
+	implementation "io.micrometer:context-propagation:${micrometerContextPropagationVersion}"
 }
 
 liquibase {
diff --git a/docker-compose.yml b/docker-compose.yml
new file mode 100644
index 0000000..e26935d
--- /dev/null
+++ b/docker-compose.yml
@@ -0,0 +1,134 @@
+version: "3.7"
+services:
+  kafka-broker:
+    image: bitnami/kafka:latest
+    container_name: kafka-broker
+    user: "${UID}"
+    hostname: kafka-broker
+    ports:
+      - "${IP:-0.0.0.0}:9094:9094"
+      - "${IP:-0.0.0.0}:9093:9093"
+      - "${IP:-0.0.0.0}:9092:9092"
+    environment:
+      KAFKA_CFG_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093,EXTERNAL://:9094
+      KAFKA_CFG_ADVERTISED_LISTENERS: PLAINTEXT://kafka-broker:9092,EXTERNAL://kafka-broker:9094
+      KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT,PLAINTEXT:PLAINTEXT
+      ALLOW_PLAINTEXT_LISTENER: yes
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka-broker:9092,EXTERNAL://kafka-broker:9094
+      KAFKA_BROKER_ID: 0
+      KAFKA_CFG_NODE_ID: 0
+      KAFKA_CFG_PROCESS_ROLES: controller,broker
+      KAFKA_CFG_CONTROLLER_LISTENER_NAMES: CONTROLLER
+      KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: 0@kafka-broker:9093
+    volumes:
+      - kafka-broker-data:/bitnami
+
+  init-kafka:
+    image: confluentinc/cp-kafka:latest
+    user: "${UID}"
+    depends_on:
+      - kafka-broker
+    entrypoint: [ '/bin/sh', '-c' ]
+    command: |
+      "
+      # blocks until kafka is reachable
+      kafka-topics --bootstrap-server kafka-broker:9092 --list
+      echo -e 'Creating kafka topics'
+      
+      kafka-topics --bootstrap-server kafka-broker:9092 --create --if-not-exists --topic sample_topic_2 --replication-factor 1 --partitions 1
+      kafka-topics --bootstrap-server kafka-broker:9092 --create --if-not-exists --topic keycloak-events --replication-factor 1 --partitions 1
+      kafka-topics --bootstrap-server kafka-broker:9092 --create --if-not-exists --topic keycloak-admin-events --replication-factor 1 --partitions 1
+      
+      echo -e 'Successfully created the following topics:'
+      kafka-topics --bootstrap-server kafka-broker:9092 --list
+      "
+
+  postgres-keycloak:
+    image: postgres:16.4
+    container_name: postgres-keycloak
+    user: "${UID}"
+    hostname: postgres-keycloak
+    volumes:
+      - postgres-data-keycloak:/var/lib/postgresql/data
+    environment:
+      POSTGRES_DB: keycloak
+      POSTGRES_USER: keycloak
+      POSTGRES_PASSWORD: keycloak
+    ports:
+      - "${IP:-0.0.0.0}:5433:5432"
+
+  keycloak:
+    image: xcodeassociated/keycloak-kafka:latest
+    user: "${UID}"
+    pull_policy: always
+    container_name: keycloak
+    hostname: keycloak
+    ports:
+      - "8090:8080"
+    depends_on:
+      - postgres-keycloak
+      - kafka-broker
+    environment:
+      KEYCLOAK_ADMIN: admin
+      KEYCLOAK_ADMIN_PASSWORD: admin
+      KAFKA_TOPIC: keycloak-events
+      KAFKA_CLIENT_ID: keycloak
+      KAFKA_BOOTSTRAP_SERVERS: kafka-broker:9092
+      KAFKA_EVENTS: "REGISTER,LOGIN,LOGOUT"
+      KAFKA_ADMIN_TOPIC: keycloak-admin-events
+      KC_DB_URL_HOST: postgres-keycloak
+      KC_DB_URL_DATABASE: keycloak
+      KC_DB_URL_PORT: 5432
+      KC_DB: postgres
+      KC_DB_USERNAME: keycloak
+      KC_DB_PASSWORD: keycloak
+      KC_DB_SCHEMA: public
+      KC_HOSTNAME_STRICT: false
+      KC_HTTP_ENABLED: true
+    command:
+      - start-dev
+    volumes:
+      - keycloak-data:/opt/keycloak/data/
+
+  postgres:
+    image: postgres:16.4
+    container_name: postgres
+    user: "${UID}"
+    hostname: postgres
+    volumes:
+      - postgres-data:/var/lib/postgresql/data
+    environment:
+      POSTGRES_DB: application
+      POSTGRES_USER: admin
+      POSTGRES_PASSWORD: admin
+    ports:
+      - "${IP:-0.0.0.0}:5432:5432"
+
+volumes:
+  kafka-broker-data:
+    driver: local
+    driver_opts:
+      o: bind
+      type: none
+      device: ./kafka+keycloak/volumes/kafka-broker
+
+  postgres-data-keycloak:
+    driver: local
+    driver_opts:
+      o: bind
+      type: none
+      device: ./kafka+keycloak/volumes/postgres
+
+  keycloak-data:
+    driver: local
+    driver_opts:
+      o: bind
+      type: none
+      device: ./kafka+keycloak/volumes/keycloak
+
+  postgres-data:
+    driver: local
+    driver_opts:
+      o: bind
+      type: none
+      device: ./postgres/volumes/postgres
\ No newline at end of file
diff --git a/http/user-app.http b/http/user-app.http
index 8221aef..b5d0a83 100644
--- a/http/user-app.http
+++ b/http/user-app.http
@@ -5,7 +5,7 @@ GET {{host}}/permissions?page=0&size=10&sort=id&direction=DESC
 Authorization: Bearer {{oauthToken}}
 
 ###
-GET {{host}}/permissions/count?search=test&createdFrom=1759429679915&createdBy=78652bda-8e80-432b-ba1b-47b4f7ee98e6
+GET {{host}}/permissions/count?search=test&createdFrom=1772265820594&createdBy=05dbf0b6-057b-44b4-aeaa-d83d662c5833
 Authorization: Bearer {{oauthToken}}
 
 ###
diff --git a/init_volumes.sh b/init_volumes.sh
new file mode 100755
index 0000000..3a29b6d
--- /dev/null
+++ b/init_volumes.sh
@@ -0,0 +1,21 @@
+#!/bin/bash
+
+set -euo pipefail
+
+DIRS=(
+  "kafka+keycloak/volumes/postgres"
+  "kafka+keycloak/volumes/keycloak"
+  "kafka+keycloak/volumes/kafka-broker"
+  "postgres/volumes/postgres"
+)
+
+for dir in "${DIRS[@]}"; do
+  if [ -d "$dir" ]; then
+    echo "Directory exists: $dir"
+  else
+    echo "Directory does not exist, creating: $dir"
+    mkdir -p "$dir"
+  fi
+done
+
+echo "done"
\ No newline at end of file
diff --git a/src/main/kotlin/com/softeno/template/SoftenoMvcJpaApp.kt b/src/main/kotlin/com/softeno/template/SoftenoMvcJpaApp.kt
index 30d2fd6..7e6cd06 100644
--- a/src/main/kotlin/com/softeno/template/SoftenoMvcJpaApp.kt
+++ b/src/main/kotlin/com/softeno/template/SoftenoMvcJpaApp.kt
@@ -4,7 +4,10 @@ import org.springframework.boot.autoconfigure.SpringBootApplication
 import org.springframework.boot.context.properties.ConfigurationPropertiesScan
 import org.springframework.boot.context.properties.EnableConfigurationProperties
 import org.springframework.boot.runApplication
+import org.springframework.context.annotation.ComponentScan
+import org.springframework.data.r2dbc.config.EnableR2dbcAuditing
 import org.springframework.data.r2dbc.repository.config.EnableR2dbcRepositories
+import org.springframework.kafka.annotation.EnableKafka
 import org.springframework.transaction.annotation.EnableTransactionManagement
 import reactor.core.publisher.Hooks
 
@@ -12,6 +15,9 @@ import reactor.core.publisher.Hooks
 @EnableTransactionManagement
 @EnableConfigurationProperties
 @EnableR2dbcRepositories
+@EnableR2dbcAuditing
+@EnableKafka
+@ComponentScan("com.softeno")
 @ConfigurationPropertiesScan("com.softeno")
 class SoftenoMvcJpaApp
 
@@ -19,4 +25,3 @@ fun main(args: Array<String>) {
 	Hooks.enableAutomaticContextPropagation()
 	runApplication<SoftenoMvcJpaApp>(*args)
 }
-
diff --git a/src/main/kotlin/com/softeno/template/app/common/PrincipalHandler.kt b/src/main/kotlin/com/softeno/template/app/common/PrincipalHandler.kt
index e18ed37..f166d20 100644
--- a/src/main/kotlin/com/softeno/template/app/common/PrincipalHandler.kt
+++ b/src/main/kotlin/com/softeno/template/app/common/PrincipalHandler.kt
@@ -12,7 +12,7 @@ interface PrincipalHandler {
     suspend fun showPrincipal(log: Log, monoPrincipal: Mono<Principal>){
         val principal = monoPrincipal.awaitSingleOrNull()
         log.info("principal: $principal, name: ${principal?.name}")
-        val authentication = ReactiveSecurityContextHolder.getContext().map { it.authentication }.awaitSingleOrNull()
+        val authentication = ReactiveSecurityContextHolder.getContext().mapNotNull { it.authentication }.awaitSingleOrNull()
         if (authentication != null) {
             val token = (authentication as JwtAuthenticationToken).token
             val userId = token.claims["sub"]
diff --git a/src/main/kotlin/com/softeno/template/app/config/opentelemetry/OpenTelemetryConfig.kt b/src/main/kotlin/com/softeno/template/app/config/opentelemetry/OpenTelemetryConfig.kt
new file mode 100644
index 0000000..36564c7
--- /dev/null
+++ b/src/main/kotlin/com/softeno/template/app/config/opentelemetry/OpenTelemetryConfig.kt
@@ -0,0 +1,28 @@
+package com.softeno.template.app.config.opentelemetry
+
+import io.opentelemetry.api.OpenTelemetry
+import io.opentelemetry.instrumentation.logback.appender.v1_0.OpenTelemetryAppender
+import org.springframework.beans.factory.InitializingBean
+import org.springframework.context.annotation.Bean
+import org.springframework.context.annotation.Configuration
+import org.springframework.core.task.support.ContextPropagatingTaskDecorator
+import org.springframework.stereotype.Component
+
+
+@Component
+class InstallOpenTelemetryAppender(
+    private val openTelemetry: OpenTelemetry
+) : InitializingBean {
+
+    override fun afterPropertiesSet() {
+        OpenTelemetryAppender.install(openTelemetry)
+    }
+}
+
+@Configuration(proxyBeanMethods = false)
+class ContextPropagationConfiguration {
+    @Bean
+    fun contextPropagatingTaskDecorator(): ContextPropagatingTaskDecorator {
+        return ContextPropagatingTaskDecorator()
+    }
+}
diff --git a/src/main/kotlin/com/softeno/template/app/config/security/ReactiveAuditorConfig.kt b/src/main/kotlin/com/softeno/template/app/config/security/ReactiveAuditorConfig.kt
index 2d69b1b..ea9d592 100644
--- a/src/main/kotlin/com/softeno/template/app/config/security/ReactiveAuditorConfig.kt
+++ b/src/main/kotlin/com/softeno/template/app/config/security/ReactiveAuditorConfig.kt
@@ -1,6 +1,5 @@
 package com.softeno.template.app.config.security
 
-import org.apache.commons.logging.LogFactory
 import org.springframework.context.annotation.Bean
 import org.springframework.context.annotation.Configuration
 import org.springframework.data.domain.ReactiveAuditorAware
@@ -13,12 +12,11 @@ import reactor.core.publisher.Mono
 
 
 class AuditorAwareImpl : ReactiveAuditorAware<String> {
-    private val log = LogFactory.getLog(javaClass)
 
     override fun getCurrentAuditor(): Mono<String> {
         return ReactiveSecurityContextHolder.getContext()
-            .map(SecurityContext::getAuthentication)
-            .map(Authentication::getPrincipal)
+            .mapNotNull(SecurityContext::getAuthentication)
+            .mapNotNull(Authentication::getPrincipal)
             .switchIfEmpty(Mono.just("anonymous"))
             .flatMap { principal ->
                 if (principal is Jwt) {
diff --git a/src/main/kotlin/com/softeno/template/app/kafka/KafkaKeycloakController.kt b/src/main/kotlin/com/softeno/template/app/kafka/KafkaKeycloakController.kt
index 20b9ddd..c6912dc 100644
--- a/src/main/kotlin/com/softeno/template/app/kafka/KafkaKeycloakController.kt
+++ b/src/main/kotlin/com/softeno/template/app/kafka/KafkaKeycloakController.kt
@@ -1,41 +1,52 @@
 package com.softeno.template.app.kafka
 
+import com.fasterxml.jackson.annotation.JsonCreator
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties
 import com.fasterxml.jackson.databind.JsonNode
-import com.fasterxml.jackson.databind.ObjectMapper
-import com.softeno.template.app.kafka.dto.KeycloakUserEvent
 import org.apache.commons.logging.LogFactory
 import org.apache.kafka.clients.consumer.ConsumerRecord
-import org.springframework.beans.factory.annotation.Qualifier
-import org.springframework.boot.CommandLineRunner
-import org.springframework.kafka.core.reactive.ReactiveKafkaConsumerTemplate
+import org.springframework.kafka.annotation.KafkaListener
 import org.springframework.stereotype.Controller
-import reactor.core.publisher.Flux
+import tools.jackson.databind.ObjectMapper
+
 
 @Controller
-class ReactiveKafkaKeycloakController(
-    @Qualifier(value = "kafkaKeycloakConsumerTemplate") private val reactiveKafkaConsumerTemplate: ReactiveKafkaConsumerTemplate<String, JsonNode>,
+class KafkaKeycloakController(
     private val objectMapper: ObjectMapper
-) : CommandLineRunner {
+) {
     private val log = LogFactory.getLog(javaClass)
 
-    private fun consumeKafkaMessage(): Flux<JsonNode> {
-        return reactiveKafkaConsumerTemplate
-            .receiveAutoAck()
-            .doOnNext { consumerRecord: ConsumerRecord<String, JsonNode> ->
-                log.debug("[kafka] rx keycloak: ConsumerRecord: key=${consumerRecord.key()}, value=${consumerRecord.value()} from topic=${consumerRecord.topic()}, offset=${consumerRecord.offset()}")
-            }
-            .map { obj: ConsumerRecord<String, JsonNode> -> obj.value() }
-            .doOnNext { message: JsonNode ->
-                val dto: KeycloakUserEvent = objectMapper.readValue(message.toString(), KeycloakUserEvent::class.java)
-                log.info("[kafka] rx keycloak: $dto")
-            }
-            .doOnError { throwable: Throwable ->
-                log.error("[kafka] keycloak: ${throwable.message}")
-            }
+    @KafkaListener(id = "\${spring.kafka.consumer.group-id}-keycloak", topics = ["\${com.softeno.kafka.keycloak}"])
+    suspend fun listen(record: ConsumerRecord<String, JsonNode>) {
+        log.debug("[kafka] rx keycloak raw: ${record.key()}: ${record.value()}")
+        val dto: KeycloakUserEvent = objectMapper.readValue(record.value().toString(), KeycloakUserEvent::class.java)
+        log.info("[kafka] rx keycloak mapped: $dto")
     }
+}
+
+enum class KeycloakEventType {
+    LOGIN,
+    LOGOUT,
+    REGISTER;
 
-    override fun run(vararg args: String) {
-        log.info("[kafka]: keycloak consumer starts")
-        consumeKafkaMessage().subscribe()
+    companion object {
+        @JsonCreator
+        fun byString(input: String): KeycloakEventType? {
+            return KeycloakEventType.entries.firstOrNull { it.name.equals(input, true) }
+        }
     }
-}
\ No newline at end of file
+}
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+data class KeycloakUserEvent(
+    val id: String,
+    val time: Long,
+    val type: KeycloakEventType,
+    val realmId: String,
+    val clientId: String?,
+    val userId: String,
+    val sessionId: String?,
+    val ipAddress: String,
+    val error: String?,
+    val details: Map<String, String>
+)
diff --git a/src/main/kotlin/com/softeno/template/app/kafka/KafkaSample.kt b/src/main/kotlin/com/softeno/template/app/kafka/KafkaSample.kt
new file mode 100644
index 0000000..dcb7ab3
--- /dev/null
+++ b/src/main/kotlin/com/softeno/template/app/kafka/KafkaSample.kt
@@ -0,0 +1,42 @@
+package com.softeno.template.app.kafka
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties
+import com.fasterxml.jackson.databind.JsonNode
+import org.apache.commons.logging.LogFactory
+import org.apache.kafka.clients.consumer.ConsumerRecord
+import org.springframework.boot.context.properties.ConfigurationProperties
+import org.springframework.kafka.annotation.KafkaListener
+import org.springframework.kafka.core.KafkaTemplate
+import org.springframework.stereotype.Component
+import org.springframework.stereotype.Controller
+
+@ConfigurationProperties(prefix = "com.softeno.kafka")
+data class KafkaApplicationProperties(val tx: String, val rx: String, val keycloak: String)
+
+@Controller
+class KafkaSampleController(
+    private val props: KafkaApplicationProperties
+) {
+    private val log = LogFactory.getLog(javaClass)
+
+    @KafkaListener(id = "\${spring.kafka.consumer.group-id}", topics = ["\${com.softeno.kafka.rx}"])
+    suspend fun listen(record: ConsumerRecord<String, JsonNode>) {
+        log.info("[kafka] rx (${props.rx}): ${record.key()}: ${record.value()}")
+    }
+}
+
+@Component
+class KafkaSampleProducer(
+    private val producer: KafkaTemplate<String, KafkaMessage>,
+    private val props: KafkaApplicationProperties
+) {
+    private val log = LogFactory.getLog(javaClass)
+
+    fun send(message: KafkaMessage) {
+        log.info("[kafka] tx (${props.tx}): $message")
+        producer.send(props.tx, message)
+    }
+}
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+data class KafkaMessage(val content: String)
\ No newline at end of file
diff --git a/src/main/kotlin/com/softeno/template/app/kafka/KafkaSampleHandler.kt b/src/main/kotlin/com/softeno/template/app/kafka/KafkaSampleHandler.kt
deleted file mode 100644
index 7dd7eea..0000000
--- a/src/main/kotlin/com/softeno/template/app/kafka/KafkaSampleHandler.kt
+++ /dev/null
@@ -1,90 +0,0 @@
-package com.softeno.template.app.kafka
-
-
-import com.fasterxml.jackson.databind.JsonNode
-import com.fasterxml.jackson.databind.ObjectMapper
-import com.softeno.template.app.kafka.config.KafkaApplicationProperties
-import com.softeno.template.app.kafka.dto.KafkaMessage
-import io.micrometer.observation.ObservationRegistry
-import io.micrometer.tracing.Span
-import io.micrometer.tracing.Tracer
-import org.apache.commons.lang3.RandomUtils
-import org.apache.commons.logging.LogFactory
-import org.apache.kafka.clients.consumer.ConsumerRecord
-import org.springframework.beans.factory.annotation.Qualifier
-import org.springframework.boot.CommandLineRunner
-import org.springframework.kafka.core.reactive.ReactiveKafkaConsumerTemplate
-import org.springframework.kafka.core.reactive.ReactiveKafkaProducerTemplate
-import org.springframework.stereotype.Controller
-import org.springframework.stereotype.Service
-import reactor.core.publisher.Flux
-import reactor.kafka.receiver.ReceiverOptions
-import reactor.kafka.sender.SenderResult
-
-
-@Controller
-class ReactiveKafkaSampleController(
-    @Qualifier(value = "kafkaSampleConsumerTemplate") private val reactiveKafkaConsumerTemplate: ReactiveKafkaConsumerTemplate<String, JsonNode>,
-    private val objectMapper: ObjectMapper,
-    private val tracer: Tracer,
-    private val observationRegistry: ObservationRegistry,
-    @Qualifier(value = "kafkaSampleOptions") private val kafkaReceiverOptions: ReceiverOptions<String, JsonNode>
-) : CommandLineRunner {
-    private val log = LogFactory.getLog(javaClass)
-
-    private fun consumeKafkaMessage(): Flux<JsonNode> {
-        return reactiveKafkaConsumerTemplate
-            .receiveAutoAck()
-            .doOnNext { consumerRecord: ConsumerRecord<String, JsonNode> ->
-                log.debug("[kafka] rx sample: ConsumerRecord: key=${consumerRecord.key()}, " +
-                        "value=${consumerRecord.value()} from topic=${consumerRecord.topic()}, " +
-                        "offset=${consumerRecord.offset()}, headers=${consumerRecord.headers()}")
-            }
-            .map { obj: ConsumerRecord<String, JsonNode> -> obj.value() }
-            .doOnNext { message: JsonNode ->
-                val kafkaMessage: KafkaMessage = objectMapper.readValue(message.toString(), KafkaMessage::class.java)
-
-                // todo: make better observation handling by reactive kafka, currently the zipkin does not show the traces properly
-                val contextWithCustomTraceId = tracer.traceContextBuilder()
-                    .traceId(kafkaMessage.traceId ?: RandomUtils.nextLong().toString())
-                    .spanId(tracer.nextSpan().name("kafka-consumer").context().spanId())
-                    .parentId(kafkaMessage.spanId ?: (tracer.currentSpan()?.context()?.spanId() ?: Span.NOOP.context().spanId()))
-                    .sampled(true)
-                    .build()
-
-                tracer.currentTraceContext().newScope(contextWithCustomTraceId).use {
-                    val span = tracer.nextSpan().name("kafka-consumer")
-                    tracer.withSpan(span.start()).use {
-                        log.info("[kafka] rx sample: $kafkaMessage")
-                        // additional processing ...
-                    }
-                    span.end()
-                }
-            }
-            .doOnError { throwable: Throwable ->
-                log.error("[kafka] sample: ${throwable.message}")
-            }
-    }
-
-    override fun run(vararg args: String) {
-        log.info("[kafka] sample: consumer starts")
-        consumeKafkaMessage().subscribe()
-    }
-}
-
-@Service
-class ReactiveKafkaSampleProducer(
-    @Qualifier(value = "kafkaSampleProducerTemplate") private val reactiveKafkaProducerTemplate: ReactiveKafkaProducerTemplate<String, KafkaMessage>,
-    private val props: KafkaApplicationProperties
-) {
-    private val log = LogFactory.getLog(javaClass)
-
-    fun send(message: KafkaMessage) {
-        log.info("[kafka] tx: topic: ${props.tx}, message: $message")
-        reactiveKafkaProducerTemplate.send(props.tx, message)
-            .doOnSuccess { senderResult: SenderResult<Void> ->
-                log.info("[kafka] tx ok, offset: ${senderResult.recordMetadata().offset()}")
-            }
-            .subscribe()
-    }
-}
\ No newline at end of file
diff --git a/src/main/kotlin/com/softeno/template/app/kafka/config/KafkaConfig.kt b/src/main/kotlin/com/softeno/template/app/kafka/config/KafkaConfig.kt
index 7229c16..f274ce1 100644
--- a/src/main/kotlin/com/softeno/template/app/kafka/config/KafkaConfig.kt
+++ b/src/main/kotlin/com/softeno/template/app/kafka/config/KafkaConfig.kt
@@ -1,84 +1,35 @@
 package com.softeno.template.app.kafka.config
 
 import com.fasterxml.jackson.databind.JsonNode
-import com.softeno.template.app.kafka.dto.KafkaMessage
+import com.softeno.template.app.kafka.KafkaMessage
 import io.micrometer.observation.ObservationRegistry
-import org.springframework.beans.factory.annotation.Qualifier
-import org.springframework.boot.autoconfigure.kafka.KafkaProperties
-import org.springframework.boot.context.properties.ConfigurationProperties
 import org.springframework.context.annotation.Bean
 import org.springframework.context.annotation.Configuration
-import org.springframework.kafka.core.reactive.ReactiveKafkaConsumerTemplate
-import org.springframework.kafka.core.reactive.ReactiveKafkaProducerTemplate
-import org.springframework.kafka.support.converter.ByteArrayJsonMessageConverter
-import org.springframework.kafka.support.converter.JsonMessageConverter
-import reactor.kafka.receiver.ReceiverOptions
-import reactor.kafka.receiver.observation.KafkaReceiverObservation
-import reactor.kafka.sender.SenderOptions
-import reactor.kafka.sender.observation.KafkaSenderObservation
-import java.util.*
+import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory
+import org.springframework.kafka.core.*
 
 
-@ConfigurationProperties(prefix = "com.softeno.kafka")
-data class KafkaApplicationProperties(val tx: String, val rx: String, val keycloak: String)
-
 @Configuration
-class JsonMessageConverterConfig {
+class KafkaConsumerConfig {
+
     @Bean
-    fun jsonMessageConverter(): JsonMessageConverter {
-        return ByteArrayJsonMessageConverter()
+    fun kafkaListenerContainerFactory(consumerFactory: ConsumerFactory<String, JsonNode>, registry: ObservationRegistry): ConcurrentKafkaListenerContainerFactory<String, JsonNode> {
+        val factory = ConcurrentKafkaListenerContainerFactory<String, JsonNode>()
+        factory.setConsumerFactory(consumerFactory)
+        factory.containerProperties.isObservationEnabled = true
+        factory.containerProperties.observationRegistry = registry
+        return factory
     }
 }
 
 @Configuration
-class ReactiveKafkaSampleConsumerConfig {
-    @Bean(value = ["kafkaSampleOptions"])
-    fun kafkaReceiverOptions(
-        kafkaProperties: KafkaProperties,
-        props: KafkaApplicationProperties,
-        observationRegistry: ObservationRegistry
-    ): ReceiverOptions<String, JsonNode> {
-        val basicReceiverOptions: ReceiverOptions<String, JsonNode> =
-            ReceiverOptions.create(kafkaProperties.buildConsumerProperties(null))
-        val basicReceiverOptionsWithObs = basicReceiverOptions
-            // todo: make better observation handling by reactive kafka, currently the zipkin does not show the traces properly
-            .withObservation(observationRegistry, KafkaReceiverObservation.DefaultKafkaReceiverObservationConvention()
-        )
-        return basicReceiverOptionsWithObs.subscription(Collections.singletonList(props.rx))
-    }
+class KafkaProducerConfig {
 
-    @Bean(value = ["kafkaSampleConsumerTemplate"])
-    fun reactiveKafkaConsumerTemplate(@Qualifier(value = "kafkaSampleOptions") kafkaReceiverOptions: ReceiverOptions<String, JsonNode>): ReactiveKafkaConsumerTemplate<String, JsonNode> {
-        return ReactiveKafkaConsumerTemplate(kafkaReceiverOptions)
-    }
-}
-
-@Configuration
-class ReactiveKafkaSampleProducerConfig {
-    @Bean(value = ["kafkaSampleProducerTemplate"])
-    fun reactiveKafkaProducerTemplate(properties: KafkaProperties, observationRegistry: ObservationRegistry): ReactiveKafkaProducerTemplate<String, KafkaMessage> {
-        val props = properties.buildProducerProperties(null)
-        val options = SenderOptions.create<String, KafkaMessage>(props)
-            // todo: make better observation handling by reactive kafka, currently the zipkin does not show the traces properly
-            .withObservation(observationRegistry, KafkaSenderObservation.DefaultKafkaSenderObservationConvention())
-        return ReactiveKafkaProducerTemplate<String, KafkaMessage>(options)
+    @Bean
+    fun kafkaTemplate(producerFactory: ProducerFactory<String, KafkaMessage>, registry: ObservationRegistry): KafkaTemplate<String, KafkaMessage> {
+        return KafkaTemplate(producerFactory).apply {
+            setObservationEnabled(true)
+            setObservationRegistry(registry)
+        }
     }
 }
-
-@Configuration
-class ReactiveKafkaKeycloakConsumerConfig {
-    @Bean(value = ["kafkaKeycloakOptions"])
-    fun kafkaReceiverOptions(
-        kafkaProperties: KafkaProperties,
-        props: KafkaApplicationProperties
-    ): ReceiverOptions<String, JsonNode> {
-        val basicReceiverOptions: ReceiverOptions<String, JsonNode> =
-            ReceiverOptions.create(kafkaProperties.buildConsumerProperties(null))
-        return basicReceiverOptions.subscription(Collections.singletonList(props.keycloak))
-    }
-
-    @Bean(value = ["kafkaKeycloakConsumerTemplate"])
-    fun reactiveKafkaConsumerTemplate(@Qualifier(value = "kafkaKeycloakOptions") kafkaReceiverOptions: ReceiverOptions<String, JsonNode>): ReactiveKafkaConsumerTemplate<String, JsonNode> {
-        return ReactiveKafkaConsumerTemplate(kafkaReceiverOptions)
-    }
-}
\ No newline at end of file
diff --git a/src/main/kotlin/com/softeno/template/app/kafka/dto/dto.kt b/src/main/kotlin/com/softeno/template/app/kafka/dto/dto.kt
deleted file mode 100644
index b0df3e9..0000000
--- a/src/main/kotlin/com/softeno/template/app/kafka/dto/dto.kt
+++ /dev/null
@@ -1,34 +0,0 @@
-package com.softeno.template.app.kafka.dto
-
-import com.fasterxml.jackson.annotation.JsonCreator
-import com.fasterxml.jackson.annotation.JsonIgnoreProperties
-
-@JsonIgnoreProperties(ignoreUnknown = true)
-data class KafkaMessage(val content: String, val traceId: String? = null, val spanId: String? = null)
-
-enum class KeycloakEventType {
-    LOGIN,
-    LOGOUT,
-    REGISTER;
-
-    companion object {
-        @JsonCreator
-        fun byString(input: String): KeycloakEventType? {
-            return KeycloakEventType.entries.firstOrNull { it.name.equals(input, true) }
-        }
-    }
-}
-
-@JsonIgnoreProperties(ignoreUnknown = true)
-data class KeycloakUserEvent(
-    val id: String,
-    val time: Long,
-    val type: KeycloakEventType,
-    val realmId: String,
-    val clientId: String?,
-    val userId: String,
-    val sessionId: String?,
-    val ipAddress: String,
-    val error: String?,
-    val details: Map<String, String>
-)
diff --git a/src/main/kotlin/com/softeno/template/app/permission/Permission.kt b/src/main/kotlin/com/softeno/template/app/permission/Permission.kt
index 6e80450..07c66a3 100644
--- a/src/main/kotlin/com/softeno/template/app/permission/Permission.kt
+++ b/src/main/kotlin/com/softeno/template/app/permission/Permission.kt
@@ -1,12 +1,17 @@
 package com.softeno.template.app.permission
 
+import jakarta.validation.constraints.NotBlank
+import jakarta.validation.constraints.NotNull
 import org.springframework.data.annotation.*
 import org.springframework.data.relational.core.mapping.Column
 import org.springframework.data.relational.core.mapping.Table
 import java.util.*
 
 open class BaseEntity(
+    @Transient
     open val uuid: UUID,
+
+    @Transient
     open var id: Long? = null,
 )
 
@@ -39,9 +44,11 @@ data class Permission(
     @Column(value = "version")
     val version: Long = 0,
 
+    @NotBlank
     @Column(value = "name")
     val name: String,
 
+    @NotNull
     @Column(value = "description")
     val description: String
 ) : BaseEntity(uuid) {
diff --git a/src/main/kotlin/com/softeno/template/app/permission/db/PermissionRepository.kt b/src/main/kotlin/com/softeno/template/app/permission/db/PermissionRepository.kt
index 3bc18cf..0878476 100644
--- a/src/main/kotlin/com/softeno/template/app/permission/db/PermissionRepository.kt
+++ b/src/main/kotlin/com/softeno/template/app/permission/db/PermissionRepository.kt
@@ -12,7 +12,6 @@ import org.springframework.data.relational.core.mapping.Table
 import org.springframework.data.repository.kotlin.CoroutineCrudRepository
 import org.springframework.data.repository.query.Param
 import org.springframework.r2dbc.core.DatabaseClient
-import org.springframework.r2dbc.core.bind
 import org.springframework.stereotype.Component
 import org.springframework.stereotype.Repository
 import reactor.core.publisher.Mono
@@ -126,7 +125,13 @@ class BatchPermissionRepositoryImpl<T : BaseEntity>(private val databaseClient:
         entities.forEachIndexed { idx, entity ->
             activeProps.forEach { (prop, _) ->
                 val value = (prop as KProperty1<T, *>).get(entity)
-                spec = spec.bind("${prop.name}$idx", value)
+                val paramName = "${prop.name}$idx"
+
+                spec = if (value != null) {
+                    spec.bind(paramName, value)
+                } else {
+                    spec.bindNull(paramName, Any::class.java)
+                }
             }
         }
 
@@ -140,28 +145,31 @@ class BatchPermissionRepositoryImpl<T : BaseEntity>(private val databaseClient:
             throw OperationNotPermittedException("Cannot update entities with null id")
 
         val sample = entities.first()
+
         val tableName = sample::class.findAnnotation<Table>()?.value
-            ?: throw OperationNotPermittedException("Missing @Table annotation on ${sample::class.simpleName}")
+            ?: throw OperationNotPermittedException(
+                "Missing @Table annotation on ${sample::class.simpleName}"
+            )
 
         val properties = sample::class.memberProperties
             .filter { it.javaField?.isAnnotationPresent(Column::class.java) == true }
             .map { prop -> prop to prop.javaField!!.getAnnotation(Column::class.java).value }
             .filter { (_, colName) -> colName != "id" && colName != "uuid" }
 
-        val setClauses = properties.mapNotNull { (prop, colName) ->
+        if (properties.isEmpty()) return emptyMap()
+
+        val setClauses = properties.map { (prop, colName) ->
             val cases = entities.mapIndexed { idx, entity ->
                 val value = (prop as KProperty1<T, *>).get(entity)
-                if (value != null) {
-                    "WHEN :uuid$idx THEN :${prop.name}$idx"
-                } else null
-            }.filterNotNull()
 
-            if (cases.isEmpty()) null
-            else "$colName = CASE uuid ${cases.joinToString(" ")} ELSE $colName END"
+                if (value != null)
+                    "WHEN :uuid$idx THEN :${prop.name}$idx"
+                else
+                    "WHEN :uuid$idx THEN NULL"
+            }
+            "$colName = CASE uuid ${cases.joinToString(" ")} ELSE $colName END"
         }
 
-        if (setClauses.isEmpty()) return emptyMap()
-
         val sql = """
         UPDATE $tableName
         SET ${setClauses.joinToString(", ")}
@@ -171,10 +179,12 @@ class BatchPermissionRepositoryImpl<T : BaseEntity>(private val databaseClient:
 
         var spec = databaseClient.sql(sql)
 
+        // Bind only what actually exists in SQL
         entities.forEachIndexed { idx, entity ->
             spec = spec.bind("uuid$idx", entity.uuid)
             properties.forEach { (prop, _) ->
                 val value = (prop as KProperty1<T, *>).get(entity)
+
                 if (value != null) {
                     spec = spec.bind("${prop.name}$idx", value)
                 }
diff --git a/src/main/kotlin/com/softeno/template/app/permission/service/PermissionService.kt b/src/main/kotlin/com/softeno/template/app/permission/service/PermissionService.kt
index 8905927..20c740d 100644
--- a/src/main/kotlin/com/softeno/template/app/permission/service/PermissionService.kt
+++ b/src/main/kotlin/com/softeno/template/app/permission/service/PermissionService.kt
@@ -1,7 +1,7 @@
 package com.softeno.template.app.permission.service
 
-import com.softeno.template.app.kafka.ReactiveKafkaSampleProducer
-import com.softeno.template.app.kafka.dto.KafkaMessage
+import com.softeno.template.app.kafka.KafkaMessage
+import com.softeno.template.app.kafka.KafkaSampleProducer
 import com.softeno.template.app.permission.Permission
 import com.softeno.template.app.permission.api.PermissionController
 import com.softeno.template.app.permission.db.PermissionRepository
@@ -14,7 +14,6 @@ import kotlinx.coroutines.reactor.awaitSingleOrNull
 import kotlinx.coroutines.slf4j.MDCContext
 import kotlinx.coroutines.withContext
 import org.apache.commons.logging.LogFactory
-import org.slf4j.MDC
 import org.springframework.data.domain.Pageable
 import org.springframework.data.r2dbc.core.R2dbcEntityTemplate
 import org.springframework.data.relational.core.query.Criteria.where
@@ -28,7 +27,7 @@ import java.time.Instant
 @Service
 class PermissionService(
     private val permissionRepository: PermissionRepository,
-    private val kafkaPublisher: ReactiveKafkaSampleProducer,
+    private val kafkaProducer: KafkaSampleProducer,
     private val template: R2dbcEntityTemplate
 ) {
     private val log = LogFactory.getLog(javaClass)
@@ -53,7 +52,7 @@ class PermissionService(
     suspend fun createPermission(permissionDto: PermissionDto): PermissionDto =
         withContext(MDCContext()) {
             val created = permissionRepository.save(Permission(name = permissionDto.name, description = permissionDto.description)).toDto()
-            kafkaPublisher.send(KafkaMessage(content = "CREATED_PREMISSION: ${created.id}", traceId = MDC.get("traceId"), spanId = MDC.get("spanId")))
+            kafkaProducer.send(KafkaMessage(content = "CREATED_PERMISSION: ${created.id}"))
             return@withContext created
         }
 
diff --git a/src/main/kotlin/com/softeno/template/sample/http/api/ExternalController.kt b/src/main/kotlin/com/softeno/template/sample/http/api/ExternalController.kt
index 4f94207..c66ba5c 100644
--- a/src/main/kotlin/com/softeno/template/sample/http/api/ExternalController.kt
+++ b/src/main/kotlin/com/softeno/template/sample/http/api/ExternalController.kt
@@ -4,6 +4,7 @@ import com.softeno.template.sample.http.config.ExternalClientConfig
 import kotlinx.coroutines.reactor.awaitSingle
 import org.apache.commons.logging.LogFactory
 import org.springframework.beans.factory.annotation.Qualifier
+import org.springframework.cloud.client.circuitbreaker.ReactiveCircuitBreaker
 import org.springframework.cloud.client.circuitbreaker.ReactiveCircuitBreakerFactory
 import org.springframework.validation.annotation.Validated
 import org.springframework.web.bind.annotation.GetMapping
@@ -11,6 +12,7 @@ import org.springframework.web.bind.annotation.PathVariable
 import org.springframework.web.bind.annotation.RequestMapping
 import org.springframework.web.bind.annotation.RestController
 import org.springframework.web.reactive.function.client.WebClient
+import org.springframework.web.reactive.function.client.bodyToMono
 import reactor.core.publisher.Mono
 import java.time.Duration
 
@@ -19,24 +21,26 @@ import java.time.Duration
 @Validated
 class ExternalController(
     @Qualifier(value = "external") private val webClient: WebClient,
-    private val reactiveCircuitBreakerFactory: ReactiveCircuitBreakerFactory<*, *>,
-    private val config: ExternalClientConfig
+    reactiveCircuitBreakerFactory: ReactiveCircuitBreakerFactory<*, *>,
+    config: ExternalClientConfig,
+    private val rcb: ReactiveCircuitBreaker = reactiveCircuitBreakerFactory.create(config.name)
 ) {
     private val log = LogFactory.getLog(javaClass)
 
     @GetMapping("/{id}")
     suspend fun getHandler(@PathVariable id: String): SampleResponseDto {
+
         log.info("[external]: GET id: $id")
         val response: SampleResponseDto = webClient.get()
             .uri("/${id}")
             .retrieve()
-            .bodyToMono(SampleResponseDto::class.java)
+            .bodyToMono<SampleResponseDto>()
             .timeout(Duration.ofMillis(1_000))
             .transform {
-                val rcb = reactiveCircuitBreakerFactory.create(config.name)
-                // note: custom exception might be thrown since exception handler is defined
+                // note: a custom exception might be thrown since the exception handler is defined
                 rcb.run(it) { Mono.just(SampleResponseDto(data = "FALLBACK")) }
             }
+            .contextCapture()
             .awaitSingle()
 
         log.info("[external]: received: $response")
diff --git a/src/main/kotlin/com/softeno/template/sample/http/api/SampleRouter.kt b/src/main/kotlin/com/softeno/template/sample/http/api/SampleRouter.kt
index 2d02c23..62111e8 100644
--- a/src/main/kotlin/com/softeno/template/sample/http/api/SampleRouter.kt
+++ b/src/main/kotlin/com/softeno/template/sample/http/api/SampleRouter.kt
@@ -57,4 +57,6 @@ class SampleService {
         return Mono.empty()
     }
 
-}
\ No newline at end of file
+}
+
+data class SampleResponseDto(val data: String)
\ No newline at end of file
diff --git a/src/main/kotlin/com/softeno/template/sample/http/api/dto.kt b/src/main/kotlin/com/softeno/template/sample/http/api/dto.kt
deleted file mode 100644
index 822aaf6..0000000
--- a/src/main/kotlin/com/softeno/template/sample/http/api/dto.kt
+++ /dev/null
@@ -1,3 +0,0 @@
-package com.softeno.template.sample.http.api
-
-data class SampleResponseDto(val data: String)
diff --git a/src/main/kotlin/com/softeno/template/sample/http/config/CircuitBreakerConfig.kt b/src/main/kotlin/com/softeno/template/sample/http/config/CircuitBreakerConfig.kt
index 681e3f3..d0696d9 100644
--- a/src/main/kotlin/com/softeno/template/sample/http/config/CircuitBreakerConfig.kt
+++ b/src/main/kotlin/com/softeno/template/sample/http/config/CircuitBreakerConfig.kt
@@ -1,13 +1,10 @@
 package com.softeno.template.sample.http.config
 
 import io.github.resilience4j.circuitbreaker.CircuitBreakerConfig
-import io.github.resilience4j.circuitbreaker.CircuitBreakerRegistry
 import io.github.resilience4j.timelimiter.TimeLimiterConfig
-import io.github.resilience4j.timelimiter.TimeLimiterRegistry
 import org.springframework.cloud.circuitbreaker.resilience4j.ReactiveResilience4JCircuitBreakerFactory
 import org.springframework.cloud.circuitbreaker.resilience4j.Resilience4JConfigBuilder
 import org.springframework.cloud.client.circuitbreaker.Customizer
-import org.springframework.cloud.client.circuitbreaker.ReactiveCircuitBreakerFactory
 import org.springframework.context.annotation.Bean
 import org.springframework.context.annotation.Configuration
 import java.time.Duration
@@ -15,12 +12,6 @@ import java.time.Duration
 
 @Configuration
 class ReactiveCircuitBreakerConfig {
-    @Bean
-    fun reactiveResilience4JCircuitBreakerFactory(): ReactiveCircuitBreakerFactory<*, *> {
-        val circuitBreakerRegistry = CircuitBreakerRegistry.ofDefaults()
-        val timeLimiterRegistry = TimeLimiterRegistry.ofDefaults()
-        return ReactiveResilience4JCircuitBreakerFactory(circuitBreakerRegistry, timeLimiterRegistry, null)
-    }
 
     @Bean
     fun defaultCustomizer(): Customizer<ReactiveResilience4JCircuitBreakerFactory> {
diff --git a/src/main/kotlin/com/softeno/template/sample/http/config/WebClientConfig.kt b/src/main/kotlin/com/softeno/template/sample/http/config/WebClientConfig.kt
index 2caaed3..032eb75 100644
--- a/src/main/kotlin/com/softeno/template/sample/http/config/WebClientConfig.kt
+++ b/src/main/kotlin/com/softeno/template/sample/http/config/WebClientConfig.kt
@@ -1,5 +1,6 @@
 package com.softeno.template.sample.http.config
 
+import io.micrometer.observation.ObservationRegistry
 import org.springframework.boot.context.properties.ConfigurationProperties
 import org.springframework.context.annotation.Bean
 import org.springframework.context.annotation.Configuration
@@ -39,15 +40,17 @@ class WebClientConfig {
     fun buildWebClient(
         config: ExternalClientConfig,
         authorizedClientManager: ReactiveOAuth2AuthorizedClientManager,
-        builder: WebClient.Builder
+        webClient: WebClient.Builder,
+        registry: ObservationRegistry
     ): WebClient {
         val oauth = ServerOAuth2AuthorizedClientExchangeFilterFunction(authorizedClientManager)
         oauth.setDefaultClientRegistrationId("keycloak")
 
-        return builder
+        return webClient
             .filter(oauth)
             .baseUrl(config.url)
             .defaultHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
+            .observationRegistry(registry)
             .build()
     }
 }
\ No newline at end of file
diff --git a/src/main/resources/application-local.properties b/src/main/resources/application-local.properties
new file mode 100644
index 0000000..f5ffa89
--- /dev/null
+++ b/src/main/resources/application-local.properties
@@ -0,0 +1,5 @@
+spring.main.allow-bean-definition-overriding=true
+
+management.tracing.export.otlp.enabled=false
+management.logging.export.otlp.enabled=false
+management.otlp.metrics.export.enabled=false
\ No newline at end of file
diff --git a/src/main/resources/application.properties b/src/main/resources/application.properties
index 89d07e9..072bb93 100644
--- a/src/main/resources/application.properties
+++ b/src/main/resources/application.properties
@@ -1,11 +1,12 @@
 spring.main.allow-bean-definition-overriding=true
 server.port=8082
 spring.application.name=SoftenoJpaPostgresApp
+spring.main.web-application-type=reactive
 
 ### custom: external services
 #com.softeno.external.url=http://localhost:4500/sample
 #com.softeno.external.url=http://localhost:8080/sample
-com.softeno.external.url=http://localhost:8080/sample-secured
+com.softeno.external.url=http://localhost:8082/sample-secured
 com.softeno.external.name=node-service
 
 com.softeno.kafka.tx=sample_topic_2
@@ -41,9 +42,9 @@ spring.security.oauth2.client.provider.keycloak.jwk-set-uri=http://localhost:809
 
 # kafka
 spring.kafka.bootstrap-servers=localhost:9094
+spring.kafka.consumer.group-id=other-group-jvm
 spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
 spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer
-spring.kafka.consumer.group-id=other-group-jvm
 spring.kafka.consumer.auto-offset-reset=earliest
 spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
 spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
@@ -55,8 +56,6 @@ spring.kafka.consumer.properties.spring.json.value.default.type=com.fasterxml.ja
 management.endpoints.web.base-path=/actuator
 management.endpoints.web.exposure.include=*
 management.endpoint.health.show-details=always
-management.endpoint.metrics.enabled=true
-management.endpoint.prometheus.enabled=true
 
 # swagger
 springdoc.api-docs.enabled=true
@@ -68,18 +67,25 @@ resilience4j.circuitbreaker.metrics.enabled=true
 resilience4j.retry.metrics.enabled=true
 resilience4j.ratelimiter.metrics.enabled=true
 
-## observation & zipkin
-management.tracing.enabled=true
-management.zipkin.tracing.endpoint=http://localhost:9411/api/v2/spans
+# metrics, prometheus & actuator
+management.otlp.metrics.export.enabled=true
+management.otlp.metrics.export.url=http://localhost:4318/v1/metrics
+
+management.tracing.export.enabled=true
+management.opentelemetry.tracing.export.otlp.endpoint=http://localhost:4318/v1/traces
+management.opentelemetry.logging.export.otlp.endpoint=http://localhost:4318/v1/logs
 management.tracing.sampling.probability=1.0
-management.tracing.propagation.consume=b3
-management.tracing.propagation.produce=b3
-management.tracing.propagation.type=b3
 
-spring.kafka.template.observation-enabled=true
-spring.kafka.listener.observation-enabled=true
-management.tracing.baggage.correlation.enabled=true
+spring.reactor.context-propagation=auto
+
+management.tracing.propagation.consume=w3c
+management.tracing.propagation.produce=w3c
+management.tracing.propagation.type=w3c
+
 management.tracing.baggage.enabled=true
-management.tracing.baggage.correlation.fields=spanId,traceId
+management.tracing.baggage.correlation.enabled=true
+
+# fix for tracing logs
+management.observations.enable.spring.security=false
+logging.level.io.micrometer.observation.contextpropagation.ObservationThreadLocalAccessor=ERROR
 
-management.otlp.metrics.export.enabled=true
diff --git a/src/main/resources/logback-spring.xml b/src/main/resources/logback-spring.xml
index 2655d5b..71fbb71 100644
--- a/src/main/resources/logback-spring.xml
+++ b/src/main/resources/logback-spring.xml
@@ -1,6 +1,7 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <configuration>
     <property name="projectName" value="template-service-c"/>
+    <property name="LOGS" value="./logs"/>
     <property name="defaultPattern"
               value="%d{ISO8601} [${projectName}] [%X{traceId:-}] [%X{spanId:-}] %highlight(%-5level) [%blue(%t)] %yellow(%C{1}): %msg%n%throwable"/>
 
@@ -12,7 +13,26 @@
         </encoder>
     </appender>
 
-    <springProfile name="dev">
+    <appender name="RollingFile" class="ch.qos.logback.core.rolling.RollingFileAppender">
+        <file>${LOGS}/spring-boot-logger.log</file>
+        <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
+            <Pattern>%d %p %C [%t] %m%n</Pattern>
+        </encoder>
+
+        <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
+            <fileNamePattern>
+                ${LOGS}/archived/spring-boot-logger-%d{yyyy-MM-dd}.%i.log
+            </fileNamePattern>
+            <timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
+                <maxFileSize>10MB</maxFileSize>
+            </timeBasedFileNamingAndTriggeringPolicy>
+        </rollingPolicy>
+    </appender>
+
+    <appender name="OpenTelemetry" class="io.opentelemetry.instrumentation.logback.appender.v1_0.OpenTelemetryAppender">
+    </appender>
+
+    <springProfile name="local">
         <root level="info">
             <appender-ref ref="Console"/>
         </root>
@@ -21,64 +41,27 @@
         </logger>
     </springProfile>
 
-    <springProfile name="!dev">
-        <property name="LOGSTASH_HOST" value="${logstash.host:-logstash}" />
-        <property name="LOGSTASH_PORT" value="${logstash.port:-50000}" />
-        <property name="LOGS" value="./logs"/>
-        <appender name="logstash" class="net.logstash.logback.appender.LogstashTcpSocketAppender">
-            <destination>${LOGSTASH_HOST}:${LOGSTASH_PORT}</destination>
-            <encoder class="net.logstash.logback.encoder.LoggingEventCompositeJsonEncoder">
-                <providers>
-                    <mdc/>
-                    <pattern>
-                        <pattern>
-                            {
-                            "application": "${projectName}",
-                            "timestamp": "%d{yyyy-MM-dd'T'HH:mm:ss.SSSZZ}",
-                            "logger": "%logger",
-                            "trace": "%X{traceId:-}",
-                            "span": "%X{spanId:-}",
-                            "thread": "%thread",
-                            "level": "%level",
-                            "message": "%message"
-                            }
-                        </pattern>
-                    </pattern>
-                    <stackTrace>
-                        <throwableConverter class="net.logstash.logback.stacktrace.ShortenedThrowableConverter">
-                            <maxDepthPerThrowable>2048</maxDepthPerThrowable>
-                            <maxLength>32768</maxLength>
-                            <shortenedClassNameLength>256</shortenedClassNameLength>
-                            <rootCauseFirst>true</rootCauseFirst>
-                        </throwableConverter>
-                    </stackTrace>
-                </providers>
-            </encoder>
-        </appender>
-        <appender name="RollingFile" class="ch.qos.logback.core.rolling.RollingFileAppender">
-            <file>${LOGS}/spring-boot-logger.log</file>
-            <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
-                <Pattern>%d %p %C [%t] %m%n</Pattern>
-            </encoder>
+    <springProfile name="dev">
+        <root level="info">
+            <appender-ref ref="Console"/>
+            <appender-ref ref="OpenTelemetry"/>
+        </root>
+        <logger name="com.softeno" level="trace" additivity="false">
+            <appender-ref ref="Console"/>
+            <appender-ref ref="OpenTelemetry"/>
+        </logger>
+    </springProfile>
 
-            <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
-                <fileNamePattern>
-                    ${LOGS}/archived/spring-boot-logger-%d{yyyy-MM-dd}.%i.log
-                </fileNamePattern>
-                <timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
-                    <maxFileSize>10MB</maxFileSize>
-                </timeBasedFileNamingAndTriggeringPolicy>
-            </rollingPolicy>
-        </appender>
+    <springProfile name="docker">
         <root level="info">
             <appender-ref ref="RollingFile"/>
             <appender-ref ref="Console"/>
-            <appender-ref ref="logstash"/>
+            <appender-ref ref="OpenTelemetry"/>
         </root>
         <logger name="com.softeno" level="trace" additivity="false">
             <appender-ref ref="RollingFile"/>
             <appender-ref ref="Console"/>
-            <appender-ref ref="logstash"/>
+            <appender-ref ref="OpenTelemetry"/>
         </logger>
         <!--    db -->
         <logger name="logging.level.io.r2dbc" level="debug" additivity="false" >
diff --git a/src/test/kotlin/com/softeno/template/app/IntegrationTest.kt b/src/test/kotlin/com/softeno/template/app/IntegrationTest.kt
index d6e1be9..152101e 100644
--- a/src/test/kotlin/com/softeno/template/app/IntegrationTest.kt
+++ b/src/test/kotlin/com/softeno/template/app/IntegrationTest.kt
@@ -28,8 +28,8 @@ import org.junit.jupiter.api.assertThrows
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.boot.context.properties.ConfigurationPropertiesScan
 import org.springframework.boot.context.properties.EnableConfigurationProperties
-import org.springframework.boot.test.autoconfigure.web.reactive.AutoConfigureWebTestClient
 import org.springframework.boot.test.context.SpringBootTest
+import org.springframework.boot.webtestclient.autoconfigure.AutoConfigureWebTestClient
 import org.springframework.core.Ordered
 import org.springframework.core.annotation.Order
 import org.springframework.data.domain.Pageable
@@ -41,8 +41,10 @@ import org.springframework.test.web.reactive.server.WebTestClient
 import org.springframework.test.web.reactive.server.expectBody
 import org.springframework.web.reactive.function.BodyInserters
 import org.springframework.web.reactive.function.client.WebClient
-import org.testcontainers.containers.PostgreSQLContainer
 import org.testcontainers.junit.jupiter.Container
+import org.testcontainers.kafka.KafkaContainer
+import org.testcontainers.postgresql.PostgreSQLContainer
+import org.testcontainers.utility.DockerImageName
 
 @SpringBootTest(
     classes = [SoftenoMvcJpaApp::class],
@@ -64,7 +66,13 @@ abstract class BaseIntegrationTest {
 
     companion object {
         @Container
-        var postgreSQLContainer = PostgreSQLContainer("postgres:16-alpine")
+        var kafka: KafkaContainer = KafkaContainer(DockerImageName.parse("apache/kafka-native:3.8.0"))
+            .withEnv("KAFKA_AUTO_CREATE_TOPICS_ENABLE", "true")
+            .withEnv("ALLOW_PLAINTEXT_LISTENER", "true")
+            .withEnv("KAFKA_CREATE_TOPICS", "sample_topic_2" + ":1:1")
+
+        @Container
+        var postgreSQLContainer = PostgreSQLContainer(DockerImageName.parse("postgres:16-alpine"))
             .withDatabaseName("application")
             .withUsername("admin")
             .withPassword("admin")
@@ -73,8 +81,12 @@ abstract class BaseIntegrationTest {
         @JvmStatic
         @DynamicPropertySource
         fun registerDynamicProperties(registry: DynamicPropertyRegistry) {
-            postgreSQLContainer.start()
+            kafka.start()
+            registry.add("spring.kafka.bootstrap-servers") {
+                kafka.bootstrapServers
+            }
 
+            postgreSQLContainer.start()
             registry.add("spring.liquibase.url") {
                 "jdbc:postgresql://${postgreSQLContainer.host}:${postgreSQLContainer.firstMappedPort}/${postgreSQLContainer.databaseName}"
             }
@@ -89,6 +101,7 @@ abstract class BaseIntegrationTest {
             registry.add("spring.r2dbc.username") { postgreSQLContainer.username }
             registry.add("spring.r2dbc.password") { postgreSQLContainer.password }
         }
+
     }
 
     @BeforeEach
@@ -110,6 +123,7 @@ class ContextLoadsTest : BaseIntegrationTest() {
     @Test
     fun testConnection() {
         assertTrue(postgreSQLContainer.isRunning)
+        assertTrue(kafka.isRunning)
     }
 }
 
@@ -240,6 +254,57 @@ class BatchPermissionRepositoryImplTest : BaseIntegrationTest(), PermissionFixtu
         assertEquals(changedSaved[1].modifiedDate, changedAnotherPermission.modifiedDate)
         assertEquals(changedSaved[1].modifiedBy, changedAnotherPermission.modifiedBy)
     }
+
+    @Test
+    fun `should allow batch update by setting null values to the props that were non-null before`() = runTest {
+        // given
+        val aPermission = aPermission(name = "1", description = "before change 1")
+            .copy(createdBy = "user 1", createdDate = 123456L, modifiedBy = "user 2", modifiedDate = 123456L, version = 0)
+
+        // when
+        val returned = permissionRepository.insertAllReturningIds(listOf(aPermission))
+        val aPermissionWithId = aPermission.copy(id = returned[aPermission])
+
+        // then
+        assertEquals(returned.size, 1)
+
+        val saved = permissionRepository.findAll().toList()
+        assertEquals(saved.size, 1)
+
+        assertEquals(saved[0].uuid, aPermission.uuid)
+        assertEquals(saved[0].name, aPermission.name)
+        assertEquals(saved[0].description, aPermission.description)
+        assertEquals(saved[0].uuid, aPermission.uuid)
+        assertEquals(saved[0].version, aPermission.version)
+        assertEquals(saved[0].createdDate, aPermission.createdDate)
+        assertEquals(saved[0].createdBy, aPermission.createdBy)
+        assertEquals(saved[0].modifiedBy, aPermission.modifiedBy)
+        assertEquals(saved[0].modifiedDate, aPermission.modifiedDate)
+
+
+        // when change
+        val changedPermission = aPermissionWithId.copy(description = "after change 1", modifiedDate = 1111111L, modifiedBy = null, createdBy = null, version = saved[0].version + 1)
+
+        val changedReturned = permissionRepository.updateAllReturningIds(listOf(changedPermission))
+
+        // then
+        assertEquals(changedPermission.uuid, saved[0].uuid)
+
+        assertEquals(changedReturned.size, 1)
+
+        val changedSaved = permissionRepository.findAll().toList()
+        assertEquals(changedSaved.size, 1)
+
+        assertEquals(changedSaved[0].id, changedPermission.id)
+        assertEquals(changedSaved[0].name, changedPermission.name)
+        assertEquals(changedSaved[0].description, changedPermission.description)
+        assertEquals(changedSaved[0].uuid, changedPermission.uuid)
+        assertEquals(changedSaved[0].version, changedPermission.version)
+        assertEquals(changedSaved[0].createdDate, changedPermission.createdDate)
+        assertEquals(changedSaved[0].createdBy, changedPermission.createdBy)
+        assertEquals(changedSaved[0].modifiedDate, changedPermission.modifiedDate)
+        assertEquals(changedSaved[0].modifiedBy, changedPermission.modifiedBy)
+    }
 }
 
 class PermissionTest : BaseIntegrationTest(), PermissionFixture {
diff --git a/src/test/resources/application.properties b/src/test/resources/application.properties
index 19ad617..af153d7 100644
--- a/src/test/resources/application.properties
+++ b/src/test/resources/application.properties
@@ -10,7 +10,6 @@ com.softeno.kafka.rx=sample_topic_2
 com.softeno.kafka.keycloak=keycloak-events
 
 ### kafka
-spring.kafka.bootstrap-servers=localhost:9092
 spring.kafka.consumer.group-id=sample-group-jvm
 spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
 spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer
@@ -19,6 +18,7 @@ spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.Str
 spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
 spring.kafka.properties.spring.json.trusted.packages=*
 spring.kafka.consumer.properties.spring.json.use.type.headers=false
+spring.kafka.consumer.properties.spring.json.value.default.type=com.fasterxml.jackson.databind.JsonNode
 
 ##### r2dbc
 spring.r2dbc.pool.enabled=true
@@ -32,9 +32,6 @@ logging.level.org.springframework.data=DEBUG
 logging.level.org.springframework.r2dbc=TRACE
 logging.level.liquibase=DEBUG
 
-# metrics, prometheus & actuator
-management.endpoint.metrics.enabled=false
-management.endpoint.prometheus.enabled=false
-
 # swagger
 springdoc.api-docs.enabled=false
+