pcvolkmer/mv64e-etl-processor
1
0
Fork 0
mirror of https://github.com/pcvolkmer/mv64e-etl-processor synced 2025-09-13 01:02:50 +00:00
Code Issues Releases Activity

fix: add missing requestId to KafkaMtbFileSender (#142)

Browse Source
This commit is contained in:
Paul-Christian Volkmer
2025-08-27 15:07:43 +02:00
committed by GitHub
parent b7aa187293
commit ba21d029d1
2 changed files with 5 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
src/main/kotlin/dev/dnpm/etl/processor/output/KafkaMtbFileSender.kt
View File

@@ -27,7 +27,6 @@ import dev.pcvolkmer.mv64e.mtb.Mtb
import dev.pcvolkmer.mv64e.mtb.MvhMetadata
import org.apache.kafka.clients.producer.ProducerRecord
import org.slf4j.LoggerFactory
import org.springframework.http.MediaType
import org.springframework.kafka.core.KafkaTemplate
import org.springframework.retry.support.RetryTemplate
@@ -47,8 +46,9 @@ class KafkaMtbFileSender(
ProducerRecord(
kafkaProperties.outputTopic,
key(request),
objectMapper.writeValueAsString(request)
objectMapper.writeValueAsString(request),
)
record.headers().add("requestId", request.requestId.value.toByteArray())
when (request) {
is DnpmV2MtbFileRequest -> record.headers()
.add(
@@ -82,7 +82,6 @@ class KafkaMtbFileSender(
ProducerRecord(
kafkaProperties.outputTopic,
key(request),
// Always use old BwhcV1FileRequest with Consent REJECT
objectMapper.writeValueAsString(
DnpmV2MtbFileRequest(
request.requestId,
@@ -90,7 +89,7 @@ class KafkaMtbFileSender(
)
)
)
record.headers().add("requestId", request.requestId.value.toByteArray())
val result = kafkaTemplate.send(record)
if (result.get() != null) {
logger.debug("Sent deletion request via KafkaMtbFileSender")

2
src/test/kotlin/dev/dnpm/etl/processor/output/KafkaMtbFileSenderTest.kt
View File

@@ -163,6 +163,8 @@ class KafkaMtbFileSenderTest {
assertThat(captor.firstValue.key()).isEqualTo("{\"pid\": \"PID\"}")
assertThat(captor.firstValue.headers().headers("contentType")).isNotNull
assertThat(captor.firstValue.headers().headers("contentType")?.firstOrNull()?.value()).isEqualTo(CustomMediaType.APPLICATION_VND_DNPM_V2_MTB_JSON_VALUE.toByteArray())
assertThat(captor.firstValue.headers().headers("requestId")).isNotNull
assertThat(captor.firstValue.headers().headers("requestId")?.firstOrNull()?.value()).isEqualTo(TEST_REQUEST_ID.value.toByteArray())
assertThat(captor.firstValue.value()).isNotNull
assertThat(captor.firstValue.value()).isEqualTo(objectMapper.writeValueAsString(dnmpV2kafkaRecordData(TEST_REQUEST_ID)))
}