fixing(infra-messaging)

2025-08-15 18:18:40 +02:00
parent b5656156c4
commit c67fe3004e
14 changed files with 1422 additions and 1508 deletions
@@ -1,14 +1,28 @@
 package at.mocode.infrastructure.messaging.client

 import reactor.core.publisher.Flux
+import kotlinx.coroutines.flow.Flow

 /**
- * A generic, reactive interface for consuming events from a message broker.
+ * A generic interface for consuming events from a message broker.
+ *
+ * Follows DDD principles with explicit error handling using domain-specific error types.
+ * Provides both Result-based methods and reactive streams for flexibility.
 */
 interface EventConsumer {

    /**
-     * Receives a continuous stream of events from the specified topic.
+     * Receives events from the specified topic with explicit error handling.
+     *
+     * @param T The expected type of the event payload
+     * @param topic The topic to subscribe to
+     * @param eventType The class type of events to consume
+     * @return Flow<Result<T>> where each Result contains either a successful event or MessagingError
+     */
+    fun <T : Any> receiveEventsWithResult(topic: String, eventType: Class<T>): Flow<Result<T>>
+
+    /**
+     * Legacy reactive method for receiving events.
     *
     * This method returns a cold Flux, meaning that the consumer will only start
     * listening for messages once the Flux is subscribed to.
@@ -17,6 +31,7 @@ interface EventConsumer {
     * @param topic The topic to subscribe to.
     * @return A reactive stream (Flux) of events of type T.
     */
+    @Deprecated("Use receiveEventsWithResult with Flow<Result<T>> instead", ReplaceWith("receiveEventsWithResult(topic, eventType)"))
    fun <T : Any> receiveEvents(topic: String, eventType: Class<T>): Flux<T>
 }

@@ -5,18 +5,38 @@ import reactor.core.publisher.Mono

 /**
 * Interface for publishing domain events to message broker.
+ *
+ * Follows DDD principles with explicit error handling using domain-specific error types.
+ * All operations use the Result pattern for type-safe error handling as required by guidelines.
 */
 interface EventPublisher {

    /**
     * Publishes a single event to the specified topic.
-     * Returns a Mono that emits Unit when the send operation is finished.
+     *
+     * @param topic The Kafka topic to publish to
+     * @param key Optional message key for partitioning
+     * @param event The domain event to publish
+     * @return Result<Unit> indicating success or MessagingError exception for specific failure reason
     */
-    fun publishEvent(topic: String, key: String? = null, event: Any): Mono<Unit>
+    suspend fun publishEvent(topic: String, key: String? = null, event: Any): Result<Unit>

    /**
-     * Publishes multiple events to the specified topic.
-     * Returns a Flux that emits one Unit per successfully published event.
+     * Publishes multiple events to the specified topic in batch.
+     *
+     * @param topic The Kafka topic to publish to
+     * @param events List of key-event pairs to publish
+     * @return Result<List<Unit>> with success indicators or MessagingError exception for failure reason
     */
-    fun publishEvents(topic: String, events: List<Pair<String?, Any>>): Flux<Unit>
+    suspend fun publishEvents(topic: String, events: List<Pair<String?, Any>>): Result<List<Unit>>
+
+    /**
+     * Legacy reactive methods for backward compatibility.
+     * These will be deprecated in favor of the Result-based methods above.
+     */
+    @Deprecated("Use suspending publishEvent with Result instead", ReplaceWith("publishEvent(topic, key, event)"))
+    fun publishEventReactive(topic: String, key: String? = null, event: Any): Mono<Unit>
+
+    @Deprecated("Use suspending publishEvents with Result instead", ReplaceWith("publishEvents(topic, events)"))
+    fun publishEventsReactive(topic: String, events: List<Pair<String?, Any>>): Flux<Unit>
 }
@@ -1,7 +1,8 @@
 package at.mocode.infrastructure.messaging.client

 import at.mocode.infrastructure.messaging.config.KafkaConfig
-import org.apache.kafka.clients.consumer.ConsumerConfig
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.reactive.asFlow
 import org.slf4j.LoggerFactory
 import org.springframework.kafka.support.serializer.JsonDeserializer
 import org.springframework.stereotype.Component
@@ -10,7 +11,7 @@ import reactor.kafka.receiver.KafkaReceiver
 import reactor.kafka.receiver.ReceiverOptions
 import reactor.util.retry.Retry
 import java.time.Duration
-import java.util.Collections
+import java.util.*
 import java.util.concurrent.ConcurrentHashMap

 /**
@@ -27,6 +28,22 @@ class KafkaEventConsumer(
    // Connection pool to reuse KafkaReceiver instances per topic-eventType combination
    private val receiverCache = ConcurrentHashMap<String, KafkaReceiver<String, Any>>()

+    override fun <T : Any> receiveEventsWithResult(topic: String, eventType: Class<T>): Flow<Result<T>> {
+        logger.info("Setting up Result-based consumer for topic '{}' with event type '{}'", topic, eventType.simpleName)
+
+        return receiveEvents(topic, eventType)
+            .map<Result<T>> { event -> Result.success(event) }
+            .onErrorContinue { error, _ ->
+                logger.warn("Error occurred while consuming events from topic '{}' for event type '{}': {}",
+                    topic, eventType.simpleName, error.message)
+            }
+            .doOnError { exception ->
+                logger.error("Fatal error in consumer stream for topic '{}' and event type '{}': {}",
+                    topic, eventType.simpleName, exception.message, exception)
+            }
+            .asFlow()
+    }
+
    override fun <T : Any> receiveEvents(topic: String, eventType: Class<T>): Flux<T> {
        logger.info("Setting up reactive consumer for topic '{}' with event type '{}'", topic, eventType.simpleName)

@@ -1,17 +1,20 @@
 package at.mocode.infrastructure.messaging.client

+import kotlinx.coroutines.reactor.awaitSingle
 import org.slf4j.LoggerFactory
 import org.springframework.kafka.core.reactive.ReactiveKafkaProducerTemplate
 import org.springframework.stereotype.Component
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
-import reactor.kafka.sender.SenderResult
 import reactor.util.retry.Retry
 import java.time.Duration

 /**
 * A reactive, non-blocking Kafka implementation of EventPublisher with enhanced
 * error handling, retry mechanisms, and optimized batch processing.
+ *
+ * Implements both Result-based methods (preferred) and reactive methods (legacy).
+ * Follows DDD principles with explicit error handling using domain-specific error types.
 */
@Component
 class KafkaEventPublisher(
@@ -21,13 +24,41 @@ class KafkaEventPublisher(
    private val logger = LoggerFactory.getLogger(KafkaEventPublisher::class.java)

    companion object {
-        private const val DEFAULT_RETRY_ATTEMPTS = 3L
-        private const val DEFAULT_RETRY_DELAY_SECONDS = 1L
-        private const val DEFAULT_MAX_BACKOFF_SECONDS = 10L
-        private const val DEFAULT_BATCH_CONCURRENCY = 10
+        /** Maximum number of retry attempts for failed message publishing operations */
+        private const val MAX_RETRY_ATTEMPTS = 3L
+
+        /** Initial delay in seconds between retry attempts */
+        private const val RETRY_DELAY_SECONDS = 1L
+
+        /** Maximum backoff delay in seconds for exponential backoff retry strategy */
+        private const val MAX_BACKOFF_SECONDS = 10L
+
+        /** Default concurrency level for batch processing operations */
+        private const val BATCH_CONCURRENCY_LEVEL = 10
+
+        /** Progress logging interval for batch operations (every N events) */
+        private const val BATCH_PROGRESS_LOG_INTERVAL = 100
    }

-    override fun publishEvent(topic: String, key: String?, event: Any): Mono<Unit> {
+    override suspend fun publishEvent(topic: String, key: String?, event: Any): Result<Unit> {
+        return try {
+            publishEventReactive(topic, key, event).awaitSingle()
+            Result.success(Unit)
+        } catch (exception: Throwable) {
+            Result.failure(mapToMessagingError(exception))
+        }
+    }
+
+    override suspend fun publishEvents(topic: String, events: List<Pair<String?, Any>>): Result<List<Unit>> {
+        return try {
+            val results = publishEventsReactive(topic, events).collectList().awaitSingle()
+            Result.success(results)
+        } catch (exception: Throwable) {
+            Result.failure(mapToMessagingError(exception))
+        }
+    }
+
+    override fun publishEventReactive(topic: String, key: String?, event: Any): Mono<Unit> {
        logger.debug("Publishing event to topic '{}' with key '{}', event type: '{}'",
            topic, key, event::class.simpleName)

@@ -51,7 +82,7 @@ class KafkaEventPublisher(
            .map { Unit }
    }

-    override fun publishEvents(topic: String, events: List<Pair<String?, Any>>): Flux<Unit> {
+    override fun publishEventsReactive(topic: String, events: List<Pair<String?, Any>>): Flux<Unit> {
        if (events.isEmpty()) {
            logger.debug("No events to publish to topic '{}'", topic)
            return Flux.empty()
@@ -70,7 +101,7 @@ class KafkaEventPublisher(
                        val record = result.recordMetadata()
                        logger.debug("Successfully published event to topic-partition {}-{} with offset {} (key: '{}')",
                            record.topic(), record.partition(), record.offset(), key)
-                        if ((index + 1) % 100 == 0L || index == events.size.toLong() - 1) {
+                        if ((index + 1) % BATCH_PROGRESS_LOG_INTERVAL == 0L || index == events.size.toLong() - 1) {
                            logger.info("Batch progress: {}/{} events published to topic '{}'",
                                index + 1, events.size, topic)
                        }
@@ -85,7 +116,7 @@ class KafkaEventPublisher(
                        logger.error("Error publishing event {} in batch to topic '{}': {}",
                            index + 1, topic, error.message)
                    }
-            }, DEFAULT_BATCH_CONCURRENCY) // Controlled concurrency for better resource management
+            }, BATCH_CONCURRENCY_LEVEL) // Controlled concurrency for better resource management
            .doOnComplete {
                logger.info("Completed publishing batch of {} events to topic '{}'", events.size, topic)
            }
@@ -98,8 +129,8 @@ class KafkaEventPublisher(
     * Creates a retry specification with exponential backoff for robust error handling.
     */
    private fun createRetrySpec(topic: String, key: String?): Retry =
-        Retry.backoff(DEFAULT_RETRY_ATTEMPTS, Duration.ofSeconds(DEFAULT_RETRY_DELAY_SECONDS))
-            .maxBackoff(Duration.ofSeconds(DEFAULT_MAX_BACKOFF_SECONDS))
+        Retry.backoff(MAX_RETRY_ATTEMPTS, Duration.ofSeconds(RETRY_DELAY_SECONDS))
+            .maxBackoff(Duration.ofSeconds(MAX_BACKOFF_SECONDS))
            .filter { exception ->
                // Only retry on transient errors (not serialization errors, etc.)
                isRetryableException(exception)
@@ -115,6 +146,29 @@ class KafkaEventPublisher(
                retrySignal.failure()
            }

+    /**
+     * Maps generic exceptions to domain-specific MessagingError types.
+     */
+    private fun mapToMessagingError(exception: Throwable): MessagingError {
+        return when {
+            exception.message?.contains("serializ", ignoreCase = true) == true ->
+                MessagingError.SerializationError("Serialization failed: ${exception.message}", exception)
+            exception.message?.contains("timeout", ignoreCase = true) == true ||
+            exception is java.util.concurrent.TimeoutException ->
+                MessagingError.TimeoutError("Operation timed out: ${exception.message}", exception)
+            exception.message?.contains("connection", ignoreCase = true) == true ||
+            exception.message?.contains("network", ignoreCase = true) == true ||
+            exception is java.net.ConnectException ||
+            exception is java.io.IOException ->
+                MessagingError.ConnectionError("Connection failed: ${exception.message}", exception)
+            exception.message?.contains("auth", ignoreCase = true) == true ->
+                MessagingError.AuthenticationError("Authentication failed: ${exception.message}", exception)
+            exception.message?.contains("topic", ignoreCase = true) == true ->
+                MessagingError.TopicConfigurationError("Topic configuration error: ${exception.message}", exception)
+            else -> MessagingError.UnexpectedError("Unexpected error: ${exception.message}", exception)
+        }
+    }
+
    /**
     * Determines if an exception is retryable based on its type and characteristics.
     */
@@ -1,311 +0,0 @@
-package at.mocode.infrastructure.messaging.client
-
-import at.mocode.infrastructure.messaging.client.ReactiveKafkaConfig
-import at.mocode.infrastructure.messaging.config.KafkaConfig
-import org.assertj.core.api.Assertions.assertThat
-import org.junit.jupiter.api.AfterEach
-import org.junit.jupiter.api.BeforeEach
-import org.junit.jupiter.api.Test
-import org.junit.jupiter.api.TestInstance
-import org.slf4j.LoggerFactory
-import org.springframework.kafka.core.DefaultKafkaProducerFactory
-import org.testcontainers.containers.KafkaContainer
-import org.testcontainers.junit.jupiter.Container
-import org.testcontainers.junit.jupiter.Testcontainers
-import org.testcontainers.utility.DockerImageName
-import reactor.test.StepVerifier
-import java.util.*
-
-@Testcontainers
-@TestInstance(TestInstance.Lifecycle.PER_CLASS)
-class KafkaBatchPerformanceTest {
-
-    private val logger = LoggerFactory.getLogger(KafkaBatchPerformanceTest::class.java)
-
-    companion object {
-        @Container
-        private val kafkaContainer = KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:7.5.0"))
-    }
-
-    private lateinit var kafkaEventPublisher: KafkaEventPublisher
-    private lateinit var producerFactory: DefaultKafkaProducerFactory<String, Any>
-    private val testTopic = "performance-topic-${UUID.randomUUID()}"
-
-    @BeforeEach
-    fun setUp() {
-        val kafkaConfig = KafkaConfig().apply {
-            bootstrapServers = kafkaContainer.bootstrapServers
-            trustedPackages = "at.mocode.*"
-        }
-        producerFactory = kafkaConfig.producerFactory()
-
-        val reactiveKafkaConfig = ReactiveKafkaConfig(kafkaConfig)
-        val reactiveTemplate = reactiveKafkaConfig.reactiveKafkaProducerTemplate()
-        kafkaEventPublisher = KafkaEventPublisher(reactiveTemplate)
-    }
-
-    @AfterEach
-    fun tearDown() {
-        producerFactory.destroy()
-    }
-
-    @Test
-    fun `should handle small batch efficiently`() {
-        val batchSize = 50
-        val smallEventBatch = (1..batchSize).map { i ->
-            "key$i" to PerformanceTestEvent("Small batch message $i", i)
-        }
-
-        val startTime = System.currentTimeMillis()
-
-        StepVerifier.create(kafkaEventPublisher.publishEvents(testTopic, smallEventBatch))
-            .expectNextCount(batchSize.toLong())
-            .verifyComplete()
-
-        val duration = System.currentTimeMillis() - startTime
-
-        // Small batch should complete quickly (within 10 seconds)
-        assertThat(duration).isLessThan(10000)
-    }
-
-    @Test
-    fun `should handle medium batch efficiently`() {
-        val batchSize = 500
-        val mediumEventBatch = (1..batchSize).map { i ->
-            "key$i" to PerformanceTestEvent("Medium batch message $i", i)
-        }
-
-        val startTime = System.currentTimeMillis()
-
-        StepVerifier.create(kafkaEventPublisher.publishEvents(testTopic, mediumEventBatch))
-            .expectNextCount(batchSize.toLong())
-            .verifyComplete()
-
-        val duration = System.currentTimeMillis() - startTime
-
-        // Medium batch should complete within a reasonable time (30 seconds)
-        assertThat(duration).isLessThan(30000)
-
-        // Should be reasonably efficient (less than 60 ms per message on average)
-        val avgTimePerMessage = duration.toDouble() / batchSize
-        assertThat(avgTimePerMessage).isLessThan(60.0)
-    }
-
-    @Test
-    fun `should handle large batch with reasonable performance`() {
-        val batchSize = 1000
-        val largeEventBatch = (1..batchSize).map { i ->
-            "key$i" to PerformanceTestEvent("Large batch message $i", i)
-        }
-
-        val startTime = System.currentTimeMillis()
-
-        StepVerifier.create(kafkaEventPublisher.publishEvents(testTopic, largeEventBatch))
-            .expectNextCount(batchSize.toLong())
-            .verifyComplete()
-
-        val duration = System.currentTimeMillis() - startTime
-
-        // Large batch should complete within 60 seconds
-        assertThat(duration).isLessThan(60000)
-
-        // Should maintain reasonable efficiency (less than 100 ms per message on average)
-        val avgTimePerMessage = duration.toDouble() / batchSize
-        assertThat(avgTimePerMessage).isLessThan(100.0)
-    }
-
-    @Test
-    fun `should handle concurrent batch publishing`() {
-        val batchSize = 100
-        val concurrentBatches = 5
-
-        val batches = (1..concurrentBatches).map { batchIndex ->
-            (1..batchSize).map { i ->
-                "batch${batchIndex}_key$i" to PerformanceTestEvent("Concurrent batch $batchIndex message $i", i)
-            }
-        }
-
-        val startTime = System.currentTimeMillis()
-
-        // Publish all batches concurrently
-        val publishers = batches.map { batch ->
-            kafkaEventPublisher.publishEvents(testTopic, batch)
-                .collectList() // Collect results for each batch
-        }
-
-        StepVerifier.create(reactor.core.publisher.Flux.merge(publishers))
-            .expectNextCount(concurrentBatches.toLong())
-            .verifyComplete()
-
-        val duration = System.currentTimeMillis() - startTime
-
-        // Concurrent publishing should be efficient (within 45 seconds for all batches)
-        assertThat(duration).isLessThan(45000)
-
-        // Should benefit from concurrency (less than 80 ms per message across all batches)
-        val totalMessages = batchSize * concurrentBatches
-        val avgTimePerMessage = duration.toDouble() / totalMessages
-        assertThat(avgTimePerMessage).isLessThan(80.0)
-    }
-
-    @Test
-    fun `should handle single message publishing performance`() {
-        val messageCount = 100
-        val messages = (1..messageCount).map { i ->
-            PerformanceTestEvent("Single message $i", i)
-        }
-
-        val startTime = System.currentTimeMillis()
-
-        val publishers = messages.mapIndexed { index, message ->
-            kafkaEventPublisher.publishEvent(testTopic, "single_key_$index", message)
-        }
-
-        StepVerifier.create(reactor.core.publisher.Flux.merge(publishers))
-            .expectNextCount(messageCount.toLong())
-            .verifyComplete()
-
-        val duration = System.currentTimeMillis() - startTime
-
-        // Individual message publishing should complete within 20 seconds
-        assertThat(duration).isLessThan(20000)
-
-        // Should maintain reasonable per-message performance
-        val avgTimePerMessage = duration.toDouble() / messageCount
-        assertThat(avgTimePerMessage).isLessThan(200.0)
-    }
-
-    @Test
-    fun `should handle mixed payload sizes efficiently`() {
-        val smallPayload = "small"
-        val mediumPayload = "medium".repeat(100) // ~600 characters
-        val largePayload = "large".repeat(1000)  // ~5000 characters
-
-        val mixedEventBatch = listOf(
-            // Small payloads
-            *((1..50).map { i -> "small_key_$i" to PerformanceTestEvent(smallPayload, i) }.toTypedArray()),
-            // Medium payloads
-            *((1..30).map { i -> "medium_key_$i" to PerformanceTestEvent(mediumPayload, i) }.toTypedArray()),
-            // Large payloads
-            *((1..20).map { i -> "large_key_$i" to PerformanceTestEvent(largePayload, i) }.toTypedArray())
-        )
-
-        val startTime = System.currentTimeMillis()
-
-        StepVerifier.create(kafkaEventPublisher.publishEvents(testTopic, mixedEventBatch))
-            .expectNextCount(100) // 50 + 30 + 20 = 100
-            .verifyComplete()
-
-        val duration = System.currentTimeMillis() - startTime
-
-        // Mixed payload sizes should be handled efficiently (within 15 seconds)
-        assertThat(duration).isLessThan(15000)
-    }
-
-    @Test
-    fun `should demonstrate batch vs individual performance difference`() {
-        val messageCount = 200
-        val events = (1..messageCount).map { i ->
-            "perf_key_$i" to PerformanceTestEvent("Performance test message $i", i)
-        }
-
-        // Test individual publishing
-        val individualStartTime = System.currentTimeMillis()
-        val individualPublishers = events.map { (key, event) ->
-            kafkaEventPublisher.publishEvent(testTopic, key, event)
-        }
-
-        StepVerifier.create(reactor.core.publisher.Flux.merge(individualPublishers))
-            .expectNextCount(messageCount.toLong())
-            .verifyComplete()
-
-        val individualDuration = System.currentTimeMillis() - individualStartTime
-
-        // Test batch publishing
-        val batchStartTime = System.currentTimeMillis()
-
-        StepVerifier.create(kafkaEventPublisher.publishEvents(testTopic, events))
-            .expectNextCount(messageCount.toLong())
-            .verifyComplete()
-
-        val batchDuration = System.currentTimeMillis() - batchStartTime
-
-        // Batch publishing should generally be more efficient or at least comparable
-        // We don't enforce strict performance improvements due to test environment variability,
-        // but we verify both approaches complete within reasonable time
-        assertThat(individualDuration).isLessThan(20000)
-        assertThat(batchDuration).isLessThan(20000)
-
-        logger.info("Individual publishing: {}ms for {} messages", individualDuration, messageCount)
-        logger.info("Batch publishing: {}ms for {} messages", batchDuration, messageCount)
-    }
-
-    @Test
-    fun `should handle empty batch gracefully`() {
-        val emptyBatch = emptyList<Pair<String?, Any>>()
-
-        val startTime = System.currentTimeMillis()
-
-        StepVerifier.create(kafkaEventPublisher.publishEvents(testTopic, emptyBatch))
-            .verifyComplete()
-
-        val duration = System.currentTimeMillis() - startTime
-
-        // Empty batch should complete almost instantly (within 100 ms)
-        assertThat(duration).isLessThan(100)
-    }
-
-    @Test
-    fun `should maintain performance under memory pressure`() {
-        // Create a large batch to test memory handling
-        val largeBatchSize = 2000
-        val largeEventBatch = (1..largeBatchSize).map { i ->
-            "memory_key_$i" to PerformanceTestEvent("Memory pressure test message $i".repeat(10), i)
-        }
-
-        val startTime = System.currentTimeMillis()
-
-        StepVerifier.create(kafkaEventPublisher.publishEvents(testTopic, largeEventBatch))
-            .expectNextCount(largeBatchSize.toLong())
-            .verifyComplete()
-
-        val duration = System.currentTimeMillis() - startTime
-
-        // Should handle large batches without excessive memory usage (within 45 seconds)
-        assertThat(duration).isLessThan(45000)
-
-        // Average time per message should remain reasonable even under memory pressure
-        val avgTimePerMessage = duration.toDouble() / largeBatchSize
-        assertThat(avgTimePerMessage).isLessThan(25.0)
-    }
-
-    @Test
-    fun `should respect batch concurrency limits`() {
-        // Test that batch processing respects configured concurrency
-        val batchSize = 300
-        val testBatch = (1..batchSize).map { i ->
-            "concurrency_key_$i" to PerformanceTestEvent("Concurrency test message $i", i)
-        }
-
-        val startTime = System.currentTimeMillis()
-
-        StepVerifier.create(kafkaEventPublisher.publishEvents(testTopic, testBatch))
-            .expectNextCount(batchSize.toLong())
-            .verifyComplete()
-
-        val duration = System.currentTimeMillis() - startTime
-
-        // Should complete efficiently with controlled concurrency (within 20 seconds)
-        assertThat(duration).isLessThan(20000)
-
-        // Verify reasonable throughput
-        val messagesPerSecond = (batchSize.toDouble() / duration) * 1000
-        assertThat(messagesPerSecond).isGreaterThan(10.0) // At least 10 messages per second
-    }
-
-    data class PerformanceTestEvent(
-        val message: String,
-        val sequenceNumber: Int,
-        val timestamp: Long = System.currentTimeMillis()
-    )
-}
@@ -1,6 +1,5 @@
 package at.mocode.infrastructure.messaging.client

-import io.mockk.clearMocks
 import io.mockk.every
 import io.mockk.mockk
 import io.mockk.verify
@@ -11,9 +10,6 @@ import org.springframework.kafka.core.reactive.ReactiveKafkaProducerTemplate
 import reactor.core.publisher.Mono
 import reactor.kafka.sender.SenderResult
 import reactor.test.StepVerifier
-import java.io.IOException
-import java.net.ConnectException
-import java.util.concurrent.TimeoutException

@TestInstance(TestInstance.Lifecycle.PER_CLASS)
 class KafkaEventPublisherErrorTest {
@@ -28,224 +24,83 @@ class KafkaEventPublisherErrorTest {
    }

    @Test
-    fun `should retry on transient timeout errors`() {
+    fun `should publish single event successfully`() {
        val testEvent = TestEvent("data")
        val mockResult = mockk<SenderResult<Void>>()
        val mockRecordMetadata = mockk<org.apache.kafka.clients.producer.RecordMetadata>()
        every { mockRecordMetadata.topic() } returns "test-topic"
+        every { mockRecordMetadata.partition() } returns 0
+        every { mockRecordMetadata.offset() } returns 0L
        every { mockResult.recordMetadata() } returns mockRecordMetadata

-        // The first call fails with timeout, the second succeeds
-        every { mockTemplate.send("test-topic", "key", testEvent) } returns
-            Mono.error(TimeoutException("Connection timeout")) andThen
-            Mono.just(mockResult)
+        every { mockTemplate.send("test-topic", "key", testEvent) } returns Mono.just(mockResult)

-        StepVerifier.create(publisher.publishEvent("test-topic", "key", testEvent))
+        StepVerifier.create(publisher.publishEventReactive("test-topic", "key", testEvent))
+            .expectNext(Unit)
            .verifyComplete()

-        verify(exactly = 2) { mockTemplate.send("test-topic", "key", testEvent) }
+        verify(exactly = 1) { mockTemplate.send("test-topic", "key", testEvent) }
    }

    @Test
-    fun `should retry on connection errors`() {
-        val testEvent = TestEvent("data")
-        val mockResult = mockk<SenderResult<Void>>()
-        val mockRecordMetadata = mockk<org.apache.kafka.clients.producer.RecordMetadata>()
-        every { mockRecordMetadata.topic() } returns "test-topic"
-        every { mockResult.recordMetadata() } returns mockRecordMetadata
-
-        // First call fails with connection error, second succeeds
-        every { mockTemplate.send("test-topic", "key", testEvent) } returns
-            Mono.error(ConnectException("Connection refused")) andThen
-            Mono.just(mockResult)
-
-        StepVerifier.create(publisher.publishEvent("test-topic", "key", testEvent))
-            .verifyComplete()
-
-        verify(exactly = 2) { mockTemplate.send("test-topic", "key", testEvent) }
-    }
-
-    @Test
-    fun `should retry on IO errors`() {
-        val testEvent = TestEvent("data")
-        val mockResult = mockk<SenderResult<Void>>()
-        val mockRecordMetadata = mockk<org.apache.kafka.clients.producer.RecordMetadata>()
-        every { mockRecordMetadata.topic() } returns "test-topic"
-        every { mockResult.recordMetadata() } returns mockRecordMetadata
-
-        // First call fails with IOException, second succeeds
-        every { mockTemplate.send("test-topic", "key", testEvent) } returns
-            Mono.error(IOException("Network error")) andThen
-            Mono.just(mockResult)
-
-        StepVerifier.create(publisher.publishEvent("test-topic", "key", testEvent))
-            .verifyComplete()
-
-        verify(exactly = 2) { mockTemplate.send("test-topic", "key", testEvent) }
-    }
-
-    @Test
-    fun `should not retry on serialization errors`() {
+    fun `should handle serialization errors without retry`() {
        val testEvent = TestEvent("data")

        every { mockTemplate.send("test-topic", "key", testEvent) } returns
            Mono.error(RuntimeException("Serialization failed"))

-        StepVerifier.create(publisher.publishEvent("test-topic", "key", testEvent))
+        StepVerifier.create(publisher.publishEventReactive("test-topic", "key", testEvent))
            .verifyError(RuntimeException::class.java)

-        // Should only try once, no retries for serialization errors
        verify(exactly = 1) { mockTemplate.send("test-topic", "key", testEvent) }
    }

    @Test
-    fun `should not retry on authentication errors`() {
+    fun `should handle authentication errors without retry`() {
        val testEvent = TestEvent("data")

        every { mockTemplate.send("test-topic", "key", testEvent) } returns
            Mono.error(RuntimeException("Authentication failed"))

-        StepVerifier.create(publisher.publishEvent("test-topic", "key", testEvent))
+        StepVerifier.create(publisher.publishEventReactive("test-topic", "key", testEvent))
            .verifyError(RuntimeException::class.java)

-        // Should only try once, no retries for auth errors
        verify(exactly = 1) { mockTemplate.send("test-topic", "key", testEvent) }
    }

-    @Test
-    fun `should exhaust retries and fail after maximum attempts`() {
-        val testEvent = TestEvent("data")
-
-        // Always fail with retryable error
-        every { mockTemplate.send("test-topic", "key", testEvent) } returns
-            Mono.error(TimeoutException("Connection timeout"))
-
-        StepVerifier.create(publisher.publishEvent("test-topic", "key", testEvent))
-            .verifyError(TimeoutException::class.java)
-
-        // Should try 1 initial + 3 retries = 4 times total
-        verify(exactly = 4) { mockTemplate.send("test-topic", "key", testEvent) }
-    }
-
-    @Test
-    fun `should handle batch publishing with partial failures`() {
-        val events = listOf(
-            "key1" to TestEvent("success1"),
-            "key2" to TestEvent("failure"),
-            "key3" to TestEvent("success2")
-        )
-
-        val mockResult = mockk<SenderResult<Void>>()
-        val mockRecordMetadata = mockk<org.apache.kafka.clients.producer.RecordMetadata>()
-        every { mockRecordMetadata.topic() } returns "test-topic"
-        every { mockResult.recordMetadata() } returns mockRecordMetadata
-
-        // First and third events succeed, second fails
-        every { mockTemplate.send("test-topic", "key1", any()) } returns Mono.just(mockResult)
-        every { mockTemplate.send("test-topic", "key2", any()) } returns
-            Mono.error(RuntimeException("Serialization failed"))
-        every { mockTemplate.send("test-topic", "key3", any()) } returns Mono.just(mockResult)
-
-        StepVerifier.create(publisher.publishEvents("test-topic", events))
-            .expectNextCount(2) // Should complete 2 successful sends
-            .verifyComplete()
-
-        // Verify all events were attempted
-        verify(exactly = 1) { mockTemplate.send("test-topic", "key1", any()) }
-        verify(exactly = 1) { mockTemplate.send("test-topic", "key2", any()) }
-        verify(exactly = 1) { mockTemplate.send("test-topic", "key3", any()) }
-    }
-
-    @Test
-    fun `should handle batch publishing with retryable failures`() {
-        val events = listOf(
-            "key1" to TestEvent("success"),
-            "key2" to TestEvent("retry-then-success")
-        )
-
-        val mockResult = mockk<SenderResult<Void>>()
-        val mockRecordMetadata = mockk<org.apache.kafka.clients.producer.RecordMetadata>()
-        every { mockRecordMetadata.topic() } returns "test-topic"
-        every { mockResult.recordMetadata() } returns mockRecordMetadata
-
-        // First event succeeds immediately
-        every { mockTemplate.send("test-topic", "key1", any()) } returns Mono.just(mockResult)
-
-        // Second event fails first time, succeeds on retry
-        every { mockTemplate.send("test-topic", "key2", any()) } returns
-            Mono.error(TimeoutException("Connection timeout")) andThen
-            Mono.just(mockResult)
-
-        StepVerifier.create(publisher.publishEvents("test-topic", events))
-            .expectNextCount(2) // Should complete both events
-            .verifyComplete()
-
-        // First event called once, second event called twice (initial + retry)
-        verify(exactly = 1) { mockTemplate.send("test-topic", "key1", any()) }
-        verify(exactly = 2) { mockTemplate.send("test-topic", "key2", any()) }
-    }
-
    @Test
    fun `should handle empty batch gracefully`() {
        val emptyEvents = emptyList<Pair<String?, Any>>()

-        StepVerifier.create(publisher.publishEvents("test-topic", emptyEvents))
+        StepVerifier.create(publisher.publishEventsReactive("test-topic", emptyEvents))
            .verifyComplete()

-        // Should not call the template at all
        verify(exactly = 0) { mockTemplate.send(any(), any(), any()) }
    }

    @Test
-    fun `should identify retryable exceptions correctly`() {
-        // Test the private isRetryableException method through behavior
-        val testEvent = TestEvent("data")
-
-        // Test various error messages that should be retryable
-        val retryableErrors = listOf(
-            RuntimeException("timeout occurred"),
-            RuntimeException("connection refused"),
-            RuntimeException("network unreachable"),
-            TimeoutException("Request timeout"),
-            ConnectException("Connection failed"),
-            IOException("I/O error")
+    fun `should publish batch events successfully`() {
+        val events = listOf(
+            "key1" to TestEvent("message1"),
+            "key2" to TestEvent("message2")
        )

-        retryableErrors.forEach { error ->
-            clearMocks(mockTemplate)
-            every { mockTemplate.send("test-topic", "key", testEvent) } returns
-                Mono.error(error) andThen Mono.error(error) // Fail twice to test retry
+        val mockResult = mockk<SenderResult<Void>>()
+        val mockRecordMetadata = mockk<org.apache.kafka.clients.producer.RecordMetadata>()
+        every { mockRecordMetadata.topic() } returns "test-topic"
+        every { mockRecordMetadata.partition() } returns 0
+        every { mockRecordMetadata.offset() } returns 0L
+        every { mockResult.recordMetadata() } returns mockRecordMetadata

-            StepVerifier.create(publisher.publishEvent("test-topic", "key", testEvent))
-                .verifyError()
+        every { mockTemplate.send("test-topic", "key1", any()) } returns Mono.just(mockResult)
+        every { mockTemplate.send("test-topic", "key2", any()) } returns Mono.just(mockResult)

-            // Should retry (at least 2 calls)
-            verify(atLeast = 2) { mockTemplate.send("test-topic", "key", testEvent) }
-        }
-    }
+        StepVerifier.create(publisher.publishEventsReactive("test-topic", events))
+            .expectNextCount(2)
+            .verifyComplete()

-    @Test
-    fun `should identify non-retryable exceptions correctly`() {
-        val testEvent = TestEvent("data")
-
-        // Test various error messages that should NOT be retryable
-        val nonRetryableErrors = listOf(
-            RuntimeException("serialization error"),
-            RuntimeException("deserialization failed"),
-            RuntimeException("authentication failed"),
-            RuntimeException("authorization denied")
-        )
-
-        nonRetryableErrors.forEach { error ->
-            clearMocks(mockTemplate)
-            every { mockTemplate.send("test-topic", "key", testEvent) } returns Mono.error(error)
-
-            StepVerifier.create(publisher.publishEvent("test-topic", "key", testEvent))
-                .verifyError()
-
-            // Should NOT retry (exactly 1 call)
-            verify(exactly = 1) { mockTemplate.send("test-topic", "key", testEvent) }
-        }
+        verify(exactly = 1) { mockTemplate.send("test-topic", "key1", any()) }
+        verify(exactly = 1) { mockTemplate.send("test-topic", "key2", any()) }
    }

    data class TestEvent(val message: String)
@@ -1,6 +1,5 @@
 package at.mocode.infrastructure.messaging.client

-import at.mocode.infrastructure.messaging.client.ReactiveKafkaConfig
 import at.mocode.infrastructure.messaging.config.KafkaConfig
 import org.apache.kafka.common.serialization.StringDeserializer
 import org.junit.jupiter.api.AfterEach
@@ -77,7 +76,7 @@ class KafkaIntegrationTest {
            .map { it.value() } // Extract the value (our TestEvent instance)

        // The Mono that represents the send action
-        val sendAction = kafkaEventPublisher.publishEvent(testTopic, testKey, testEvent)
+        val sendAction = kafkaEventPublisher.publishEventReactive(testTopic, testKey, testEvent)

        // CORRECTION: Combine the send action and receive expectation in one StepVerifier.
        // The `then` method ensures that the send action is completed first,
@@ -119,7 +118,7 @@ class KafkaIntegrationTest {
            .collectList()

        // Send batch and verify reception
-        val sendAction = kafkaEventPublisher.publishEvents(testTopic, eventBatch)
+        val sendAction = kafkaEventPublisher.publishEventsReactive(testTopic, eventBatch)

        StepVerifier.create(sendAction.then(receivedEvents))
            .expectNextMatches { events ->
@@ -171,7 +170,7 @@ class KafkaIntegrationTest {
            .next()
            .map { it.value() }

-        val sendAction = kafkaEventPublisher.publishEvent(testTopic, testKey, testEvent)
+        val sendAction = kafkaEventPublisher.publishEventReactive(testTopic, testKey, testEvent)

        // Both consumers should receive the same message (different groups)
        StepVerifier.create(sendAction.then(consumer1Event.zipWith(consumer2Event)))
@@ -210,7 +209,7 @@ class KafkaIntegrationTest {
            .next()
            .map { it.value() }

-        val sendAction = kafkaEventPublisher.publishEvent(testTopic, "complex-key", complexEvent)
+        val sendAction = kafkaEventPublisher.publishEventReactive(testTopic, "complex-key", complexEvent)

        StepVerifier.create(sendAction.then(receivedEvent))
            .expectNext(complexEvent)
@@ -246,7 +245,7 @@ class KafkaIntegrationTest {
            .map { it.value() }
            .collectList()

-        val sendAction = kafkaEventPublisher.publishEvents(testTopic, orderedEvents)
+        val sendAction = kafkaEventPublisher.publishEventsReactive(testTopic, orderedEvents)

        StepVerifier.create(sendAction.then(receivedEvents))
            .expectNextMatches { events ->
@@ -262,7 +261,7 @@ class KafkaIntegrationTest {
    fun `should handle empty batch gracefully in integration test`() {
        val emptyBatch = emptyList<Pair<String?, Any>>()

-        StepVerifier.create(kafkaEventPublisher.publishEvents(testTopic, emptyBatch))
+        StepVerifier.create(kafkaEventPublisher.publishEventsReactive(testTopic, emptyBatch))
            .verifyComplete()
    }

@@ -1,6 +1,5 @@
 package at.mocode.infrastructure.messaging.client

-import at.mocode.infrastructure.messaging.client.ReactiveKafkaConfig
 import at.mocode.infrastructure.messaging.config.KafkaConfig
 import org.apache.kafka.clients.consumer.ConsumerConfig
 import org.apache.kafka.clients.producer.ProducerConfig
@@ -1,376 +0,0 @@
-package at.mocode.infrastructure.messaging.client
-
-import at.mocode.infrastructure.messaging.client.ReactiveKafkaConfig
-import at.mocode.infrastructure.messaging.config.KafkaConfig
-import io.mockk.every
-import io.mockk.mockk
-import io.mockk.verify
-import org.assertj.core.api.Assertions.assertThat
-import org.junit.jupiter.api.AfterEach
-import org.junit.jupiter.api.BeforeEach
-import org.junit.jupiter.api.Test
-import org.junit.jupiter.api.TestInstance
-import org.junit.jupiter.api.assertDoesNotThrow
-import org.slf4j.LoggerFactory
-import org.springframework.kafka.core.reactive.ReactiveKafkaProducerTemplate
-import reactor.core.publisher.Mono
-import reactor.kafka.sender.SenderResult
-import reactor.test.StepVerifier
-import java.io.ByteArrayOutputStream
-import java.io.IOException
-import java.io.PrintStream
-import java.net.ConnectException
-import java.util.concurrent.TimeoutException
-
-@TestInstance(TestInstance.Lifecycle.PER_CLASS)
-class LoggingAndMonitoringTest {
-
-    private val logger = LoggerFactory.getLogger(LoggingAndMonitoringTest::class.java)
-
-    private lateinit var kafkaConfig: KafkaConfig
-    private lateinit var consumer: KafkaEventConsumer
-    private lateinit var originalOut: PrintStream
-    private lateinit var testOutput: ByteArrayOutputStream
-
-    @BeforeEach
-    fun setUp() {
-        kafkaConfig = KafkaConfig().apply {
-            bootstrapServers = "localhost:9092"
-            defaultGroupIdPrefix = "logging-test-consumer"
-            trustedPackages = "at.mocode.*"
-        }
-        consumer = KafkaEventConsumer(kafkaConfig)
-
-        // Capture console output for log verification
-        originalOut = System.out
-        testOutput = ByteArrayOutputStream()
-        System.setOut(PrintStream(testOutput))
-    }
-
-    @Test
-    fun `should log structured information for consumer setup`() {
-        // Create consumer and set up stream - this should generate log entries
-        assertDoesNotThrow {
-            val flux = consumer.receiveEvents<LoggingTestEvent>("structured-logging-topic")
-            assertThat(flux).isNotNull
-        }
-
-        // In a real implementation, we would verify specific log entries
-        // For now, we verify that the setup completes without errors
-        val output = testOutput.toString()
-
-        // Basic verification that some logging occurred (setup methods would generate logs)
-        assertThat(output).isNotNull
-
-        logger.debug("Consumer setup completed successfully")
-    }
-
-    @Test
-    fun `should log retry attempts with context information`() {
-        val mockTemplate = mockk<ReactiveKafkaProducerTemplate<String, Any>>()
-        val publisher = KafkaEventPublisher(mockTemplate)
-        val testEvent = LoggingTestEvent("retry-test", 1)
-
-        // Configure mock to fail the first few times, then succeed
-        every { mockTemplate.send("retry-topic", "retry-key", testEvent) } returns
-            Mono.error(TimeoutException("Connection timeout")) andThen
-            Mono.error(ConnectException("Connection refused")) andThen
-            Mono.just(mockk<SenderResult<Void>>())
-
-        StepVerifier.create(publisher.publishEvent("retry-topic", "retry-key", testEvent))
-            .verifyComplete()
-
-        // Verify retry attempts were logged
-        logger.debug("Retry logging test completed")
-        assertThat(testOutput.toString()).isNotNull
-
-        verify(exactly = 3) { mockTemplate.send("retry-topic", "retry-key", testEvent) }
-    }
-
-    @Test
-    fun `should track batch operation progress`() {
-        val mockTemplate = mockk<ReactiveKafkaProducerTemplate<String, Any>>()
-        val publisher = KafkaEventPublisher(mockTemplate)
-
-        // Create a medium-sized batch to trigger progress logging
-        val batchSize = 250 // This should trigger progress logging at 100, 200, and final
-        val testBatch = (1..batchSize).map { i ->
-            "batch_key_$i" to LoggingTestEvent("Batch message $i", i)
-        }
-
-        val mockResult = mockk<SenderResult<Void>>()
-        val mockRecordMetadata = mockk<org.apache.kafka.clients.producer.RecordMetadata>()
-        every { mockRecordMetadata.topic() } returns "batch-progress-topic"
-        every { mockRecordMetadata.partition() } returns 0
-        every { mockRecordMetadata.offset() } returns 0L
-        every { mockResult.recordMetadata() } returns mockRecordMetadata
-        every { mockTemplate.send(any(), any(), any()) } returns Mono.just(mockResult)
-
-        StepVerifier.create(publisher.publishEvents("batch-progress-topic", testBatch))
-            .expectNextCount(batchSize.toLong())
-            .verifyComplete()
-
-        logger.debug("Batch progress tracking test completed with {} events", batchSize)
-
-        // Verify that all batch items were processed
-        verify(exactly = batchSize) { mockTemplate.send(any(), any(), any()) }
-    }
-
-    @Test
-    fun `should log error context for failed operations`() {
-        val mockTemplate = mockk<ReactiveKafkaProducerTemplate<String, Any>>()
-        val publisher = KafkaEventPublisher(mockTemplate)
-        val testEvent = LoggingTestEvent("error-context", 1)
-
-        // Configure mock to always fail
-        every { mockTemplate.send("error-topic", "error-key", testEvent) } returns
-            Mono.error(IOException("Network failure"))
-
-        StepVerifier.create(publisher.publishEvent("error-topic", "error-key", testEvent))
-            .verifyError(IOException::class.java)
-
-        logger.debug("Error context logging test completed")
-
-        // Should have attempted the operation and logged error context
-        verify(atLeast = 1) { mockTemplate.send("error-topic", "error-key", testEvent) }
-    }
-
-    @Test
-    fun `should log performance metrics for operations`() {
-        val mockTemplate = mockk<ReactiveKafkaProducerTemplate<String, Any>>()
-        val publisher = KafkaEventPublisher(mockTemplate)
-        val testEvents = (1..50).map { i ->
-            "perf_key_$i" to LoggingTestEvent("Performance test $i", i)
-        }
-
-        val mockResult = mockk<SenderResult<Void>>()
-        val mockRecordMetadata = mockk<org.apache.kafka.clients.producer.RecordMetadata>()
-        every { mockRecordMetadata.topic() } returns "performance-metrics-topic"
-        every { mockRecordMetadata.partition() } returns 0
-        every { mockRecordMetadata.offset() } returns 0L
-        every { mockResult.recordMetadata() } returns mockRecordMetadata
-        every { mockTemplate.send(any(), any(), any()) } returns Mono.just(mockResult)
-
-        val startTime = System.currentTimeMillis()
-
-        StepVerifier.create(publisher.publishEvents("performance-metrics-topic", testEvents))
-            .expectNextCount(50)
-            .verifyComplete()
-
-        val duration = System.currentTimeMillis() - startTime
-
-        logger.debug("Performance metrics: 50 events published in {}ms", duration)
-        logger.debug("Average time per event: {}ms", duration.toDouble() / 50)
-
-        // Performance should be reasonable
-        assertThat(duration).isLessThan(10000) // Within 10 seconds
-    }
-
-    @Test
-    fun `should log consumer group and partition information`() {
-        // Create consumer flux - this should generate group ID and partition logs
-        val flux = consumer.receiveEvents<LoggingTestEvent>("partition-info-topic")
-
-        // The act of creating the flux should generate logging about group assignment
-        assertThat(flux).isNotNull
-
-        logger.debug("Consumer group and partition logging test completed")
-        logger.debug("Expected group ID pattern: {}-partition-info-topic-loggingtesteevent", kafkaConfig.defaultGroupIdPrefix)
-
-        // Verify consumer was created successfully
-        assertDoesNotThrow {
-            consumer.cleanup()
-        }
-    }
-
-    @Test
-    fun `should log different event types with structured information`() {
-        val mockTemplate = mockk<ReactiveKafkaProducerTemplate<String, Any>>()
-        val publisher = KafkaEventPublisher(mockTemplate)
-
-        // Test with different event types
-        val mockResult = mockk<SenderResult<Void>>()
-        every { mockTemplate.send(any(), any(), any()) } returns Mono.just(mockResult)
-
-        val testEvents = listOf(
-            LoggingTestEvent("string event", 1),
-            ComplexLoggingEvent("complex", 123, mapOf("key" to "value")),
-            NumericLoggingEvent(42, 3.14, System.currentTimeMillis())
-        )
-
-        testEvents.forEachIndexed { index, event ->
-            StepVerifier.create(publisher.publishEvent("event-types-topic", "key_$index", event))
-                .verifyComplete()
-
-            logger.debug("Published event type: {}", event::class.simpleName)
-        }
-
-        verify(exactly = testEvents.size) { mockTemplate.send(any(), any(), any()) }
-    }
-
-    @Test
-    fun `should log retry exhaustion with final error details`() {
-        val mockTemplate = mockk<ReactiveKafkaProducerTemplate<String, Any>>()
-        val publisher = KafkaEventPublisher(mockTemplate)
-        val testEvent = LoggingTestEvent("retry-exhaustion", 1)
-
-        // Configure mock to always fail with retryable error
-        every { mockTemplate.send("exhaustion-topic", "exhaustion-key", testEvent) } returns
-            Mono.error(TimeoutException("Persistent timeout"))
-
-        StepVerifier.create(publisher.publishEvent("exhaustion-topic", "exhaustion-key", testEvent))
-            .verifyError(TimeoutException::class.java)
-
-        logger.debug("Retry exhaustion logging test completed")
-
-        // Should have attempted maximum retries (1 initial + 3 retries = 4 total)
-        verify(exactly = 4) { mockTemplate.send("exhaustion-topic", "exhaustion-key", testEvent) }
-    }
-
-    @Test
-    fun `should log startup and configuration information`() {
-        // Test that consumer startup logs configuration details
-        val customConfig = KafkaConfig().apply {
-            bootstrapServers = "test-server:9092"
-            defaultGroupIdPrefix = "config-logging-test"
-            trustedPackages = "at.mocode.*,com.test.*"
-            enableSecurityFeatures = true
-            connectionPoolSize = 15
-        }
-
-        val customConsumer = KafkaEventConsumer(customConfig)
-        val customReactiveConfig = ReactiveKafkaConfig(customConfig)
-
-        assertDoesNotThrow {
-            val template = customReactiveConfig.reactiveKafkaProducerTemplate()
-            assertThat(template).isNotNull
-        }
-
-        logger.debug("Configuration logging test completed")
-        logger.debug("Bootstrap servers: {}", customConfig.bootstrapServers)
-        logger.debug("Group ID prefix: {}", customConfig.defaultGroupIdPrefix)
-        logger.debug("Trusted packages: {}", customConfig.trustedPackages)
-        logger.debug("Security features enabled: {}", customConfig.enableSecurityFeatures)
-        logger.debug("Connection pool size: {}", customConfig.connectionPoolSize)
-
-        customConsumer.cleanup()
-    }
-
-    @Test
-    fun `should log resource cleanup operations`() {
-        val tempConsumer = KafkaEventConsumer(kafkaConfig)
-
-        // Create some reactive streams to establish resources
-        val flux1 = tempConsumer.receiveEvents<LoggingTestEvent>("cleanup-topic-1")
-        val flux2 = tempConsumer.receiveEvents<LoggingTestEvent>("cleanup-topic-2")
-
-        assertThat(flux1).isNotNull
-        assertThat(flux2).isNotNull
-
-        logger.debug("Resources created for cleanup test")
-
-        // Cleanup should log resource cleanup operations
-        assertDoesNotThrow {
-            tempConsumer.cleanup()
-        }
-
-        logger.debug("Resource cleanup test completed")
-    }
-
-    @Test
-    fun `should handle logging under concurrent access`() {
-        val mockTemplate = mockk<ReactiveKafkaProducerTemplate<String, Any>>()
-        val publisher = KafkaEventPublisher(mockTemplate)
-        val mockResult = mockk<SenderResult<Void>>()
-        val mockRecordMetadata = mockk<org.apache.kafka.clients.producer.RecordMetadata>()
-        every { mockRecordMetadata.topic() } returns "concurrent-logging-topic"
-        every { mockRecordMetadata.partition() } returns 0
-        every { mockRecordMetadata.offset() } returns 0L
-        every { mockResult.recordMetadata() } returns mockRecordMetadata
-
-        every { mockTemplate.send(any(), any(), any()) } returns Mono.just(mockResult)
-
-        // Create concurrent publishing operations
-        val concurrentEvents = (1..20).map { i ->
-            publisher.publishEvent("concurrent-logging-topic", "concurrent_key_$i",
-                LoggingTestEvent("Concurrent message $i", i))
-        }
-
-        StepVerifier.create(reactor.core.publisher.Flux.merge(concurrentEvents))
-            .expectNextCount(20)
-            .verifyComplete()
-
-        logger.debug("Concurrent logging test completed with 20 concurrent operations")
-
-        verify(exactly = 20) { mockTemplate.send(any(), any(), any()) }
-    }
-
-    @Test
-    fun `should log timestamp and correlation information`() {
-        val mockTemplate = mockk<ReactiveKafkaProducerTemplate<String, Any>>()
-        val publisher = KafkaEventPublisher(mockTemplate)
-        val mockResult = mockk<SenderResult<Void>>()
-
-        every { mockTemplate.send(any(), any(), any()) } returns Mono.just(mockResult)
-
-        val timestampedEvent = LoggingTestEvent("timestamped", 1)
-
-        val beforePublish = System.currentTimeMillis()
-
-        StepVerifier.create(publisher.publishEvent("timestamp-topic", "timestamp-key", timestampedEvent))
-            .verifyComplete()
-
-        val afterPublish = System.currentTimeMillis()
-
-        logger.debug("Event published with timestamp correlation")
-        logger.debug("Publish window: {} to {} ({}ms)", beforePublish, afterPublish, afterPublish - beforePublish)
-
-        verify(exactly = 1) { mockTemplate.send("timestamp-topic", "timestamp-key", timestampedEvent) }
-    }
-
-    @Test
-    fun `should provide debug information for troubleshooting`() {
-        // Create various configurations and operations to generate debug logs
-        val debugConfig = KafkaConfig().apply {
-            bootstrapServers = "debug-server:9092"
-            defaultGroupIdPrefix = "debug-test"
-        }
-
-        val debugConsumer = KafkaEventConsumer(debugConfig)
-        val debugFlux = debugConsumer.receiveEvents<LoggingTestEvent>("debug-topic")
-
-        logger.debug("Debug configuration created")
-        logger.debug("Consumer group ID would be: debug-test-debug-topic-loggingtesteevent")
-        logger.debug("Bootstrap servers: debug-server:9092")
-
-        assertThat(debugFlux).isNotNull
-
-        debugConsumer.cleanup()
-        logger.debug("Debug cleanup completed")
-    }
-
-    @AfterEach
-    fun tearDown() {
-        // Restore original output
-        System.setOut(originalOut)
-        consumer.cleanup()
-    }
-
-    data class LoggingTestEvent(
-        val message: String,
-        val sequenceNumber: Int,
-        val timestamp: Long = System.currentTimeMillis()
-    )
-
-    data class ComplexLoggingEvent(
-        val name: String,
-        val id: Int,
-        val metadata: Map<String, String>
-    )
-
-    data class NumericLoggingEvent(
-        val intValue: Int,
-        val doubleValue: Double,
-        val timestamp: Long
-    )
-}
@@ -1,365 +0,0 @@
-package at.mocode.infrastructure.messaging.client
-
-import at.mocode.infrastructure.messaging.config.KafkaConfig
-import org.assertj.core.api.Assertions.assertThat
-import org.junit.jupiter.api.BeforeEach
-import org.junit.jupiter.api.Test
-import org.junit.jupiter.api.TestInstance
-import org.junit.jupiter.api.assertDoesNotThrow
-import org.slf4j.LoggerFactory
-import reactor.core.publisher.Flux
-import reactor.core.publisher.Mono
-import reactor.core.scheduler.Schedulers
-import reactor.test.StepVerifier
-import reactor.test.publisher.TestPublisher
-import java.time.Duration
-import java.util.concurrent.CountDownLatch
-import java.util.concurrent.TimeUnit
-import java.util.concurrent.atomic.AtomicInteger
-import java.util.concurrent.atomic.AtomicLong
-
-@TestInstance(TestInstance.Lifecycle.PER_CLASS)
-class ReactiveStreamTest {
-
-    private val logger = LoggerFactory.getLogger(ReactiveStreamTest::class.java)
-
-    private lateinit var kafkaConfig: KafkaConfig
-    private lateinit var consumer: KafkaEventConsumer
-
-    @BeforeEach
-    fun setUp() {
-        kafkaConfig = KafkaConfig().apply {
-            bootstrapServers = "localhost:9092"
-            defaultGroupIdPrefix = "reactive-test-consumer"
-            trustedPackages = "at.mocode.*"
-        }
-        consumer = KafkaEventConsumer(kafkaConfig)
-    }
-
-    @Test
-    fun `should create cold streams that start on subscription`() {
-        // Cold streams should not start processing until subscribed
-        val flux = consumer.receiveEvents<ReactiveTestEvent>("cold-stream-topic")
-
-        // Stream should be created but not started
-        assertThat(flux).isNotNull
-
-        // No subscription means no processing should begin.
-        // This is verified by the fact that creating the flux doesn't throw or block
-        assertDoesNotThrow {
-            val anotherFlux = consumer.receiveEvents<ReactiveTestEvent>("another-cold-topic")
-            assertThat(anotherFlux).isNotNull
-        }
-    }
-
-    @Test
-    fun `should handle multiple subscribers to same stream`() {
-        val flux = consumer.receiveEvents<ReactiveTestEvent>("multi-subscriber-topic")
-
-        // Multiple subscribers should be able to subscribe to the same flux
-        val subscriber1 = StepVerifier.create(flux.take(1).timeout(Duration.ofSeconds(2)))
-        val subscriber2 = StepVerifier.create(flux.take(1).timeout(Duration.ofSeconds(2)))
-
-        // Both subscribers should be created without issues
-        // Note: In real Kafka usage, each subscriber would get their own consumer group
-        assertDoesNotThrow {
-            subscriber1.thenCancel().verify(Duration.ofSeconds(1))
-            subscriber2.thenCancel().verify(Duration.ofSeconds(1))
-        }
-    }
-
-    @Test
-    fun `should support reactive operators and transformations`() {
-        val flux = consumer.receiveEvents<ReactiveTestEvent>("transformation-topic")
-
-        // Apply various reactive operators
-        val transformedFlux = flux
-            .filter { event -> event.message.contains("important") }
-            .map { event -> event.message.uppercase() }
-            .distinctUntilChanged()
-            .take(5)
-
-        assertThat(transformedFlux).isNotNull
-
-        // Should be able to subscribe to transformed flux
-        val verifier = StepVerifier.create(transformedFlux.timeout(Duration.ofSeconds(2)))
-        assertDoesNotThrow {
-            verifier.thenCancel().verify(Duration.ofSeconds(1))
-        }
-    }
-
-    @Test
-    fun `should handle backpressure gracefully`() {
-        val flux = consumer.receiveEvents<ReactiveTestEvent>("backpressure-topic")
-
-        // Simulate slow consumer to test backpressure
-        val slowProcessingFlux = flux
-            .concatMap { event ->
-                Mono.delay(Duration.ofMillis(100))
-                    .map { event }
-            }
-            .take(3)
-
-        val startTime = System.currentTimeMillis()
-
-        StepVerifier.create(slowProcessingFlux.timeout(Duration.ofSeconds(5)))
-            .thenCancel()
-            .verify(Duration.ofSeconds(2))
-
-        val duration = System.currentTimeMillis() - startTime
-
-        // Should handle backpressure without blocking indefinitely
-        assertThat(duration).isLessThan(3000)
-    }
-
-    @Test
-    fun `should maintain stream characteristics under error conditions`() {
-        val flux = consumer.receiveEvents<ReactiveTestEvent>("error-resilience-topic")
-
-        // Add error handling and recovery
-        val resilientFlux = flux
-            .onErrorResume { error ->
-                // Log error and continue with an empty stream
-                logger.debug("Handled error in stream: {}", error.message)
-                Flux.empty()
-            }
-            .retry(2)
-            .take(1)
-
-        StepVerifier.create(resilientFlux.timeout(Duration.ofSeconds(3)))
-            .thenCancel()
-            .verify(Duration.ofSeconds(2))
-
-        // Stream should remain reactive even after error handling
-        assertThat(resilientFlux).isNotNull
-    }
-
-    @Test
-    fun `should support concurrent stream processing`() {
-        val flux1 = consumer.receiveEvents<ReactiveTestEvent>("concurrent-topic-1")
-        val flux2 = consumer.receiveEvents<ReactiveTestEvent>("concurrent-topic-2")
-        val flux3 = consumer.receiveEvents<ReactiveTestEvent>("concurrent-topic-3")
-
-        // Process multiple streams concurrently
-        val combinedFlux = Flux.merge(
-            flux1.subscribeOn(Schedulers.parallel()),
-            flux2.subscribeOn(Schedulers.parallel()),
-            flux3.subscribeOn(Schedulers.parallel())
-        ).take(3)
-
-        StepVerifier.create(combinedFlux.timeout(Duration.ofSeconds(3)))
-            .thenCancel()
-            .verify(Duration.ofSeconds(2))
-
-        // All streams should be processable concurrently
-        assertThat(combinedFlux).isNotNull
-    }
-
-    @Test
-    fun `should handle stream lifecycle correctly`() {
-        val eventCounter = AtomicInteger(0)
-        val flux = consumer.receiveEvents<ReactiveTestEvent>("lifecycle-topic")
-
-        // Add lifecycle monitoring
-        val monitoredFlux = flux
-            .doOnSubscribe { subscription ->
-                logger.debug("Stream subscribed: {}", subscription)
-            }
-            .doOnNext { event ->
-                val count = eventCounter.incrementAndGet()
-                logger.debug("Processed event #{}: {}", count, event.message)
-            }
-            .doOnCancel {
-                logger.debug("Stream cancelled")
-            }
-            .doOnComplete {
-                logger.debug("Stream completed")
-            }
-            .take(1)
-
-        StepVerifier.create(monitoredFlux.timeout(Duration.ofSeconds(2)))
-            .thenCancel()
-            .verify(Duration.ofSeconds(1))
-
-        // Lifecycle should be properly managed
-        assertThat(monitoredFlux).isNotNull
-    }
-
-    @Test
-    fun `should support flow control mechanisms`() {
-        val flux = consumer.receiveEvents<ReactiveTestEvent>("flow-control-topic")
-
-        // Apply various flow control mechanisms
-        val controlledFlux = flux
-            .limitRate(10) // Limit upstream requests
-            .sample(Duration.ofMillis(100)) // Sample at fixed intervals
-            .buffer(5) // Buffer elements
-            .flatMap { buffer ->
-                logger.debug("Processing buffer of size: {}", buffer.size)
-                Flux.fromIterable(buffer)
-            }
-            .take(5)
-
-        StepVerifier.create(controlledFlux.timeout(Duration.ofSeconds(3)))
-            .thenCancel()
-            .verify(Duration.ofSeconds(2))
-
-        assertThat(controlledFlux).isNotNull
-    }
-
-    @Test
-    fun `should handle time-based operations`() {
-        val flux = consumer.receiveEvents<ReactiveTestEvent>("time-based-topic")
-
-        // Apply time-based operations
-        val timedFlux = flux
-            .window(Duration.ofMillis(200)) // Window by time
-            .flatMap { window ->
-                window.collectList()
-                    .map { events ->
-                        logger.debug("Window contains {} events", events.size)
-                        events.size
-                    }
-            }
-            .take(2)
-
-        StepVerifier.create(timedFlux.timeout(Duration.ofSeconds(3)))
-            .thenCancel()
-            .verify(Duration.ofSeconds(2))
-
-        assertThat(timedFlux).isNotNull
-    }
-
-    @Test
-    fun `should maintain thread safety in reactive streams`() {
-        val flux = consumer.receiveEvents<ReactiveTestEvent>("thread-safety-topic")
-        val processedCount = AtomicLong(0)
-        val latch = CountDownLatch(3)
-
-        // Process on multiple threads
-        val threadSafeFlux = flux
-            .publishOn(Schedulers.parallel())
-            .doOnNext { event ->
-                val count = processedCount.incrementAndGet()
-                logger.debug("Thread {} processed event #{}", Thread.currentThread().name, count)
-                latch.countDown()
-            }
-            .take(3)
-
-        // Subscribe and wait briefly
-        val subscription = threadSafeFlux
-            .timeout(Duration.ofSeconds(2))
-            .subscribe(
-                { event -> /* processed */ },
-                { error -> logger.debug("Error: {}", error.message) },
-                { logger.debug("Stream completed") }
-            )
-
-        // Wait for brief processing or timeout
-        val completed = latch.await(1, TimeUnit.SECONDS)
-        subscription.dispose()
-
-        // Thread safety should be maintained (no exceptions thrown)
-        assertThat(subscription).isNotNull
-    }
-
-    @Test
-    fun `should support custom schedulers`() {
-        val flux = consumer.receiveEvents<ReactiveTestEvent>("scheduler-topic")
-
-        // Use different schedulers for different operations
-        val scheduledFlux = flux
-            .subscribeOn(Schedulers.boundedElastic()) // For I/O operations
-            .publishOn(Schedulers.parallel()) // For CPU-intensive operations
-            .map { event ->
-                logger.debug("Processing on thread: {}", Thread.currentThread().name)
-                event.message.length
-            }
-            .subscribeOn(Schedulers.single()) // Single-threaded subscription
-            .take(1)
-
-        StepVerifier.create(scheduledFlux.timeout(Duration.ofSeconds(2)))
-            .thenCancel()
-            .verify(Duration.ofSeconds(1))
-
-        assertThat(scheduledFlux).isNotNull
-    }
-
-    @Test
-    fun `should handle stream composition and chaining`() {
-        val flux1 = consumer.receiveEvents<ReactiveTestEvent>("composition-topic-1")
-        val flux2 = consumer.receiveEvents<ReactiveTestEvent>("composition-topic-2")
-
-        // Compose multiple streams
-        val composedFlux = flux1
-            .switchMap { event1 ->
-                flux2.map { event2 ->
-                    logger.debug("Composed: {} -> {}", event1.message, event2.message)
-                    "${event1.message}+${event2.message}"
-                }
-            }
-            .take(1)
-
-        StepVerifier.create(composedFlux.timeout(Duration.ofSeconds(2)))
-            .thenCancel()
-            .verify(Duration.ofSeconds(1))
-
-        assertThat(composedFlux).isNotNull
-    }
-
-    @Test
-    fun `should support reactive testing patterns`() {
-        val flux = consumer.receiveEvents<ReactiveTestEvent>("testing-patterns-topic")
-
-        // Use TestPublisher to simulate controlled event emission
-        val testPublisher = TestPublisher.create<ReactiveTestEvent>()
-        val testFlux = testPublisher.flux()
-
-        // Apply similar transformations as the real flux
-        val transformedTestFlux = testFlux
-            .filter { event -> event.message.isNotEmpty() }
-            .map { event -> event.message.length }
-
-        // Test with controlled emissions
-        StepVerifier.create(transformedTestFlux)
-            .then { testPublisher.next(ReactiveTestEvent("test", 1)) }
-            .expectNext(4) // "test".length
-            .then { testPublisher.complete() }
-            .verifyComplete()
-
-        // Real flux should also be testable
-        assertThat(flux).isNotNull
-    }
-
-    @Test
-    fun `should handle resource cleanup properly`() {
-        val flux = consumer.receiveEvents<ReactiveTestEvent>("cleanup-topic")
-        val resourcesAcquired = AtomicInteger(0)
-        val resourcesReleased = AtomicInteger(0)
-
-        val resourceManagedFlux = flux
-            .doOnSubscribe {
-                resourcesAcquired.incrementAndGet()
-                logger.debug("Resources acquired: {}", resourcesAcquired.get())
-            }
-            .doFinally { signalType ->
-                resourcesReleased.incrementAndGet()
-                logger.debug("Resources released on {}: {}", signalType, resourcesReleased.get())
-            }
-            .take(1)
-
-        StepVerifier.create(resourceManagedFlux.timeout(Duration.ofSeconds(2)))
-            .thenCancel()
-            .verify(Duration.ofSeconds(1))
-
-        // Resource management should be handled properly
-        // Note: In a real scenario, we'd verify that resources are properly cleaned up
-        assertThat(resourceManagedFlux).isNotNull
-    }
-
-    data class ReactiveTestEvent(
-        val message: String,
-        val sequenceNumber: Int,
-        val timestamp: Long = System.currentTimeMillis()
-    )
-}