diff --git a/services/pcp-route-processing-service/README.md b/services/pcp-route-processing-service/README.md index 945e289..50af451 100644 --- a/services/pcp-route-processing-service/README.md +++ b/services/pcp-route-processing-service/README.md @@ -1,35 +1,92 @@ -# route-passport-kafka-consumer +# pcp-route-processing-service -Минимальный сервис на Kotlin + Spring Boot, который: -- слушает Kafka topic; -- получает в сообщении JSON паспорта маршрута; -- парсит JSON в DTO; -- больше ничего не делает. +## Назначение сервиса -## Что внутри +`pcp-route-processing-service` принимает route passport из Kafka, парсит входной JSON, нормализует GeoJSON geometry в WKT, сохраняет паспорт маршрута в БД и создаёт downstream events в outbox. -- `RoutePassportConsumerApplication` — точка входа Spring Boot. -- `RoutePassportMessageListener` — Kafka listener. -- `RoutePassportJsonParserService` — сервис парсинга входного JSON в DTO. -- `RoutePassportDto` — итоговый DTO из `pcp-types-lib`. -- `RawPassportEnvelopeDto` и связанные `Raw*` DTO — модель входного JSON из `pcp-types-lib`. +Публикация downstream events выполняется не из Kafka listener, а через scheduled outbox worker: -## Как запустить +1. `RoutePassportMessageListener` читает input Kafka record. +2. `RoutePassportJsonParserService` валидирует и парсит payload. +3. `RoutePassportIngestionService` в одной DB transaction сохраняет `route_passports` и создаёт `route_passport_outbox`. +4. Listener подтверждает offset только после успешного ingestion. +5. `RoutePassportOutboxWorker` claim'ит due outbox rows и публикует их через `RoutePassportKafkaPublisherService`. -1. Поднять Kafka. -2. Указать адрес брокера в `application.yml` или через переменную окружения `KAFKA_BOOTSTRAP_SERVERS`. -3. При необходимости указать topic через `APP_KAFKA_TOPIC`. -4. Запустить приложение: +## Kafka topics -```bash -./gradlew bootRun +Input topic: +- `pcp.request.survey-georeference.v1` + +DLQ topic: +- `pcp.request.survey-georeference-dlq.v1` + +Output topics: +- `pcp.route.in.v1` - событие для mission-planning flow. +- `pcp.route.georeference.v1` - событие для request-service matching flow. + +## DB tables + +- `route_passports` - сохранённые паспорта маршрутов и source metadata входного Kafka record. +- `route_passport_outbox` - pending/published/failed downstream events для Kafka publication. + +## Horizontal scaling + +Несколько pod'ов сервиса могут безопасно запускать outbox worker одновременно. + +Outbox claim использует DB-level concurrency: +- `SELECT ... FOR UPDATE SKIP LOCKED` выбирает due `PENDING` rows. +- `lockedUntil` выставляется в той же transaction, что и claim. +- После commit другие instances не claim'ят row, пока `lockedUntil` в будущем. +- Если instance умер после claim, abandoned row снова станет доступна после истечения `lockedUntil`. +- In-memory locks, leader election и single-instance assumptions не используются. + +## Delivery semantics + +Delivery semantics: at-least-once. + +Exactly-once не обещается. Kafka transactions не используются. Downstream consumers должны быть идемпотентны к повторной доставке одного и того же события. + +## Error handling + +Input Kafka flow: +- contract/parsing errors не подтверждают offset и обрабатываются configured retry/DLQ policy; +- после исчерпания retry record публикуется в `pcp.request.survey-georeference-dlq.v1`; +- временные ошибки ingestion retry'ятся bounded policy. + +Output Kafka flow: +- downstream events сначала сохраняются в `route_passport_outbox`; +- output Kafka errors не теряются, а переводят row обратно в `PENDING` с bounded exponential backoff; +- после исчерпания `maxAttempts` из outbox row событие получает статус `FAILED`. + +## Observability + +Логи содержат operational identifiers без полного payload: +- source topic, partition, offset, key и payload size для input ingestion; +- `routeId`, `routeNameFull`, `traceId`, `correlationId`; +- outbox id, event type, target topic, message key, attempts, status, `nextAttemptAt`, `lockedUntil`, `publishedAt`; +- exception class/message для failure path. + +Доступные Micrometer counters: +- `route_passport_ingest_success_total` +- `route_passport_ingest_failure_total` +- `route_passport_outbox_published_total` +- `route_passport_outbox_retry_total` +- `route_passport_outbox_failed_total` + +## Operational notes + +Test controller is disabled by default and must not be exposed in production unless explicitly enabled with: + +```yaml +app: + route-processing: + test-controller: + enabled: true ``` -## Сообщение, которое ожидает сервис +## Current business limitations -Сервис ожидает, что value Kafka-сообщения — это JSON того же формата, что и `src/test/resources/passport_example.json`. -Поле `geometry` должно приходить строкой в формате WKT. - -## Примечание - -Поскольку в запросе был указан просто «слушает очередь», реализация сделана через Kafka topic как наиболее типичный вариант для стека Kotlin + Spring и Kafka-ориентированного сервиса. +- `routeStatus` сейчас остаётся `PROCESSED`. +- Сервис не делает request matching. +- Сервис не считает coverage. +- Сервис не вызывает request-service или mission-service по HTTP. diff --git a/services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/controller/RouteProcessingTestController.kt b/services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/controller/RouteProcessingTestController.kt index 278ffac..8be8dbb 100644 --- a/services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/controller/RouteProcessingTestController.kt +++ b/services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/controller/RouteProcessingTestController.kt @@ -2,6 +2,7 @@ package space.nstart.pcp.routepassportconsumer.controller import io.swagger.v3.oas.annotations.Operation import io.swagger.v3.oas.annotations.tags.Tag +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty import org.springframework.core.io.ClassPathResource import org.springframework.kafka.core.KafkaTemplate import org.springframework.web.bind.annotation.PostMapping @@ -14,6 +15,11 @@ import java.nio.charset.StandardCharsets @RestController @RequestMapping("/api/route-processing") @Tag(name = "Route Processing", description = "Тестовые endpoint'ы сервиса обработки маршрутов") +@ConditionalOnProperty( + name = ["app.route-processing.test-controller.enabled"], + havingValue = "true", + matchIfMissing = false, +) class RouteProcessingTestController( private val kafkaTemplate: KafkaTemplate, private val kafkaTopicsProperties: KafkaTopicsProperties, diff --git a/services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/listener/RoutePassportMessageListener.kt b/services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/listener/RoutePassportMessageListener.kt index ca7e332..57221a2 100644 --- a/services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/listener/RoutePassportMessageListener.kt +++ b/services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/listener/RoutePassportMessageListener.kt @@ -8,6 +8,7 @@ import org.springframework.stereotype.Component import space.nstart.pcp.routepassportconsumer.config.KafkaTopicsProperties import space.nstart.pcp.routepassportconsumer.service.RoutePassportIngestionService import space.nstart.pcp.routepassportconsumer.service.RoutePassportJsonParserService +import space.nstart.pcp.routepassportconsumer.service.RoutePassportMetrics import space.nstart.pcp.routepassportconsumer.service.RoutePassportSourceRecordMetadata /** @@ -22,6 +23,9 @@ class RoutePassportMessageListener( /** Сервис транзакционного сохранения паспорта маршрута и outbox-событий. */ private val routePassportIngestionService: RoutePassportIngestionService, + /** Метрики ingestion flow. */ + private val routePassportMetrics: RoutePassportMetrics, + /** Набор Kafka topic'ов, который приходит из config server и валидируется при старте. */ val kafkaTopicsProperties: KafkaTopicsProperties, ) { @@ -59,6 +63,7 @@ class RoutePassportMessageListener( ) val parsedMessage = routePassportJsonParserService.parse(rawPayload) + val routePassportDto = parsedMessage.routePassportDto routePassportIngestionService.ingest( parsedMessage = parsedMessage, rawPayload = rawPayload, @@ -67,25 +72,29 @@ class RoutePassportMessageListener( // Transactional ingest commits before returning; only then the Kafka offset is acknowledged. acknowledgment.acknowledge() + routePassportMetrics.incrementIngestSuccess() log.info( - "Route passport ingested: sourceTopic={}, partition={}, offset={}, key={}, payloadSize={}, routeId={}, traceId={}, correlationId={}", + "Route passport ingested: sourceTopic={}, partition={}, offset={}, key={}, payloadSize={}, routeId={}, routeNameFull={}, traceId={}, correlationId={}", sourceRecord.topic, sourceRecord.partition, sourceRecord.offset, sourceRecord.key, rawPayload.length, - parsedMessage.routePassportDto.routeId, + routePassportDto.routeId, + routePassportDto.routeNameFull, parsedMessage.metadata.traceId, parsedMessage.metadata.correlationId, ) } catch (exception: Exception) { + routePassportMetrics.incrementIngestFailure() log.error( - "Failed to process Kafka record: sourceTopic={}, partition={}, offset={}, key={}, error={}", + "Failed to process Kafka record: sourceTopic={}, partition={}, offset={}, key={}, exceptionClass={}, exceptionMessage={}", sourceRecord.topic, sourceRecord.partition, sourceRecord.offset, sourceRecord.key, + exception::class.java.name, exception.message, exception, ) diff --git a/services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportIngestionService.kt b/services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportIngestionService.kt index 8d6ef1c..b1da427 100644 --- a/services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportIngestionService.kt +++ b/services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportIngestionService.kt @@ -1,5 +1,6 @@ package space.nstart.pcp.routepassportconsumer.service +import org.slf4j.LoggerFactory import org.springframework.stereotype.Service import org.springframework.transaction.annotation.Transactional import space.nstart.pcp.pcp_types_lib.dto.routes.ParsedRoutePassportMessage @@ -27,6 +28,7 @@ class RoutePassportIngestionService( private val objectMapper: ObjectMapper, private val kafkaTopicsProperties: KafkaTopicsProperties, ) { + private val log = LoggerFactory.getLogger(this::class.java) @Transactional fun ingest( @@ -36,8 +38,10 @@ class RoutePassportIngestionService( ) { val now = LocalDateTime.now() val routePassportDto = parsedMessage.routePassportDto + val existingRoutePassport = routePassportRepository.findById(routePassportDto.routeId) + val routeCreated = existingRoutePassport.isEmpty - val routePassportEntity = routePassportRepository.findById(routePassportDto.routeId) + val routePassportEntity = existingRoutePassport .map { existingEntity -> existingEntity.applyRoutePassport( routePassportDto = routePassportDto, @@ -58,17 +62,33 @@ class RoutePassportIngestionService( routePassportRepository.save(routePassportEntity) - createOutboxIfAbsent( - eventType = PcpKafkaEvent.ModeStatusChangedEvent, - topic = kafkaTopicsProperties.processedRoute, - parsedMessage = parsedMessage, - now = now, + val outboxResults = listOf( + createOutboxIfAbsent( + eventType = PcpKafkaEvent.ModeStatusChangedEvent, + topic = kafkaTopicsProperties.processedRoute, + parsedMessage = parsedMessage, + now = now, + ), + createOutboxIfAbsent( + eventType = PcpKafkaEvent.RouteGeoRefEvent, + topic = kafkaTopicsProperties.routeGeoreference, + parsedMessage = parsedMessage, + now = now, + ), ) - createOutboxIfAbsent( - eventType = PcpKafkaEvent.RouteGeoRefEvent, - topic = kafkaTopicsProperties.routeGeoreference, - parsedMessage = parsedMessage, - now = now, + + log.info( + "Route passport persisted: routeId={}, routeNameFull={}, traceId={}, correlationId={}, sourceTopic={}, sourcePartition={}, sourceOffset={}, routeCreated={}, outboxEventsCreated={}, outboxEventsExisting={}", + routePassportDto.routeId, + routePassportDto.routeNameFull, + parsedMessage.metadata.traceId, + parsedMessage.metadata.correlationId, + sourceRecord.topic, + sourceRecord.partition, + sourceRecord.offset, + routeCreated, + outboxResults.filter { it.created }.map { "${it.eventType}:${it.topic}" }, + outboxResults.filterNot { it.created }.map { "${it.eventType}:${it.topic}" }, ) } @@ -77,34 +97,57 @@ class RoutePassportIngestionService( topic: String, parsedMessage: ParsedRoutePassportMessage, now: LocalDateTime, - ) { + ): OutboxEventResult { val routeId = parsedMessage.routePassportDto.routeId + val routeNameFull = parsedMessage.routePassportDto.routeNameFull val eventTypeName = eventType.name if (routePassportOutboxRepository.existsByRouteIdAndEventType(routeId, eventTypeName)) { - return + log.debug( + "Route passport outbox event already exists: routeId={}, routeNameFull={}, traceId={}, correlationId={}, eventType={}, topic={}", + routeId, + routeNameFull, + parsedMessage.metadata.traceId, + parsedMessage.metadata.correlationId, + eventTypeName, + topic, + ) + return OutboxEventResult(eventType = eventTypeName, topic = topic, created = false) } - routePassportOutboxRepository.save( - RoutePassportOutboxEntity( - id = UUID.randomUUID(), - routeId = routeId, - eventType = eventTypeName, - topic = topic, - messageKey = routeId.toString(), - payload = buildOutboxPayload(eventType, parsedMessage), - headers = null, - status = RoutePassportOutboxStatus.PENDING, - attempts = 0, - maxAttempts = 10, - nextAttemptAt = now, - lockedUntil = null, - lastError = null, - createdAt = now, - updatedAt = now, - publishedAt = null, - ), + val outbox = RoutePassportOutboxEntity( + id = UUID.randomUUID(), + routeId = routeId, + eventType = eventTypeName, + topic = topic, + messageKey = routeId.toString(), + payload = buildOutboxPayload(eventType, parsedMessage), + headers = null, + status = RoutePassportOutboxStatus.PENDING, + attempts = 0, + maxAttempts = 10, + nextAttemptAt = now, + lockedUntil = null, + lastError = null, + createdAt = now, + updatedAt = now, + publishedAt = null, ) + routePassportOutboxRepository.save(outbox) + + log.debug( + "Route passport outbox event created: outboxId={}, routeId={}, routeNameFull={}, traceId={}, correlationId={}, eventType={}, topic={}, status={}", + outbox.id, + routeId, + routeNameFull, + parsedMessage.metadata.traceId, + parsedMessage.metadata.correlationId, + eventTypeName, + topic, + outbox.status, + ) + + return OutboxEventResult(eventType = eventTypeName, topic = topic, created = true) } private fun buildOutboxPayload( @@ -197,4 +240,10 @@ class RoutePassportIngestionService( createdAt = now, updatedAt = now, ) + + private data class OutboxEventResult( + val eventType: String, + val topic: String, + val created: Boolean, + ) } 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 b02b718..2db6e2c 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 @@ -9,6 +9,7 @@ import tools.jackson.databind.ObjectMapper import java.nio.charset.StandardCharsets import java.util.concurrent.ExecutionException import java.util.concurrent.TimeUnit +import kotlin.time.Duration.Companion.nanoseconds /** * Сервис физической публикации одной готовой outbox-записи в Kafka. @@ -30,19 +31,23 @@ class RoutePassportKafkaPublisherService( 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)) + val startedAt = System.nanoTime() try { val result = kafkaTemplate.send(record) .get(DEFAULT_PUBLISH_TIMEOUT_SECONDS, TimeUnit.SECONDS) + val sendDurationMs = (System.nanoTime() - startedAt).nanoseconds.inWholeMilliseconds log.info( - "Kafka outbox message published: outboxId={}, routeId={}, eventType={}, topic={}, partition={}, offset={}", + "Kafka outbox message published: outboxId={}, routeId={}, eventType={}, topic={}, messageKey={}, partition={}, offset={}, sendDurationMs={}", outbox.id, outbox.routeId, outbox.eventType, outbox.topic, + outbox.messageKey, result.recordMetadata.partition(), result.recordMetadata.offset(), + sendDurationMs, ) } catch (exception: InterruptedException) { Thread.currentThread().interrupt() @@ -64,11 +69,14 @@ class RoutePassportKafkaPublisherService( private fun logPublishFailure(outbox: RoutePassportOutboxEntity, exception: Throwable) { log.error( - "Failed to publish kafka outbox message: outboxId={}, routeId={}, eventType={}, topic={}", + "Failed to publish kafka outbox message: outboxId={}, routeId={}, eventType={}, topic={}, messageKey={}, exceptionClass={}, exceptionMessage={}", outbox.id, outbox.routeId, outbox.eventType, outbox.topic, + outbox.messageKey, + exception::class.java.name, + exception.message, exception, ) } diff --git a/services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportMetrics.kt b/services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportMetrics.kt new file mode 100644 index 0000000..419e648 --- /dev/null +++ b/services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportMetrics.kt @@ -0,0 +1,42 @@ +package space.nstart.pcp.routepassportconsumer.service + +import io.micrometer.core.instrument.Counter +import io.micrometer.core.instrument.MeterRegistry +import org.springframework.stereotype.Component + +/** + * Centralizes route passport counters so operational metrics names stay consistent. + */ +@Component +class RoutePassportMetrics( + meterRegistry: MeterRegistry, +) { + private val ingestSuccessCounter = meterRegistry.counter("route_passport_ingest_success_total") + private val ingestFailureCounter = meterRegistry.counter("route_passport_ingest_failure_total") + private val outboxPublishedCounter = meterRegistry.counter("route_passport_outbox_published_total") + private val outboxRetryCounter = meterRegistry.counter("route_passport_outbox_retry_total") + private val outboxFailedCounter = meterRegistry.counter("route_passport_outbox_failed_total") + + fun incrementIngestSuccess() { + ingestSuccessCounter.increment() + } + + fun incrementIngestFailure() { + ingestFailureCounter.increment() + } + + fun incrementOutboxPublished() { + outboxPublishedCounter.increment() + } + + fun incrementOutboxRetry() { + outboxRetryCounter.increment() + } + + fun incrementOutboxFailed() { + outboxFailedCounter.increment() + } + + private fun MeterRegistry.counter(name: String): Counter = + Counter.builder(name).register(this) +} diff --git a/services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportOutboxService.kt b/services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportOutboxService.kt index 2a7d20b..052c820 100644 --- a/services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportOutboxService.kt +++ b/services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportOutboxService.kt @@ -1,5 +1,6 @@ package space.nstart.pcp.routepassportconsumer.service +import org.slf4j.LoggerFactory import org.springframework.stereotype.Service import org.springframework.transaction.annotation.Transactional import space.nstart.pcp.routepassportconsumer.config.RoutePassportOutboxProperties @@ -17,7 +18,9 @@ import java.util.UUID class RoutePassportOutboxService( private val routePassportOutboxRepository: RoutePassportOutboxRepository, private val outboxProperties: RoutePassportOutboxProperties, + private val routePassportMetrics: RoutePassportMetrics, ) { + private val log = LoggerFactory.getLogger(this::class.java) /** * Claims due pending rows in one DB transaction using SELECT FOR UPDATE SKIP LOCKED. @@ -25,6 +28,7 @@ class RoutePassportOutboxService( @Transactional fun claimDueBatch(): List { if (!outboxProperties.enabled) { + log.debug("Route passport outbox claim skipped because worker is disabled") return emptyList() } @@ -37,7 +41,28 @@ class RoutePassportOutboxService( outbox.updatedAt = now } - return routePassportOutboxRepository.saveAll(claimedRows).toList() + val savedRows = routePassportOutboxRepository.saveAll(claimedRows).toList() + + log.debug( + "Route passport outbox rows claimed: claimedCount={}, batchSize={}, lockedUntil={}", + savedRows.size, + outboxProperties.batchSize, + lockedUntil, + ) + savedRows.forEach { outbox -> + log.debug( + "Route passport outbox row claimed: outboxId={}, routeId={}, eventType={}, status={}, attempts={}, nextAttemptAt={}, lockedUntil={}", + outbox.id, + outbox.routeId, + outbox.eventType, + outbox.status, + outbox.attempts, + outbox.nextAttemptAt, + outbox.lockedUntil, + ) + } + + return savedRows } /** @@ -55,6 +80,18 @@ class RoutePassportOutboxService( outbox.updatedAt = now routePassportOutboxRepository.save(outbox) + routePassportMetrics.incrementOutboxPublished() + + log.info( + "Route passport outbox status changed: outboxId={}, routeId={}, eventType={}, transition=PENDING->PUBLISHED, attempts={}, nextAttemptAt={}, lockedUntil={}, publishedAt={}", + outbox.id, + outbox.routeId, + outbox.eventType, + outbox.attempts, + outbox.nextAttemptAt, + outbox.lockedUntil, + outbox.publishedAt, + ) } /** @@ -78,6 +115,25 @@ class RoutePassportOutboxService( } routePassportOutboxRepository.save(outbox) + if (outbox.status == RoutePassportOutboxStatus.FAILED) { + routePassportMetrics.incrementOutboxFailed() + } else { + routePassportMetrics.incrementOutboxRetry() + } + + log.warn( + "Route passport outbox status changed after publish failure: outboxId={}, routeId={}, eventType={}, transition=LOCKED->{}, attempts={}, maxAttempts={}, nextAttemptAt={}, lockedUntil={}, exceptionClass={}, exceptionMessage={}", + outbox.id, + outbox.routeId, + outbox.eventType, + outbox.status, + outbox.attempts, + outbox.maxAttempts, + outbox.nextAttemptAt, + outbox.lockedUntil, + exception::class.java.name, + exception.message, + ) } private fun findOutbox(outboxId: UUID): RoutePassportOutboxEntity = diff --git a/services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportOutboxWorker.kt b/services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportOutboxWorker.kt index 58721b6..47aa5b7 100644 --- a/services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportOutboxWorker.kt +++ b/services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportOutboxWorker.kt @@ -20,11 +20,14 @@ class RoutePassportOutboxWorker( @Scheduled(fixedDelayString = "\${app.route-passport.outbox.fixed-delay-ms:1000}") fun publishDueOutboxRows() { + log.debug("Route passport outbox worker tick started: enabled={}", outboxProperties.enabled) if (!outboxProperties.enabled) { + log.debug("Route passport outbox worker tick skipped because worker is disabled") return } val outboxRows = routePassportOutboxService.claimDueBatch() + log.debug("Route passport outbox worker batch claimed: batchSize={}", outboxRows.size) outboxRows.forEach { outbox -> logClaimed(outbox) } @@ -46,6 +49,7 @@ class RoutePassportOutboxWorker( ) } } + log.debug("Route passport outbox worker tick finished: batchSize={}", outboxRows.size) } private fun publishOne(outbox: RoutePassportOutboxEntity) { @@ -54,7 +58,7 @@ class RoutePassportOutboxWorker( routePassportOutboxService.markPublished(outbox.id) log.info( - "Kafka outbox row marked published: outboxId={}, routeId={}, eventType={}, topic={}, attempts={}, status={}", + "Kafka outbox row publish succeeded: outboxId={}, routeId={}, eventType={}, topic={}, attempts={}, status={}", outbox.id, outbox.routeId, outbox.eventType, diff --git a/services/pcp-route-processing-service/src/test/kotlin/space/nstart/pcp/routepassportconsumer/listener/RoutePassportMessageListenerTest.kt b/services/pcp-route-processing-service/src/test/kotlin/space/nstart/pcp/routepassportconsumer/listener/RoutePassportMessageListenerTest.kt index a807be0..6caf039 100644 --- a/services/pcp-route-processing-service/src/test/kotlin/space/nstart/pcp/routepassportconsumer/listener/RoutePassportMessageListenerTest.kt +++ b/services/pcp-route-processing-service/src/test/kotlin/space/nstart/pcp/routepassportconsumer/listener/RoutePassportMessageListenerTest.kt @@ -1,5 +1,6 @@ package space.nstart.pcp.routepassportconsumer.listener +import io.micrometer.core.instrument.simple.SimpleMeterRegistry import org.apache.kafka.clients.consumer.ConsumerRecord import org.junit.jupiter.api.Assertions.assertThrows import org.junit.jupiter.api.Test @@ -18,6 +19,7 @@ import space.nstart.pcp.pcp_types_lib.dto.routes.RoutePassportMessageMetadata import space.nstart.pcp.routepassportconsumer.config.KafkaTopicsProperties import space.nstart.pcp.routepassportconsumer.service.RoutePassportIngestionService import space.nstart.pcp.routepassportconsumer.service.RoutePassportJsonParserService +import space.nstart.pcp.routepassportconsumer.service.RoutePassportMetrics import space.nstart.pcp.routepassportconsumer.service.RoutePassportSourceRecordMetadata import java.math.BigDecimal import java.time.LocalDateTime @@ -33,6 +35,7 @@ class RoutePassportMessageListenerTest { val listener = RoutePassportMessageListener( routePassportJsonParserService = parserService, routePassportIngestionService = ingestionService, + routePassportMetrics = routePassportMetrics(), kafkaTopicsProperties = kafkaTopicsProperties(), ) val rawPayload = """{"id":"route-1"}""" @@ -61,6 +64,7 @@ class RoutePassportMessageListenerTest { val listener = RoutePassportMessageListener( routePassportJsonParserService = parserService, routePassportIngestionService = ingestionService, + routePassportMetrics = routePassportMetrics(), kafkaTopicsProperties = kafkaTopicsProperties(), ) val rawPayload = """{"id":"route-1"}""" @@ -86,6 +90,7 @@ class RoutePassportMessageListenerTest { val listener = RoutePassportMessageListener( routePassportJsonParserService = parserService, routePassportIngestionService = ingestionService, + routePassportMetrics = routePassportMetrics(), kafkaTopicsProperties = kafkaTopicsProperties(), ) val rawPayload = """{"id":"route-1"}""" @@ -121,6 +126,9 @@ class RoutePassportMessageListenerTest { routeGeoreference = "pcp.route.georeference.v1", ) + private fun routePassportMetrics(): RoutePassportMetrics = + RoutePassportMetrics(SimpleMeterRegistry()) + private fun consumerRecord(rawPayload: String): ConsumerRecord = ConsumerRecord( INPUT_TOPIC, diff --git a/services/pcp-route-processing-service/src/test/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportOutboxServiceTest.kt b/services/pcp-route-processing-service/src/test/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportOutboxServiceTest.kt index 20bb5cc..9a4d76d 100644 --- a/services/pcp-route-processing-service/src/test/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportOutboxServiceTest.kt +++ b/services/pcp-route-processing-service/src/test/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportOutboxServiceTest.kt @@ -1,5 +1,6 @@ package space.nstart.pcp.routepassportconsumer.service +import io.micrometer.core.instrument.simple.SimpleMeterRegistry import org.junit.jupiter.api.Test import org.mockito.ArgumentMatchers.any import org.mockito.ArgumentMatchers.anyInt @@ -170,6 +171,7 @@ class RoutePassportOutboxServiceTest { RoutePassportOutboxService( routePassportOutboxRepository = repository, outboxProperties = properties, + routePassportMetrics = RoutePassportMetrics(SimpleMeterRegistry()), ) private fun properties(