From 7671aa9f44df07a4fe3719ab7952db999feab8df Mon Sep 17 00:00:00 2001
From: Paul-Christian Volkmer
Date: Fri, 4 Sep 2026 13:49:03 +0200
Subject: chore: clean up KafkaResponseProcessor (#315)

---
 .../services/kafka/KafkaResponseProcessor.kt       | 56 +++++++++++-----------
 1 file changed, 27 insertions(+), 29 deletions(-)

(limited to 'src/main/kotlin/dev/dnpm/etl/processor')

diff --git a/src/main/kotlin/dev/dnpm/etl/processor/services/kafka/KafkaResponseProcessor.kt b/src/main/kotlin/dev/dnpm/etl/processor/services/kafka/KafkaResponseProcessor.kt
index 41f7720..571888d 100644
--- a/src/main/kotlin/dev/dnpm/etl/processor/services/kafka/KafkaResponseProcessor.kt
+++ b/src/main/kotlin/dev/dnpm/etl/processor/services/kafka/KafkaResponseProcessor.kt
@@ -48,42 +48,40 @@ class KafkaResponseProcessor(
             logger.error("Cannot process Kafka response", e)
             Optional.empty()
         }
-            .ifPresentOrElse(
-                { responseBody ->
-                    val event =
-                        ResponseEvent(
-                            RequestId(responseBody.requestId),
-                            Instant.ofEpochMilli(data.timestamp()),
-                            responseBody.statusCode.asRequestStatus(),
-                            when (responseBody.statusCode.asRequestStatus()) {
-                                RequestStatus.SUCCESS -> {
-                                    Optional.empty()
-                                }
+            .ifPresent { responseBody ->
+                val requestStatus = responseBody.statusCode.asRequestStatus()
+                val event =
+                    ResponseEvent(
+                        RequestId(responseBody.requestId),
+                        Instant.ofEpochMilli(data.timestamp()),
+                        requestStatus,
+                        when (requestStatus) {
+                            RequestStatus.SUCCESS -> {
+                                Optional.empty()
+                            }
 
-                                RequestStatus.WARNING,
-                                RequestStatus.ERROR -> {
-                                    Optional.of(jsonMapper.writeValueAsString(responseBody.statusBody))
-                                }
+                            RequestStatus.WARNING,
+                            RequestStatus.ERROR -> {
+                                Optional.of(jsonMapper.writeValueAsString(responseBody.statusBody))
+                            }
 
-                                else -> {
-                                    logger.error(
-                                        "Kafka response: Unknown response code '{}'!",
-                                        responseBody.statusCode,
-                                    )
-                                    Optional.empty()
-                                }
-                            },
-                        )
-                    eventPublisher.publishEvent(event)
-                },
-                { logger.error("No requestId in Kafka response") },
-            )
+                            else -> {
+                                logger.error(
+                                    "Kafka response: Unknown response code '{}'!",
+                                    responseBody.statusCode,
+                                )
+                                Optional.empty()
+                            }
+                        },
+                    )
+                eventPublisher.publishEvent(event)
+            }
     }
 
     @JsonNaming(PropertyNamingStrategies.SnakeCaseStrategy::class)
     data class ResponseBody(
         val requestId: String,
         val statusCode: Int,
-        val statusBody: Map<String, Any>,
+        val statusBody: Map<String, Any> = emptyMap(),
     )
 }
-- 
cgit v1.2.3