dc-observatio/services/pcp-route-processing-service

pcp-route-processing-service

Назначение сервиса

pcp-route-processing-service принимает route passport из Kafka, парсит входной JSON, нормализует GeoJSON geometry в WKT, сохраняет паспорт маршрута в БД и создаёт downstream events в outbox.

Публикация 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.

Kafka topics

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:

app:
  route-processing:
    test-controller:
      enabled: true

Current business limitations

• routeStatus сейчас остаётся PROCESSED.
• Сервис не делает request matching.
• Сервис не считает coverage.
• Сервис не вызывает request-service или mission-service по HTTP.