diff options
| author | Paul-Christian Volkmer | 2026-09-04 13:49:03 +0200 |
|---|---|---|
| committer | GitHub | 2026-09-04 11:49:03 +0000 |
| commit | 7671aa9f44df07a4fe3719ab7952db999feab8df (patch) | |
| tree | 0de5bdd99613c82b1d2976851a7294917e7c805f /src/main | |
| parent | a82b5dbd699991e9bff898f00c4bcff91343a945 (diff) | |
chore: clean up KafkaResponseProcessor (#315)
Diffstat (limited to 'src/main')
| -rw-r--r-- | src/main/kotlin/dev/dnpm/etl/processor/services/kafka/KafkaResponseProcessor.kt | 56 |
1 files changed, 27 insertions, 29 deletions
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(), ) } |