From 9d6f838692af6f3d8c6b37a154e97063995eaa74 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=94=D0=BC=D0=B8=D1=82=D1=80=D0=B8=D0=B9=20=D0=A1=D0=BE?= =?UTF-8?q?=D0=BB=D0=BE=D0=B2=D1=8C=D0=B5=D0=B2?= Date: Mon, 25 May 2026 14:52:26 +0300 Subject: [PATCH] =?UTF-8?q?=D0=BF=D1=83=D0=B1=D0=BB=D0=B8=D0=BA=D0=B0?= =?UTF-8?q?=D1=86=D0=B8=D1=8F=20=D0=BE=D0=B4=D0=BD=D0=BE=D0=B9=20outbox-?= =?UTF-8?q?=D0=B7=D0=B0=D0=BF=D0=B8=D1=81=D0=B8=20=D0=BF=D0=B0=D1=81=D0=BF?= =?UTF-8?q?=D0=BE=D1=80=D1=82=D0=B0=20=D0=BC=D0=B0=D1=80=D1=88=D1=80=D1=83?= =?UTF-8?q?=D1=82=D0=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - добавить надёжные настройки Kafka producer - публиковать один record из RoutePassportOutboxEntity с timeout - сериализовать JsonNode payload напрямую как JSON object - пробрасывать ошибки Kafka send вызывающему коду - обновить тесты publisher и producer config --- config-repo/pcp-route-processing-service.yaml | 5 + .../config/KafkaProducerConfig.kt | 16 ++ .../RoutePassportKafkaPublisherService.kt | 137 ++++------ .../config/KafkaProducerConfigTest.kt | 34 +++ .../RoutePassportKafkaPublisherServiceTest.kt | 248 ++++++++++-------- settings.gradle.kts | 2 +- 6 files changed, 246 insertions(+), 196 deletions(-) create mode 100644 services/pcp-route-processing-service/src/test/kotlin/space/nstart/pcp/routepassportconsumer/config/KafkaProducerConfigTest.kt diff --git a/config-repo/pcp-route-processing-service.yaml b/config-repo/pcp-route-processing-service.yaml index a6a4614..5828e64 100644 --- a/config-repo/pcp-route-processing-service.yaml +++ b/config-repo/pcp-route-processing-service.yaml @@ -38,6 +38,11 @@ spring: producer: key-serializer: org.apache.kafka.common.serialization.StringSerializer value-serializer: org.apache.kafka.common.serialization.StringSerializer + acks: all + retries: 10 + properties: + enable.idempotence: true + max.in.flight.requests.per.connection: 5 listener: missing-topics-fatal: false ack-mode: manual_immediate diff --git a/services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/config/KafkaProducerConfig.kt b/services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/config/KafkaProducerConfig.kt index 4995630..0468cac 100644 --- a/services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/config/KafkaProducerConfig.kt +++ b/services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/config/KafkaProducerConfig.kt @@ -21,6 +21,18 @@ import org.springframework.kafka.core.ProducerFactory class KafkaProducerConfig( @param:Value("\${spring.kafka.bootstrap-servers}") private val bootstrapServers: String, + + @param:Value("\${spring.kafka.producer.acks:all}") + private val acks: String, + + @param:Value("\${spring.kafka.producer.retries:10}") + private val retries: Int, + + @param:Value("\${spring.kafka.producer.properties.enable.idempotence:true}") + private val enableIdempotence: Boolean, + + @param:Value("\${spring.kafka.producer.properties.max.in.flight.requests.per.connection:5}") + private val maxInFlightRequestsPerConnection: Int, ) { @Bean @@ -30,6 +42,10 @@ class KafkaProducerConfig( ProducerConfig.BOOTSTRAP_SERVERS_CONFIG to bootstrapServers, ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG to StringSerializer::class.java, ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG to StringSerializer::class.java, + ProducerConfig.ACKS_CONFIG to acks, + ProducerConfig.RETRIES_CONFIG to retries, + ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG to enableIdempotence, + ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION to maxInFlightRequestsPerConnection, ), ) diff --git a/services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportKafkaPublisherService.kt b/services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportKafkaPublisherService.kt index bc39e4a..b02b718 100644 --- a/services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportKafkaPublisherService.kt +++ b/services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportKafkaPublisherService.kt @@ -3,111 +3,78 @@ package space.nstart.pcp.routepassportconsumer.service import org.apache.kafka.clients.producer.ProducerRecord import org.slf4j.LoggerFactory import org.springframework.kafka.core.KafkaTemplate -import org.springframework.kafka.support.SendResult import org.springframework.stereotype.Service -import space.nstart.pcp.routepassportconsumer.config.KafkaTopicsProperties -import space.nstart.pcp.pcp_types_lib.dto.routes.ParsedRoutePassportMessage -import space.nstart.pcp.pcp_types_lib.message.KafkaMessage -import space.nstart.pcp.pcp_types_lib.message.PcpKafkaEvent +import space.nstart.pcp.routepassportconsumer.entity.RoutePassportOutboxEntity import tools.jackson.databind.ObjectMapper import java.nio.charset.StandardCharsets -import java.util.concurrent.CompletableFuture +import java.util.concurrent.ExecutionException +import java.util.concurrent.TimeUnit /** - * Сервис публикации обработанного маршрута в Kafka. + * Сервис физической публикации одной готовой outbox-записи в Kafka. */ @Service class RoutePassportKafkaPublisherService( - /** ObjectMapper сериализует итоговый DTO в JSON перед отправкой. */ + /** ObjectMapper сериализует сохранённый JsonNode payload в JSON object string перед отправкой. */ private val objectMapper: ObjectMapper, - /** KafkaTemplate отправляет одно и то же сообщение в несколько topic'ов. */ + /** KafkaTemplate отправляет готовую outbox-запись в её целевой topic. */ private val kafkaTemplate: KafkaTemplate, - - /** Набор Kafka topic'ов, в которые публикуется результат обработки. */ - private val kafkaTopicsProperties: KafkaTopicsProperties, ) { private val log = LoggerFactory.getLogger(this::class.java) /** - * Публикует полный RoutePassportDto в оба выходных topic'а. + * Публикует ровно одну outbox-запись и пробрасывает ошибку отправки вызывающему коду. */ - fun publish(parsedMessage: ParsedRoutePassportMessage) { - val routePassportDto = parsedMessage.routePassportDto + fun publish(outbox: RoutePassportOutboxEntity) { + val payload = objectMapper.writeValueAsString(outbox.payload) + val record = ProducerRecord(outbox.topic, outbox.messageKey, payload) + record.headers().add(TYPE_HEADER, outbox.eventType.toByteArray(StandardCharsets.UTF_8)) - log.info( - "Publishing route passport: routeNameFull={}, processedRouteTopic={}, routeGeoreferenceTopic={}", - routePassportDto.routeNameFull, - kafkaTopicsProperties.processedRoute, - kafkaTopicsProperties.routeGeoreference, - ) + try { + val result = kafkaTemplate.send(record) + .get(DEFAULT_PUBLISH_TIMEOUT_SECONDS, TimeUnit.SECONDS) - sendWithLogging( - topic = kafkaTopicsProperties.processedRoute, - eventType = PcpKafkaEvent.ModeStatusChangedEvent, - payload = buildPayload( - eventType = PcpKafkaEvent.ModeStatusChangedEvent, - parsedMessage = parsedMessage, - ), - traceId = parsedMessage.metadata.traceId, - routeNameFull = routePassportDto.routeNameFull, - ) - sendWithLogging( - topic = kafkaTopicsProperties.routeGeoreference, - eventType = PcpKafkaEvent.RouteGeoRefEvent, - payload = buildPayload( - eventType = PcpKafkaEvent.RouteGeoRefEvent, - parsedMessage = parsedMessage, - ), - traceId = parsedMessage.metadata.traceId, - routeNameFull = routePassportDto.routeNameFull, - ) - } + log.info( + "Kafka outbox message published: outboxId={}, routeId={}, eventType={}, topic={}, partition={}, offset={}", + outbox.id, + outbox.routeId, + outbox.eventType, + outbox.topic, + result.recordMetadata.partition(), + result.recordMetadata.offset(), + ) + } catch (exception: InterruptedException) { + Thread.currentThread().interrupt() + logPublishFailure(outbox, exception) + throw exception + } catch (exception: ExecutionException) { + val failure = exception.cause + logPublishFailure(outbox, failure ?: exception) - private fun buildPayload(eventType: PcpKafkaEvent, parsedMessage: ParsedRoutePassportMessage): String { - val message = KafkaMessage( - type = eventType, - data = parsedMessage.routePassportDto, - traceId = parsedMessage.metadata.traceId, - correlationId = parsedMessage.metadata.correlationId, - group = "pcp", - ).apply { - source = "route-processing" - } - - return objectMapper.writeValueAsString(message) - } - - private fun sendWithLogging( - topic: String, - eventType: PcpKafkaEvent, - payload: String, - traceId: String, - routeNameFull: String, - ) { - val record = ProducerRecord(topic, traceId, payload) - record.headers().add("type", eventType.name.toByteArray(StandardCharsets.UTF_8)) - - val future: CompletableFuture> = kafkaTemplate.send(record) - future.whenComplete { result, exception -> - if (exception == null) { - log.info( - "Kafka message published: routeNameFull={}, topic={}, type={}, partition={}, offset={}", - routeNameFull, - topic, - eventType, - result.recordMetadata.partition(), - result.recordMetadata.offset(), - ) - } else { - log.error( - "Failed to publish kafka message: routeNameFull={}, topic={}, type={}", - routeNameFull, - topic, - eventType, - exception, - ) + when (failure) { + is Exception -> throw failure + else -> throw exception } + } catch (exception: Exception) { + logPublishFailure(outbox, exception) + throw exception } } + + private fun logPublishFailure(outbox: RoutePassportOutboxEntity, exception: Throwable) { + log.error( + "Failed to publish kafka outbox message: outboxId={}, routeId={}, eventType={}, topic={}", + outbox.id, + outbox.routeId, + outbox.eventType, + outbox.topic, + exception, + ) + } + + private companion object { + const val TYPE_HEADER = "type" + const val DEFAULT_PUBLISH_TIMEOUT_SECONDS = 10L + } } diff --git a/services/pcp-route-processing-service/src/test/kotlin/space/nstart/pcp/routepassportconsumer/config/KafkaProducerConfigTest.kt b/services/pcp-route-processing-service/src/test/kotlin/space/nstart/pcp/routepassportconsumer/config/KafkaProducerConfigTest.kt new file mode 100644 index 0000000..379f06e --- /dev/null +++ b/services/pcp-route-processing-service/src/test/kotlin/space/nstart/pcp/routepassportconsumer/config/KafkaProducerConfigTest.kt @@ -0,0 +1,34 @@ +package space.nstart.pcp.routepassportconsumer.config + +import org.apache.kafka.clients.producer.ProducerConfig +import org.apache.kafka.common.serialization.StringSerializer +import org.assertj.core.api.Assertions.assertThat +import org.junit.jupiter.api.Test +import org.springframework.kafka.core.DefaultKafkaProducerFactory + +class KafkaProducerConfigTest { + + @Test + fun `producer factory applies reliability settings`() { + val config = KafkaProducerConfig( + bootstrapServers = "localhost:9092", + acks = "all", + retries = 10, + enableIdempotence = true, + maxInFlightRequestsPerConnection = 5, + ) + + @Suppress("UNCHECKED_CAST") + val producerFactory = config.producerFactory() as DefaultKafkaProducerFactory + val properties = producerFactory.configurationProperties + + assertThat(properties) + .containsEntry(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092") + .containsEntry(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer::class.java) + .containsEntry(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer::class.java) + .containsEntry(ProducerConfig.ACKS_CONFIG, "all") + .containsEntry(ProducerConfig.RETRIES_CONFIG, 10) + .containsEntry(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true) + .containsEntry(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5) + } +} diff --git a/services/pcp-route-processing-service/src/test/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportKafkaPublisherServiceTest.kt b/services/pcp-route-processing-service/src/test/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportKafkaPublisherServiceTest.kt index 7e550a4..1a559f2 100644 --- a/services/pcp-route-processing-service/src/test/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportKafkaPublisherServiceTest.kt +++ b/services/pcp-route-processing-service/src/test/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportKafkaPublisherServiceTest.kt @@ -1,6 +1,8 @@ package space.nstart.pcp.routepassportconsumer.service import org.apache.kafka.clients.producer.ProducerRecord +import org.apache.kafka.clients.producer.RecordMetadata +import org.apache.kafka.common.TopicPartition import org.junit.jupiter.api.Test import org.mockito.ArgumentCaptor import org.mockito.ArgumentMatchers.any @@ -11,134 +13,160 @@ import org.mockito.Mockito.times import org.mockito.Mockito.verify import org.springframework.kafka.core.KafkaTemplate import org.springframework.kafka.support.SendResult -import space.nstart.pcp.routepassportconsumer.config.KafkaTopicsProperties -import space.nstart.pcp.pcp_types_lib.dto.routes.AngleRangeDto -import space.nstart.pcp.pcp_types_lib.dto.routes.ParsedRoutePassportMessage -import space.nstart.pcp.pcp_types_lib.dto.routes.RoutePassportDto -import space.nstart.pcp.pcp_types_lib.dto.routes.RoutePassportMessageMetadata -import space.nstart.pcp.pcp_types_lib.message.KafkaMessage import space.nstart.pcp.pcp_types_lib.message.PcpKafkaEvent -import java.math.BigDecimal +import space.nstart.pcp.routepassportconsumer.entity.RoutePassportOutboxEntity +import space.nstart.pcp.routepassportconsumer.entity.RoutePassportOutboxStatus +import tools.jackson.databind.ObjectMapper +import tools.jackson.databind.json.JsonMapper +import tools.jackson.module.kotlin.KotlinModule import java.nio.charset.StandardCharsets import java.time.LocalDateTime import java.util.UUID import java.util.concurrent.CompletableFuture -import tools.jackson.databind.ObjectMapper +import kotlin.test.assertEquals +import kotlin.test.assertFalse +import kotlin.test.assertSame +import kotlin.test.assertTrue +import kotlin.test.assertFailsWith class RoutePassportKafkaPublisherServiceTest { - private fun anyProducerRecord(): ProducerRecord { - @Suppress("UNCHECKED_CAST") - return (any(ProducerRecord::class.java) as ProducerRecord?) - ?: ProducerRecord("stub-topic", "stub-key", "stub-payload") - } + private val objectMapper: ObjectMapper = JsonMapper.builder() + .addModule(KotlinModule.Builder().build()) + .build() - private fun captureProducerRecord( - captor: ArgumentCaptor>, - ): ProducerRecord { - return captor.capture() ?: ProducerRecord("stub-topic", "stub-key", "stub-payload") + @Test + fun `successful publish sends one outbox record with topic key payload and type header`() { + val kafkaTemplate = kafkaTemplate() + val publisherService = publisherService(kafkaTemplate) + val outbox = outbox() + + doAnswer { invocation -> + @Suppress("UNCHECKED_CAST") + val record = invocation.arguments[0] as ProducerRecord + CompletableFuture.completedFuture(sendResult(record)) + }.`when`(kafkaTemplate).send(anyProducerRecord()) + + publisherService.publish(outbox) + + val record = captureSingleRecord(kafkaTemplate) + assertEquals(outbox.topic, record.topic()) + assertEquals(outbox.messageKey, record.key()) + assertEquals(objectMapper.writeValueAsString(outbox.payload), record.value()) + assertEquals( + outbox.eventType, + record.headers().lastHeader("type").value().toString(StandardCharsets.UTF_8), + ) } @Test - fun `should publish kafka message to both output topics with dedicated event types`() { - val objectMapper = mock(ObjectMapper::class.java) - @Suppress("UNCHECKED_CAST") - val kafkaTemplate = mock(KafkaTemplate::class.java) as KafkaTemplate - val kafkaTopicsProperties = KafkaTopicsProperties( - input = "pcp.request.survey-georeference.v1", - processedRoute = "pcp.route.in.v1", - routeGeoreference = "pcp.route.georeference.v1", - ) - val publisherService = RoutePassportKafkaPublisherService( - objectMapper = objectMapper, - kafkaTemplate = kafkaTemplate, - kafkaTopicsProperties = kafkaTopicsProperties, - ) - val parsedMessage = ParsedRoutePassportMessage( - metadata = RoutePassportMessageMetadata( - traceId = "trace-1", - correlationId = "corr-1", - ), - routePassportDto = RoutePassportDto( - routeId = UUID.fromString("51b07d12-1823-4bd5-b55d-de02a5304c03"), - kaShort = "S1A", - kaFull = "sentinel-1a", - routeNameFull = "ROUTE-001", - routeNameShort = "S1A-GRDH", - orbitNumber = 59796L, - orbitState = "Descending", - intervalBegin = LocalDateTime.of(2026, 4, 1, 10, 15, 0), - intervalEnd = LocalDateTime.of(2026, 4, 1, 10, 16, 30), - rollAngle = AngleRangeDto( - min = BigDecimal("-15.5"), - max = BigDecimal("-12.0"), - ), - visirAngle = AngleRangeDto( - min = BigDecimal("21.0"), - max = BigDecimal("24.5"), - ), - resolutionRange = BigDecimal("10.0"), - resolutionAzimuth = BigDecimal("10.0"), - polarisation = listOf("VH", "VV"), - processingLevel = "GRD", - geometry = "POLYGON((0 0, 1 0, 1 1, 0 0))", - ), - ) - val sendResult = mock(SendResult::class.java) as SendResult + fun `send failure is propagated to caller`() { + val kafkaTemplate = kafkaTemplate() + val publisherService = publisherService(kafkaTemplate) + val outbox = outbox() + val failure = IllegalStateException("broker unavailable") + val failedFuture = CompletableFuture>() + .apply { completeExceptionally(failure) } - doAnswer { invocation -> - val message = invocation.arguments[0] as KafkaMessage<*> - """{"type":"${message.type.enumCast()?.name}","routeNameFull":"${(message.data as RoutePassportDto).routeNameFull}"}""" - }.`when`(objectMapper).writeValueAsString(any()) - doReturn(CompletableFuture.completedFuture(sendResult)) + doReturn(failedFuture) .`when`(kafkaTemplate) .send(anyProducerRecord()) - publisherService.publish(parsedMessage) - - @Suppress("UNCHECKED_CAST") - val recordCaptor = ArgumentCaptor.forClass(ProducerRecord::class.java) as ArgumentCaptor> - @Suppress("UNCHECKED_CAST") - val messageCaptor = ArgumentCaptor.forClass(KafkaMessage::class.java) as ArgumentCaptor> - - verify(objectMapper, times(2)).writeValueAsString(messageCaptor.capture()) - verify(kafkaTemplate, times(2)).send(captureProducerRecord(recordCaptor)) - - kotlin.test.assertEquals( - listOf(PcpKafkaEvent.ModeStatusChangedEvent, PcpKafkaEvent.RouteGeoRefEvent), - messageCaptor.allValues.map { it.type }, - ) - messageCaptor.allValues.forEach { message -> - kotlin.test.assertEquals("trace-1", message.traceId) - kotlin.test.assertEquals("corr-1", message.correlationId) - kotlin.test.assertEquals("pcp", message.group) - kotlin.test.assertEquals("route-processing", message.source) - kotlin.test.assertEquals("application/json", message.dataContentType) - kotlin.test.assertEquals("1.0", message.specVersion) - kotlin.test.assertEquals(parsedMessage.routePassportDto, message.data) - kotlin.test.assertTrue(message.id.isNotBlank()) - kotlin.test.assertNotNull(message.time) + val thrown = assertFailsWith { + publisherService.publish(outbox) } - kotlin.test.assertEquals( - listOf("pcp.route.in.v1", "pcp.route.georeference.v1"), - recordCaptor.allValues.map { it.topic() }, + + assertSame(failure, thrown) + verify(kafkaTemplate, times(1)).send(anyProducerRecord()) + } + + @Test + fun `payload JsonNode is serialized as JSON object instead of escaped JSON string`() { + val kafkaTemplate = kafkaTemplate() + val publisherService = publisherService(kafkaTemplate) + val outbox = outbox( + payload = """{"type":"ModeStatusChangedEvent","data":{"routeId":"51b07d12-1823-4bd5-b55d-de02a5304c03"}}""", ) - kotlin.test.assertEquals( - listOf("trace-1", "trace-1"), - recordCaptor.allValues.map { it.key() }, - ) - kotlin.test.assertEquals( - listOf( - """{"type":"ModeStatusChangedEvent","routeNameFull":"ROUTE-001"}""", - """{"type":"RouteGeoRefEvent","routeNameFull":"ROUTE-001"}""", - ), - recordCaptor.allValues.map { it.value() }, - ) - kotlin.test.assertEquals( - listOf("ModeStatusChangedEvent", "RouteGeoRefEvent"), - recordCaptor.allValues.map { - it.headers().lastHeader("type").value().toString(StandardCharsets.UTF_8) - }, + + doAnswer { invocation -> + @Suppress("UNCHECKED_CAST") + val record = invocation.arguments[0] as ProducerRecord + CompletableFuture.completedFuture(sendResult(record)) + }.`when`(kafkaTemplate).send(anyProducerRecord()) + + publisherService.publish(outbox) + + val sentPayload = captureSingleRecord(kafkaTemplate).value() + assertFalse(sentPayload.startsWith("\"")) + assertTrue(objectMapper.readTree(sentPayload).isObject) + assertEquals( + PcpKafkaEvent.ModeStatusChangedEvent.name, + objectMapper.readTree(sentPayload).path("type").stringValue(), ) } + + private fun publisherService( + kafkaTemplate: KafkaTemplate, + ): RoutePassportKafkaPublisherService = + RoutePassportKafkaPublisherService( + objectMapper = objectMapper, + kafkaTemplate = kafkaTemplate, + ) + + @Suppress("UNCHECKED_CAST") + private fun kafkaTemplate(): KafkaTemplate = + mock(KafkaTemplate::class.java) as KafkaTemplate + + private fun captureSingleRecord( + kafkaTemplate: KafkaTemplate, + ): ProducerRecord { + val captor = producerRecordCaptor() + verify(kafkaTemplate, times(1)).send(captor.capture()) + return captor.value + } + + private fun sendResult(record: ProducerRecord): SendResult = + SendResult( + record, + RecordMetadata( + TopicPartition(record.topic(), 2), + 42L, + 0, + 0L, + record.key()?.length ?: -1, + record.value()?.length ?: -1, + ), + ) + + private fun outbox( + payload: String = """{"type":"ModeStatusChangedEvent","data":{"routeNameFull":"ROUTE-001"}}""", + ): RoutePassportOutboxEntity { + val routeId = UUID.fromString("51b07d12-1823-4bd5-b55d-de02a5304c03") + val now = LocalDateTime.of(2026, 4, 1, 10, 15, 0) + + return RoutePassportOutboxEntity( + id = UUID.fromString("3ed5e75c-a5c0-4d63-88be-a701b5222b62"), + routeId = routeId, + eventType = PcpKafkaEvent.ModeStatusChangedEvent.name, + topic = "pcp.route.in.v1", + messageKey = routeId.toString(), + payload = objectMapper.readTree(payload), + headers = null, + status = RoutePassportOutboxStatus.PENDING, + attempts = 0, + maxAttempts = 10, + nextAttemptAt = now, + createdAt = now, + updatedAt = now, + ) + } + + @Suppress("UNCHECKED_CAST") + private fun producerRecordCaptor(): ArgumentCaptor> = + ArgumentCaptor.forClass(ProducerRecord::class.java) as ArgumentCaptor> + + @Suppress("UNCHECKED_CAST") + private fun anyProducerRecord(): ProducerRecord = + (any(ProducerRecord::class.java) as ProducerRecord?) + ?: ProducerRecord("stub-topic", "stub-key", "stub-payload") } diff --git a/settings.gradle.kts b/settings.gradle.kts index bc5adbf..9520fbf 100644 --- a/settings.gradle.kts +++ b/settings.gradle.kts @@ -1,6 +1,6 @@ import org.gradle.kotlin.dsl.maven -rootProject.name = "pcp" +rootProject.name = "observatio-terrae" rootProject.buildFileName = "build.gradle.kts" include(":config-repo")