публикация одной outbox-записи паспорта маршрута

- добавить надёжные настройки Kafka producer - публиковать один record из RoutePassportOutboxEntity с timeout - сериализовать JsonNode payload напрямую как JSON object - пробрасывать ошибки Kafka send вызывающему коду - обновить тесты publisher и producer config
2026-05-25 14:52:26 +03:00
parent d48ddd2657
commit 9d6f838692
6 changed files with 246 additions and 196 deletions
@@ -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
@@ -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,
            ),
        )

@@ -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<String, String>,
-
-    /** Набор 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))
+
+        try {
+            val result = kafkaTemplate.send(record)
+                .get(DEFAULT_PUBLISH_TIMEOUT_SECONDS, TimeUnit.SECONDS)

            log.info(
-            "Publishing route passport: routeNameFull={}, processedRouteTopic={}, routeGeoreferenceTopic={}",
-            routePassportDto.routeNameFull,
-            kafkaTopicsProperties.processedRoute,
-            kafkaTopicsProperties.routeGeoreference,
-        )
-
-        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,
-        )
-    }
-
-    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<SendResult<String, String>> = kafkaTemplate.send(record)
-        future.whenComplete { result, exception ->
-            if (exception == null) {
-                log.info(
-                    "Kafka message published: routeNameFull={}, topic={}, type={}, partition={}, offset={}",
-                    routeNameFull,
-                    topic,
-                    eventType,
+                "Kafka outbox message published: outboxId={}, routeId={}, eventType={}, topic={}, partition={}, offset={}",
+                outbox.id,
+                outbox.routeId,
+                outbox.eventType,
+                outbox.topic,
                result.recordMetadata.partition(),
                result.recordMetadata.offset(),
            )
-            } else {
+        } catch (exception: InterruptedException) {
+            Thread.currentThread().interrupt()
+            logPublishFailure(outbox, exception)
+            throw exception
+        } catch (exception: ExecutionException) {
+            val failure = exception.cause
+            logPublishFailure(outbox, failure ?: 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 message: routeNameFull={}, topic={}, type={}",
-                    routeNameFull,
-                    topic,
-                    eventType,
+            "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
    }
 }
@@ -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<String, String>
+        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)
+    }
+}
@@ -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<String, String> {
-        @Suppress("UNCHECKED_CAST")
-        return (any(ProducerRecord::class.java) as ProducerRecord<String, String>?)
-            ?: ProducerRecord("stub-topic", "stub-key", "stub-payload")
-    }
+    private val objectMapper: ObjectMapper = JsonMapper.builder()
+        .addModule(KotlinModule.Builder().build())
+        .build()

-    private fun captureProducerRecord(
-        captor: ArgumentCaptor<ProducerRecord<String, String>>,
-    ): ProducerRecord<String, String> {
-        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<String, String>
+            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<String, String>
-        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<String, String>
+    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<SendResult<String, String>>()
+            .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<ProducerRecord<String, String>>
-        @Suppress("UNCHECKED_CAST")
-        val messageCaptor = ArgumentCaptor.forClass(KafkaMessage::class.java) as ArgumentCaptor<KafkaMessage<RoutePassportDto>>
-
-        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<IllegalStateException> {
+            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() },
+
+        doAnswer { invocation ->
+            @Suppress("UNCHECKED_CAST")
+            val record = invocation.arguments[0] as ProducerRecord<String, String>
+            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(),
        )
-        kotlin.test.assertEquals(
-            listOf(
-                """{"type":"ModeStatusChangedEvent","routeNameFull":"ROUTE-001"}""",
-                """{"type":"RouteGeoRefEvent","routeNameFull":"ROUTE-001"}""",
+    }
+
+    private fun publisherService(
+        kafkaTemplate: KafkaTemplate<String, String>,
+    ): RoutePassportKafkaPublisherService =
+        RoutePassportKafkaPublisherService(
+            objectMapper = objectMapper,
+            kafkaTemplate = kafkaTemplate,
+        )
+
+    @Suppress("UNCHECKED_CAST")
+    private fun kafkaTemplate(): KafkaTemplate<String, String> =
+        mock(KafkaTemplate::class.java) as KafkaTemplate<String, String>
+
+    private fun captureSingleRecord(
+        kafkaTemplate: KafkaTemplate<String, String>,
+    ): ProducerRecord<String, String> {
+        val captor = producerRecordCaptor()
+        verify(kafkaTemplate, times(1)).send(captor.capture())
+        return captor.value
+    }
+
+    private fun sendResult(record: ProducerRecord<String, String>): SendResult<String, String> =
+        SendResult(
+            record,
+            RecordMetadata(
+                TopicPartition(record.topic(), 2),
+                42L,
+                0,
+                0L,
+                record.key()?.length ?: -1,
+                record.value()?.length ?: -1,
            ),
-            recordCaptor.allValues.map { it.value() },
        )
-        kotlin.test.assertEquals(
-            listOf("ModeStatusChangedEvent", "RouteGeoRefEvent"),
-            recordCaptor.allValues.map {
-                it.headers().lastHeader("type").value().toString(StandardCharsets.UTF_8)
-            },
+
+    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<ProducerRecord<String, String>> =
+        ArgumentCaptor.forClass(ProducerRecord::class.java) as ArgumentCaptor<ProducerRecord<String, String>>
+
+    @Suppress("UNCHECKED_CAST")
+    private fun anyProducerRecord(): ProducerRecord<String, String> =
+        (any(ProducerRecord::class.java) as ProducerRecord<String, String>?)
+            ?: ProducerRecord("stub-topic", "stub-key", "stub-payload")
 }
@@ -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")