pcvolkmer/etl-processor
1
0
Fork 0
mirror of https://github.com/pcvolkmer/etl-processor.git synced 2025-07-02 22:42:55 +00:00
Code Issues Releases Activity

feat: use requestId from incoming Kafka Record Header

Browse Source
This commit is contained in:
Paul-Christian Volkmer
2024-03-08 15:42:04 +01:00
parent bed91439db
commit fc1901211d
3 changed files with 57 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
src/test/kotlin/dev/dnpm/etl/processor/input/KafkaInputListenerTest.kt
View File

@ -25,6 +25,9 @@ import de.ukw.ccc.bwhc.dto.MtbFile
import de.ukw.ccc.bwhc.dto.Patient
import dev.dnpm.etl.processor.services.RequestProcessor
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.header.internals.RecordHeader
import org.apache.kafka.common.header.internals.RecordHeaders
import org.apache.kafka.common.record.TimestampType
import org.junit.jupiter.api.BeforeEach
import org.junit.jupiter.api.Test
import org.junit.jupiter.api.extension.ExtendWith
@ -34,6 +37,7 @@ import org.mockito.junit.jupiter.MockitoExtension
import org.mockito.kotlin.any
import org.mockito.kotlin.times
import org.mockito.kotlin.verify
import java.util.*
@ExtendWith(MockitoExtension::class)
class KafkaInputListenerTest {
@ -76,4 +80,33 @@ class KafkaInputListenerTest {
verify(requestProcessor, times(1)).processDeletion(anyString())
}
@Test
fun shouldProcessMtbFileRequestWithExistingRequestId() {
val mtbFile = MtbFile.builder()
.withPatient(Patient.builder().withId("DUMMY_12345678").build())
.withConsent(Consent.builder().withStatus(Consent.Status.ACTIVE).build())
.build()
val headers = RecordHeaders(listOf(RecordHeader("requestId", UUID.randomUUID().toString().toByteArray())))
kafkaInputListener.onMessage(
ConsumerRecord("testtopic", 0, 0, -1L, TimestampType.NO_TIMESTAMP_TYPE, -1, -1, "", this.objectMapper.writeValueAsString(mtbFile), headers, Optional.empty())
)
verify(requestProcessor, times(1)).processMtbFile(any(), anyString())
}
@Test
fun shouldProcessDeleteRequestWithExistingRequestId() {
val mtbFile = MtbFile.builder()
.withPatient(Patient.builder().withId("DUMMY_12345678").build())
.withConsent(Consent.builder().withStatus(Consent.Status.REJECTED).build())
.build()
val headers = RecordHeaders(listOf(RecordHeader("requestId", UUID.randomUUID().toString().toByteArray())))
kafkaInputListener.onMessage(
ConsumerRecord("testtopic", 0, 0, -1L, TimestampType.NO_TIMESTAMP_TYPE, -1, -1, "", this.objectMapper.writeValueAsString(mtbFile), headers, Optional.empty())
)
verify(requestProcessor, times(1)).processDeletion(anyString(), anyString())
}
}