Перейти к основному содержимому

Consumer: Talent Handbook Sync (pin-handbook-talent-sync)

Консолидированный документ Kafka-синка из talent.microservices.handbook в контекст handbook PIN. Один consumer group обрабатывает четыре потока событий (country, city + derive региона, currency, language); пятый топик источника (contact-info) не потребляется. Правила синка R1–R9, таблицы маппинга полей и HandbookSyncState — дом talent-replica.md; целевые реплики — country.md, region.md, city.md, currency.md, language.md. Retry/DLQ/replay-политика группы описана здесь единожды на все потоки.

Назначение​

Consumer group pin-handbook-talent-sync потребляет upsert-события Talent Handbook и поддерживает в PIN реплики пяти справочников: страны, регионы (derive из city-событий), города, валюты, языки. Каждое применённое изменение re-publish-ится собственным событием PIN в compacted-топики pin.handbook.*.v1, чтобы downstream-сервисы PIN (chessboard и будущие потребители) не подписывались на топики Talent напрямую (ADR-0063 п.1).

Owning-сервис и триггер​

ПолеЗначение
Сервис/контекстhandbook
Consumer grouppin-handbook-talent-sync
Producertalent.microservices.handbook (внешняя система; контракты — upstream source of truth, schema registry Talent; имена топиков — зафиксированное assumption ADR-0063)
Топик 1handbooks.after-country-information-update — полный snapshot страны
Топик 2handbooks.after-city-information-update — полный snapshot города + вложенные CountryEventModel / RegionEventModel
Топик 3handbooks.after-currency-information-update — полный snapshot валюты
Топик 4handbooks.after-language-information-update — полный snapshot языка
ContractVersionожидается 2 для всех событий источника; при несовпадении — warn-and-proceed + метрика handbook_sync_contract_mismatch_total
Порядокkey события = id записи → все события одной записи в одной партиции, строгий порядок per-record; кросс-справочный порядок (country→city) НЕ гарантирован — правило R4 talent-replica.md
DLQpin.handbook.talent-sync-dlq.v1

Маппинг «событие → эффект» (бизнес-термины):

ПотокСобытие (топик)Эффект
Acountry-событиеСоздание/обновление CountryEntity + re-publish country-upserted
Bcity-событиеDerive RegionEntity из вложенной RegionEventModel (+ re-publish region-upserted при фактическом изменении) + создание/обновление CityEntity + re-publish city-upserted
Ccurrency-событиеСоздание/обновление CurrencyEntity + re-publish currency-upserted
Dlanguage-событиеСоздание/обновление LanguageEntity + re-publish language-upserted
Econtact-info-событие (handbooks.after-contact-information-update)Не потребляется — ContactInfoType вне scope v1 (ADR-0063 п.3)

Контекст события​

  • Из метаданных события доступны: MessageId (идентификатор для дедупликации), envelope CreatedAt (время публикации события источником — источник монотонности R2 для всех типов записей, включая регион), ContractVersion схемы (статический, = 2), идентификатор записи-источника (key).
  • Per-record Version в payload источника нет — порядок применения upsert-ов задаётся временем публикации (envelope CreatedAt) и порядком партиции Kafka (key = Id не даёт переупорядочивания апдейтов одной записи).
  • Актор — система синка (не пользовательский контекст); справочники платформенные, tenant-фильтр не применяется, PII нет.
  • Корреляция сквозного трейса восстанавливается из метаданных и пробрасывается в исходящие pin.handbook.*.
  • Ключ идемпотентности: handbook-sync:{eventType}:{MessageId}, eventType ∈ {country, city, currency, language}; бизнес-идемпотентность — monotonic-guard R2 (CreatedAt > LastEventAt для всех типов записей, включая регион).

Входные данные / payload событий​

Контракты внешние (schema registry Talent — upstream source of truth); ниже — поля, которые consumer реально читает. Полные таблицы маппинга «поле события → поле реплики» — дом talent-replica.md; здесь — связка «поле payload → шаг алгоритма».

Country-событие (полный snapshot страны)​

Поле payloadТипОбязательностьКак используется (маппинг)
IdshortДаКлюч поиска/создания страны → CountryEntity.Id
ShortNamestring?НетПерезапись ShortName
FullNamestringДаПерезапись FullName (display Name = FullName); > 64 символов — truncate + warn
IsoCode2stringДаНормализация upper → IsoCode2; guard INV-HND-2
IsoCode3stringДаНормализация upper → IsoCode3; guard INV-HND-2
DefaultCurrencyмодель? {CurrencyId short, IsoCode, Sign, Name}НетТолько CurrencyId → DefaultCurrencyId (логическая ссылка без FK); display-поля игнорируются
DefaultLanguageмодель? {LanguageId short, IsoCode2, IsoCode3, Name}НетТолько LanguageId → DefaultLanguageId
CreatedAt (envelope, метаданные)DateTimeOffsetДаМонотонный guard R2 → LastEventAt (per-record Version в payload нет)

City-событие (полный snapshot города + вложенные модели)​

Поле payloadТипОбязательностьКак используется (маппинг)
IdlongДаКлюч поиска/создания города → CityEntity.Id
NamestringДаПерезапись CityEntity.Name; > 256 символов — truncate + warn
CountryCountryEventModel {CountryId short, Name, IsoCode2, IsoCode3}ДаТолько CountryId → CityEntity.CountryId; страна обязана существовать — иначе retriable R4; вложенная модель страну НЕ создаёт
RegionRegionEventModel? {RegionId long, Name}НетDerive RegionEntity: Id = Region.RegionId, Name, CountryId — из Country.CountryId этого же события (RegionEventModel CountryId не несёт, ADR-0063 п.4) + CityEntity.RegionId = Region.RegionId; guard INV-HND-3
CreatedAt (envelope, метаданные)DateTimeOffsetДаМонотонный guard R2 → LastEventAt и города, и региона (per-record Version в payload нет; источник монотонности — envelope CreatedAt + порядок партиции)

Currency-событие (полный snapshot валюты)​

Поле payloadТипОбязательностьКак используется (маппинг)
IdshortДаКлюч поиска/создания валюты → CurrencyEntity.Id
NamestringДаПерезапись Name; > 64 символов — truncate + warn
IsoCodestringДаНормализация upper → IsoCode; guard INV-HND-2
Signstring?НетПерезапись Sign
CreatedAt (envelope, метаданные)DateTimeOffsetДаМонотонный guard R2 → LastEventAt (per-record Version в payload нет)

Language-событие (полный snapshot языка)​

Поле payloadТипОбязательностьКак используется (маппинг)
IdshortДаКлюч поиска/создания языка → LanguageEntity.Id
NamestringДаПерезапись Name; > 64 символов — truncate + warn
IsoCode2stringДаНормализация lower → IsoCode2; guard INV-HND-2
IsoCode3stringДаНормализация lower → IsoCode3; guard INV-HND-2
FlagIsoCode2string?НетНормализация upper → FlagIsoCode2 (код страны флага, display)
CreatedAt (envelope, метаданные)DateTimeOffsetДаМонотонный guard R2 → LastEventAt (per-record Version в payload нет)

Алгоритм​

Отдельный проход для каждого потока. Общая рамка каждого шага 1: восстановить метаданные (MessageId, envelope CreatedAt, key); сверить ContractVersion схемы с ожидаемым 2 — при несовпадении warn-and-proceed + метрика; идемпотентность по ключу handbook-sync:{eventType}:{MessageId} — дубль → лог skipped-duplicate, ack.

A. Country (country-событие)​

  1. Контекст события. Рамка выше, eventType = country.
  2. Проверка входа. Id задан; FullName непустое (truncate + warn при > 64); IsoCode2/IsoCode3 нормализуются в upper; envelope CreatedAt задан. Существование записи не проверяется — операция upsert.
  3. Что читаем. Страна по Id (может отсутствовать — первое событие); её HandbookSyncState; для guard INV-HND-2 — наличие другой неархивной страны с тем же IsoCode2 или IsoCode3.
  4. Правила и переходы.
    • Monotonic-guard (R2/INV-HND-4): страна есть и envelope CreatedAt <= LastEventAt → событие устаревшее/повтор ресинка: метрика handbook_sync_stale_events_total, лог skipped-stale, обновить в HandbookSyncState только LastEventId/LastEventAt (отметка живости для reconciliation), к шагу 7, ack.
    • ISO-guard (R6/INV-HND-2): IsoCode2/IsoCode3 занят другой неархивной страной → отказ HANDBOOK_SYNC_ISO_CONFLICT (non-retriable → DLQ + алерт, HandbookSyncState.Status = Conflict).
    • Auto-restore (R5): запись заархивирована в PIN, но пришла из источника → IsArchived = false + warn-метрика handbook_sync_auto_restores_total (источник главнее локального архива).
    • Создание/обновление: snapshot-семантика R3 — перезапись всех синк-владеемых полей по таблице маппинга talent-replica.md; LastEventAt = envelope CreatedAt.
  5. Что меняется. Страна (создание/обновление). HandbookSyncState: при отсутствии создаётся (AggregateType = Country, ExternalId = Id), затем LastEventId = MessageId, LastEventAt = envelope CreatedAt, Status = Synced, LastError = null.
  6. События. country-upserted → pin.handbook.countries.v1 (compacted, key = Id, ContractVersion 1, полный snapshot реплики + IsArchived + LastEventAt) — контракт.
  7. Согласованность. Реплика + HandbookSyncState + отметка идемпотентности + исходящее событие — атомарно (R8); offset подтверждается только после фиксации.
  8. Результат. Успех → ack; retriable-код → retry по политике (таблица ниже); non-retriable → DLQ + HandbookSyncState.Status = Conflict/DeadLettered + алерт.

B. City (city-событие + derive региона)​

  1. Контекст события. Рамка выше, eventType = city.
  2. Проверка входа. Id задан; Name непустое (truncate + warn при > 256); Country.CountryId задан; envelope CreatedAt задан; при наличии Region — Region.RegionId и Region.Name заданы (CountryId у RegionEventModel нет — берётся из Country.CountryId события).
  3. Что читаем. Страна по Country.CountryId — отсутствует → отказ HANDBOOK_SYNC_COUNTRY_NOT_FOUND, retriable (R4): страна появится своим событием, city-событие не теряется. Далее: регион по Region.RegionId (может отсутствовать — derive создаст); город по Id (может отсутствовать); HandbookSyncState региона и города.
  4. Правила и переходы. Порядок внутри события: страна (проверка) → регион (derive) → город; guard-ы региона и города независимы (stale-город не блокирует более свежий регион, и наоборот).
    • Регион (derive, только при наличии Region):
      • INV-HND-3 GeoConsistency: страна региона берётся из Country.CountryId вмещающего city-события (RegionEventModel собственного CountryId не несёт, ADR-0063 п.4); для существующего региона это значение должно совпасть с CountryId его записи (перенос региона между странами не поддерживается); нарушение → HANDBOOK_SYNC_GEO_MISMATCH (non-retriable → DLQ + алерт).
      • Monotonic-guard (R2): envelope CreatedAt <= LastEventAt существующего региона → derive пропускается (метрика stale + отметка живости в HandbookSyncState региона).
      • Auto-restore (R5): регион заархивирован → IsArchived = false + warn-метрика.
      • Upsert: создание записи региона либо перезапись Name/CountryId; LastEventAt = envelope CreatedAt.
    • Город:
      • Monotonic-guard (R2/INV-HND-4): envelope CreatedAt <= LastEventAt → skip города (метрика stale + отметка живости), регион при этом мог примениться.
      • INV-HND-3: RegionId задан, а регион после derive не существует либо принадлежит другой стране → HANDBOOK_SYNC_GEO_MISMATCH → DLQ.
      • Auto-restore (R5): город заархивирован → IsArchived = false + warn-метрика.
      • Upsert snapshot-семантикой (R3): CountryId, RegionId, Name; LastEventAt = envelope CreatedAt.
  5. Что меняется. Регион (возможно) и город; HandbookSyncState обеих записей (AggregateType = Region и AggregateType = City — монотонность обеих по LastEventAt, отдельного поля версии нет).
  6. События. region-upserted → pin.handbook.regions.v1 — только при фактическом изменении записи региона (создание, смена Name, restore) — дедупликация шторма одинаковых событий от городов одного региона (контракт); city-upserted → pin.handbook.cities.v1 — при применённом изменении города (контракт).
  7. Согласованность. Регион + город + оба HandbookSyncState + отметка идемпотентности + оба исходящих события — одно согласованное изменение (R8).
  8. Результат. Как A8.

C. Currency (currency-событие)​

  1. Контекст события. Рамка выше, eventType = currency.
  2. Проверка входа. Id задан; Name непустое (truncate + warn при > 64); IsoCode нормализуется в upper; envelope CreatedAt задан.
  3. Что читаем. Валюта по Id; её HandbookSyncState; наличие другой неархивной валюты с тем же IsoCode (INV-HND-2).
  4. Правила и переходы. Monotonic-guard R2 (envelope CreatedAt <= LastEventAt → skip + отметка живости); ISO-guard R6 (HANDBOOK_SYNC_ISO_CONFLICT → DLQ); auto-restore R5; upsert snapshot-семантикой R3 (LastEventAt = envelope CreatedAt).
  5. Что меняется. Валюта; HandbookSyncState (AggregateType = Currency).
  6. События. currency-upserted → pin.handbook.currencies.v1 (контракт).
  7. Согласованность. Атомарно (R8).
  8. Результат. Как A8.

D. Language (language-событие)​

  1. Контекст события. Рамка выше, eventType = language.
  2. Проверка входа. Id задан; Name непустое (truncate + warn при > 64); IsoCode2/IsoCode3 нормализуются в lower, FlagIsoCode2 — в upper; envelope CreatedAt задан.
  3. Что читаем. Язык по Id; его HandbookSyncState; наличие другого неархивного языка с тем же IsoCode2 или IsoCode3 (INV-HND-2).
  4. Правила и переходы. Monotonic-guard R2 (envelope CreatedAt <= LastEventAt → skip); ISO-guard R6 (HANDBOOK_SYNC_ISO_CONFLICT → DLQ); auto-restore R5; upsert snapshot-семантикой R3 (LastEventAt = envelope CreatedAt).
  5. Что меняется. Язык; HandbookSyncState (AggregateType = Language).
  6. События. language-upserted → pin.handbook.languages.v1 (контракт); текущих подписчиков нет — топик публикуется для будущих потребителей (compacted-хранение).
  7. Согласованность. Атомарно (R8).
  8. Результат. Как A8.

E. Contact-info​

Вне scope v1

handbooks.after-contact-information-update не потребляется — ContactInfoType вне scope v1 (ADR-0063 п.3); подключение при появлении потребителя — additive (новый поток + реплика + свой pin.handbook.*-топик).

Диаграмма (поток B — самый сложный: derive региона)​

Идемпотентность и replay​

  • По событию: дедупликация по MessageId (ключи handbook-sync:{eventType}:{MessageId}) — повторная доставка after-commit-crash не меняет состояние.
  • По бизнес-правилу: monotonic-guard R2 — событие с envelope CreatedAt <= LastEventAt (для всех типов записей, включая регион) пропускается; при stale-skip обновляется отметка живости HandbookSyncState.LastEventId/LastEventAt — основа reconciliation-отчёта.
  • Replay: да, безопасен. Полный re-read топиков источника с offset 0 и re-эмиссия всех строк при пересинхронизации (POST sync-handbooks) безопасны: upsert + monotonic-guard превращают повторы в no-op реплики; runbook — talent-initial-import.md. Точечный replay DLQ — после правки данных в Talent Handbook (перевод HandbookSyncState.Status → Synced + повторная подача события).

Retry и DLQ​

DLQ-топик группы: pin.handbook.talent-sync-dlq.v1 (исходное событие + заголовки x-dlq-reason, x-dlq-error-code, x-source-topic, x-original-offset).

СбойКодRetryDLQРучное действие
Инфраструктурный (PostgreSQL/шина недоступны, timeout)—Бесконечный exponential backoff 1s→60s с паузой партиции (событие не теряется)НетВосстановить зависимость
Страна не найдена (out-of-order country→city, R4)HANDBOOK_SYNC_COUNTRY_NOT_FOUNDRetriable: 6 попыток, backoff 30s → 1m → 2m → 4m → 8m → 16mДа, после исчерпания + алертПроверить, пришло ли country-событие; при необходимости — пересинхронизация; replay из DLQ
Гео-рассогласование (INV-HND-3: CountryId вмещающего city-события не совпадает с CountryId существующей записи региона — смена страны региона)HANDBOOK_SYNC_GEO_MISMATCHНет (non-retriable)Да + алерт + HandbookSyncState.Status = ConflictПравка данных в Talent Handbook → replay из DLQ (автопочинки в PIN нет)
ISO-конфликт (INV-HND-2: код занят другой неархивной записью)HANDBOOK_SYNC_ISO_CONFLICTНет (non-retriable)Да + алерт + HandbookSyncState.Status = ConflictПравка данных в источнике → replay из DLQ
Гонка с admin archive/restore (конкурентное изменение записи)—3 попытки немедленно (перечитать запись)После исчерпанияРасследование гонки

Наблюдаемость и безопасность​

  • Logs (structured, по событию): consumed / skipped-duplicate / skipped-stale / created / updated / auto-restored / conflict / dead-lettered с полями MessageId, AggregateType, ExternalId, envelope CreatedAt, entity.LastEventAt. PII в справочных данных нет.
  • Metrics: handbook_sync_lag_seconds (задержка применения: момент применения − envelope CreatedAt, по AggregateType), handbook_sync_stale_events_total, handbook_sync_dlq_total, handbook_sync_contract_mismatch_total, handbook_sync_auto_restores_total, handbook_sync_retries_total, handbook_sync_consume_duration_ms.
  • Traces: consume-span → span обработки → spans хранилища; корреляция из метаданных пробрасывается в исходящие pin.handbook.*.
  • Alerts: DLQ > 0; HandbookSyncState.Status = Conflict > 0; лаг применения > 15 мин при ненулевом потоке; contract_mismatch > 0 (warn — сигнал смены контракта источника, ContractVersion 2 → 3).
  • Безопасность: consumer работает под system-actor; доступ к топикам Talent — m2m ACL Kafka; справочники платформенные (не tenanted), PII нет.

Пред/постусловия​

  • До обработки события. Реплика может содержать или не содержать целевую запись (операция upsert). Порядок per-record гарантирован ключом источника; кросс-справочный порядок country→city источником не гарантируется — компенсируется R4 (retry с backoff до появления страны).
  • После обработки применённого события. Целевая реплика отражает последний snapshot источника (LastEventAt = envelope CreatedAt — единый механизм для всех типов записей, включая регион; синк-владеемые поля перезаписаны; IsArchived снят при auto-restore); HandbookSyncState = Synced (либо Conflict/DeadLettered при R6/INV-HND-3/исчерпании retry); ушли соответствующие события pin.handbook.*; offset ack — только после фиксации. Устаревшее или дублирующее событие — состояние реплики не изменено (skip), отметка живости обновлена, ack. Downstream-сервисы PIN зависят только от pin.handbook.*, на топики Talent не подписаны.

Обратные ссылки​

Автоссылки: где используется этот документ.

Связанные документы​