This commit is contained in:
Дмитрий Соловьев
2026-05-25 15:57:11 +03:00
parent fa357a78b4
commit 1ef912be66
9 changed files with 34 additions and 44 deletions
@@ -5,9 +5,9 @@ import jakarta.persistence.Entity
import jakarta.persistence.Id
import jakarta.persistence.Table
import jakarta.persistence.Version
import com.fasterxml.jackson.databind.JsonNode
import org.hibernate.annotations.JdbcTypeCode
import org.hibernate.type.SqlTypes
import tools.jackson.databind.JsonNode
import java.math.BigDecimal
import java.time.LocalDateTime
import java.util.UUID
@@ -6,9 +6,9 @@ import jakarta.persistence.EnumType
import jakarta.persistence.Enumerated
import jakarta.persistence.Id
import jakarta.persistence.Table
import com.fasterxml.jackson.databind.JsonNode
import org.hibernate.annotations.JdbcTypeCode
import org.hibernate.type.SqlTypes
import tools.jackson.databind.JsonNode
import java.time.LocalDateTime
import java.util.UUID
@@ -13,7 +13,8 @@ import space.nstart.pcp.routepassportconsumer.entity.RoutePassportOutboxEntity
import space.nstart.pcp.routepassportconsumer.entity.RoutePassportOutboxStatus
import space.nstart.pcp.routepassportconsumer.repository.RoutePassportOutboxRepository
import space.nstart.pcp.routepassportconsumer.repository.RoutePassportRepository
import tools.jackson.databind.JsonNode
import com.fasterxml.jackson.databind.JsonNode
import com.fasterxml.jackson.databind.ObjectMapper as JpaJsonObjectMapper
import tools.jackson.databind.ObjectMapper
import java.time.LocalDateTime
import java.util.UUID
@@ -29,6 +30,7 @@ class RoutePassportIngestionService(
private val kafkaTopicsProperties: KafkaTopicsProperties,
) {
private val log = LoggerFactory.getLogger(this::class.java)
private val jpaJsonObjectMapper = JpaJsonObjectMapper().findAndRegisterModules()
@Transactional
fun ingest(
@@ -164,9 +166,12 @@ class RoutePassportIngestionService(
source = "route-processing"
}
return objectMapper.valueToTree(message)
return toJpaJsonNode(message)
}
private fun toJpaJsonNode(value: Any): JsonNode =
jpaJsonObjectMapper.readTree(objectMapper.writeValueAsString(value))
private fun RoutePassportEntity.applyRoutePassport(
routePassportDto: RoutePassportDto,
parsedMessage: ParsedRoutePassportMessage,
@@ -188,7 +193,7 @@ class RoutePassportIngestionService(
visirAngleMax = routePassportDto.visirAngle.max
resolutionRange = routePassportDto.resolutionRange
resolutionAzimuth = routePassportDto.resolutionAzimuth
polarisation = objectMapper.valueToTree(routePassportDto.polarisation)
polarisation = toJpaJsonNode(routePassportDto.polarisation)
processingLevel = routePassportDto.processingLevel
geometryWkt = routePassportDto.geometry
routeStatus = routePassportDto.routeStatus.name
@@ -198,7 +203,7 @@ class RoutePassportIngestionService(
sourcePartition = sourceRecord.partition
sourceOffset = sourceRecord.offset
sourceKey = sourceRecord.key
this.rawPayload = objectMapper.readTree(rawPayload)
this.rawPayload = jpaJsonObjectMapper.readTree(rawPayload)
this.updatedAt = updatedAt
return this
@@ -226,7 +231,7 @@ class RoutePassportIngestionService(
visirAngleMax = visirAngle.max,
resolutionRange = resolutionRange,
resolutionAzimuth = resolutionAzimuth,
polarisation = objectMapper.valueToTree(polarisation),
polarisation = toJpaJsonNode(polarisation),
processingLevel = processingLevel,
geometryWkt = geometry,
routeStatus = routeStatus.name,
@@ -236,7 +241,7 @@ class RoutePassportIngestionService(
sourcePartition = sourceRecord.partition,
sourceOffset = sourceRecord.offset,
sourceKey = sourceRecord.key,
rawPayload = objectMapper.readTree(rawPayload),
rawPayload = jpaJsonObjectMapper.readTree(rawPayload),
createdAt = now,
updatedAt = now,
)
@@ -5,7 +5,6 @@ import org.slf4j.LoggerFactory
import org.springframework.kafka.core.KafkaTemplate
import org.springframework.stereotype.Service
import space.nstart.pcp.routepassportconsumer.entity.RoutePassportOutboxEntity
import tools.jackson.databind.ObjectMapper
import java.nio.charset.StandardCharsets
import java.util.concurrent.ExecutionException
import java.util.concurrent.TimeUnit
@@ -16,9 +15,6 @@ import kotlin.time.Duration.Companion.nanoseconds
*/
@Service
class RoutePassportKafkaPublisherService(
/** ObjectMapper сериализует сохранённый JsonNode payload в JSON object string перед отправкой. */
private val objectMapper: ObjectMapper,
/** KafkaTemplate отправляет готовую outbox-запись в её целевой topic. */
private val kafkaTemplate: KafkaTemplate<String, String>,
) {
@@ -28,7 +24,7 @@ class RoutePassportKafkaPublisherService(
* Публикует ровно одну outbox-запись и пробрасывает ошибку отправки вызывающему коду.
*/
fun publish(outbox: RoutePassportOutboxEntity) {
val payload = objectMapper.writeValueAsString(outbox.payload)
val payload = outbox.payload.toString()
val record = ProducerRecord(outbox.topic, outbox.messageKey, payload)
record.headers().add(TYPE_HEADER, outbox.eventType.toByteArray(StandardCharsets.UTF_8))
val startedAt = System.nanoTime()
@@ -7,7 +7,7 @@ spring:
import: "configserver:"
cloud:
config:
uri: ${CONFIG_SERVER_URI:http://192.168.100.160:38888}
uri: ${CONFIG_SERVER_URI:http://192.168.60.201:8888}
fail-fast: ${CONFIG_SERVER_FAIL_FAST:true}
profile: ${SPRING_CLOUD_CONFIG_PROFILE:${SPRING_PROFILES_ACTIVE:${spring.profiles.default}}}
label: ${SPRING_CLOUD_CONFIG_LABEL:master}
@@ -21,6 +21,7 @@ import space.nstart.pcp.routepassportconsumer.entity.RoutePassportOutboxEntity
import space.nstart.pcp.routepassportconsumer.entity.RoutePassportOutboxStatus
import space.nstart.pcp.routepassportconsumer.repository.RoutePassportOutboxRepository
import space.nstart.pcp.routepassportconsumer.repository.RoutePassportRepository
import com.fasterxml.jackson.databind.ObjectMapper as JpaJsonObjectMapper
import tools.jackson.databind.ObjectMapper
import tools.jackson.databind.json.JsonMapper
import tools.jackson.module.kotlin.KotlinModule
@@ -33,6 +34,7 @@ class RoutePassportIngestionServiceTest {
private val objectMapper: ObjectMapper = JsonMapper.builder()
.addModule(KotlinModule.Builder().build())
.build()
private val jpaJsonObjectMapper = JpaJsonObjectMapper().findAndRegisterModules()
private val kafkaTopicsProperties = KafkaTopicsProperties(
input = "pcp.request.survey-georeference.v1",
@@ -137,7 +139,7 @@ class RoutePassportIngestionServiceTest {
assertEquals(3, routeEntity.sourcePartition)
assertEquals(42L, routeEntity.sourceOffset)
assertEquals("source-key-1", routeEntity.sourceKey)
assertEquals(objectMapper.readTree(fixture.rawPayload), routeEntity.rawPayload)
assertEquals(jpaJsonObjectMapper.readTree(fixture.rawPayload), routeEntity.rawPayload)
assertEquals("trace-1", routeEntity.traceId)
assertEquals("corr-1", routeEntity.correlationId)
}
@@ -160,14 +162,14 @@ class RoutePassportIngestionServiceTest {
val modeStatusPayload = savedOutbox.first().payload
assertTrue(modeStatusPayload.isObject)
assertEquals(PcpKafkaEvent.ModeStatusChangedEvent.name, modeStatusPayload.path("type").stringValue())
assertEquals("trace-1", modeStatusPayload.path("traceId").stringValue())
assertEquals("corr-1", modeStatusPayload.path("correlationId").stringValue())
assertEquals("route-processing", modeStatusPayload.path("source").stringValue())
assertEquals("pcp", modeStatusPayload.path("group").stringValue())
assertEquals(PcpKafkaEvent.ModeStatusChangedEvent.name, modeStatusPayload.path("type").asText())
assertEquals("trace-1", modeStatusPayload.path("traceId").asText())
assertEquals("corr-1", modeStatusPayload.path("correlationId").asText())
assertEquals("route-processing", modeStatusPayload.path("source").asText())
assertEquals("pcp", modeStatusPayload.path("group").asText())
assertTrue(modeStatusPayload.path("data").isObject)
assertEquals(fixture.routeId.toString(), modeStatusPayload.path("data").path("routeId").stringValue())
assertEquals("ROUTE-001", modeStatusPayload.path("data").path("routeNameFull").stringValue())
assertEquals(fixture.routeId.toString(), modeStatusPayload.path("data").path("routeId").asText())
assertEquals("ROUTE-001", modeStatusPayload.path("data").path("routeNameFull").asText())
}
private fun service(
@@ -213,7 +215,7 @@ class RoutePassportIngestionServiceTest {
eventType = PcpKafkaEvent.ModeStatusChangedEvent.name,
topic = "topic",
messageKey = "key",
payload = objectMapper.readTree("{}"),
payload = jpaJsonObjectMapper.readTree("{}"),
status = RoutePassportOutboxStatus.PENDING,
attempts = 0,
maxAttempts = 10,
@@ -285,7 +287,7 @@ class RoutePassportIngestionServiceTest {
visirAngleMax = BigDecimal.ONE,
resolutionRange = BigDecimal.TEN,
resolutionAzimuth = BigDecimal.TEN,
polarisation = objectMapper.readTree("""["VH"]"""),
polarisation = jpaJsonObjectMapper.readTree("""["VH"]"""),
processingLevel = "GRD",
geometryWkt = "POLYGON((0 0, 1 0, 1 1, 0 0))",
routeStatus = "PROCESSED",
@@ -16,9 +16,7 @@ import org.springframework.kafka.support.SendResult
import space.nstart.pcp.pcp_types_lib.message.PcpKafkaEvent
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 com.fasterxml.jackson.databind.ObjectMapper
import java.nio.charset.StandardCharsets
import java.time.LocalDateTime
import java.util.UUID
@@ -31,9 +29,7 @@ import kotlin.test.assertFailsWith
class RoutePassportKafkaPublisherServiceTest {
private val objectMapper: ObjectMapper = JsonMapper.builder()
.addModule(KotlinModule.Builder().build())
.build()
private val objectMapper = ObjectMapper().findAndRegisterModules()
@Test
fun `successful publish sends one outbox record with topic key payload and type header`() {
@@ -101,7 +97,7 @@ class RoutePassportKafkaPublisherServiceTest {
assertTrue(objectMapper.readTree(sentPayload).isObject)
assertEquals(
PcpKafkaEvent.ModeStatusChangedEvent.name,
objectMapper.readTree(sentPayload).path("type").stringValue(),
objectMapper.readTree(sentPayload).path("type").asText(),
)
}
@@ -109,7 +105,6 @@ class RoutePassportKafkaPublisherServiceTest {
kafkaTemplate: KafkaTemplate<String, String>,
): RoutePassportKafkaPublisherService =
RoutePassportKafkaPublisherService(
objectMapper = objectMapper,
kafkaTemplate = kafkaTemplate,
)
@@ -14,9 +14,7 @@ import space.nstart.pcp.routepassportconsumer.config.RoutePassportOutboxProperti
import space.nstart.pcp.routepassportconsumer.entity.RoutePassportOutboxEntity
import space.nstart.pcp.routepassportconsumer.entity.RoutePassportOutboxStatus
import space.nstart.pcp.routepassportconsumer.repository.RoutePassportOutboxRepository
import tools.jackson.databind.ObjectMapper
import tools.jackson.databind.json.JsonMapper
import tools.jackson.module.kotlin.KotlinModule
import com.fasterxml.jackson.databind.ObjectMapper
import java.time.Duration
import java.time.LocalDateTime
import java.util.Optional
@@ -28,9 +26,7 @@ import kotlin.test.assertTrue
class RoutePassportOutboxServiceTest {
private val objectMapper: ObjectMapper = JsonMapper.builder()
.addModule(KotlinModule.Builder().build())
.build()
private val objectMapper = ObjectMapper().findAndRegisterModules()
@Test
fun `claimDueBatch returns empty when outbox is disabled`() {
@@ -10,17 +10,13 @@ import space.nstart.pcp.pcp_types_lib.message.PcpKafkaEvent
import space.nstart.pcp.routepassportconsumer.config.RoutePassportOutboxProperties
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 com.fasterxml.jackson.databind.ObjectMapper
import java.time.LocalDateTime
import java.util.UUID
class RoutePassportOutboxWorkerTest {
private val objectMapper: ObjectMapper = JsonMapper.builder()
.addModule(KotlinModule.Builder().build())
.build()
private val objectMapper = ObjectMapper().findAndRegisterModules()
@Test
fun `when disabled no claim or publish happens`() {