fix(acm2-96): schd 单条上限超限分轮，文档与实现对齐 C-9
This commit is contained in:
windyboy
@@ -42,7 +42,7 @@ AODB 经 CIIMS adapter 把 XML 报文写入 `CMINMSGS`，格式以架构指定
|
||||
| 主题 | 已确定的消息语义 | 尚需确定 |
|
||||
|---|---|---|
|
||||
| `msg` | 单条航班变更通知；Kafka value 是整条 `MSG` 的 JSON（`META` 加对应业务体，空字段不输出），变更与删除由 `META` 的类型与子类型区分（`Q5`）；航班动态与删除处理完成后投递；发送失败自动重试，一直失败的记录保留可查并告警（`US-08`）；同一 `FLID` 的变更保序，对外按至少一次投递（`CLM-3`，单分区）。删除通知的来源有三处：`FDEL` 删除（`US-06`）、日计划完整名单覆盖范围内缺席删除（`US-07`）、历史清理在物理删除前必要时登记（架构 `D1`）。 | 编码方式（JSON 之外的压缩或封装是否引入）。 |
|
||||
| `schd` | 定时批量发送最新航班状态；两次 tick 之间积累的航班组成 `SCHD.FLTR` 数组 JSON，整批作为单条 record 发出，条数不设上限（沿用旧系统，`Q1`）；空字段不输出，字段与类型见 [XSD](../legacy/unisysaodbsis.xsd) 的 `FLTR`；没有变化不发；删除航班不进入本主题，由 `msg` 发一条删除通知；发送失败自动重试，对外按至少一次投递（`US-08`）。 | — |
|
||||
| `schd` | 定时批量发送最新航班状态；待发航班按单批上限聚成 `SCHD.FLTR` 数组 JSON，每批作为单条 record 发出，超限部分后续轮次发出（`C-9`、`Q1`）；空字段不输出，字段与类型见 [XSD](../legacy/unisysaodbsis.xsd) 的 `FLTR`；没有变化不发；删除航班不进入本主题，由 `msg` 发一条删除通知；发送失败自动重试，对外按至少一次投递（`US-08`）。 | — |
|
||||
|
||||
旧系统线索（来源：旧项目用户故事「前端通知」「动态类（FLOP-*）处理」）：
|
||||
|
||||
|
||||
@@ -231,12 +231,12 @@ PENDING → SENT → DONE
|
||||
|
||||
### 8.2 `KAFKA:schd`
|
||||
|
||||
按 `FLID` 单行 upsert，只保留最新 `STATE_VERSION`。每次更新换 `EVENT_ID`（写代次）。
|
||||
写入侧按 `FLID` 单行 upsert，只保留最新 `STATE_VERSION`。每次更新换 `EVENT_ID`（写代次）。
|
||||
|
||||
- **只进不退**：`STATE_VERSION` 只升不降。
|
||||
- **条件标记**：`SENT` 时 `WHERE STATE_VERSION = 本批版本`；已被新版本覆盖则下轮重发。
|
||||
|
||||
发送：取未发送行（`PARAM:msgx.schd.flush-limit`）→ 把各行载荷聚成 `SCHD.FLTR` JSON 数组发出（`C-9`）→ 条件标 `SENT`。`MSG_EVENT` 里每条 `schd` 载荷、以及 Redis 投影 value，都是单条 `FLTR` JSON（由 PG 当前态重建，`C-11`）；删航班只走 `msg`。
|
||||
发送侧取未发送行（上限见 `PARAM:msgx.schd.flush-limit`）→ 把本批各行载荷聚成一条 `SCHD.FLTR` JSON 数组发出（`C-9`）→ 条件标 `SENT`；超限的剩余行留待后续轮次发出。日计划整批也登记 `schd` 事件，旧系统只发散单变更。`MSG_EVENT` 里每条 `schd` 载荷、以及 Redis 投影 value，都是单条 `FLTR` JSON（由 PG 当前态重建，`C-11`）；删航班只走 `msg`。
|
||||
|
||||
周期与批上限见 [reference.md](reference.md)。`msg` 与 `schd` 不承诺顺序。
|
||||
|
||||
|
||||
+1
-1
@@ -18,7 +18,7 @@
|
||||
| `msgx.pipeline.delivery-batch` | `200` 条 | 一次最多领取多少条待发的 `msg` 消息 | 暂定 |
|
||||
| `msgx.pipeline.delivery-drain-rounds` | `10` 轮 | 连续领取多少批后，先尝试发送一批 `schd` 消息 | 暂定 |
|
||||
| `msgx.schd.flush-period` | `3s` | 每隔多久尝试发送一批 `schd` 消息 | 沿用旧系统 |
|
||||
| `msgx.schd.flush-limit` | `500` 条 | 一批 `schd` 消息最多包含多少个航班 | 暂定 |
|
||||
| `msgx.schd.flush-limit` | `500` 条 | 单条 `schd` 消息最多包含多少个航班；超限部分留待后续轮次发出 | 暂定 |
|
||||
| `msgx.schd.snapshot-batch` | `200` 个航班 | 日计划一批最多写多少个航班，每批一个事务；缺席清扫按同一上限分批（`US-07` AC4） | 暂定 |
|
||||
| `msgx.pipeline.event-retention` | `7d` | 发送成功的消息在数据库保留多久，从 `SENT_AT` 起算 | 暂定 |
|
||||
| `msgx.pipeline.terminal-retention` | `90d` | 处理记录（PROC_STATE 终态行）保留多久，从 `UPDATED_AT` 起算；到期且已回填才删除（`US-11`） | 暂定 |
|
||||
|
||||
@@ -60,7 +60,7 @@
|
||||
|
||||
### 2.4 下游（admin-api、网页客户端）
|
||||
|
||||
- **C-9** 航班变更发 Kafka：`msg` 一条变更一条消息，消息体是整条 `MSG` 的 JSON（`META` 加对应业务体），变更与删除由 `META` 的类型与子类型区分（`Q5`）；`schd` 把两轮采集之间积累的变化合成一条消息，内容是 `SCHD.FLTR` JSON 数组，条数不设上限（`Q1`）。空字段不输出，没变化不发；删航班只走 `msg`；不设 message key；可能重复投递。
|
||||
- **C-9** 航班变更发 Kafka：`msg` 一条变更一条消息，消息体是整条 `MSG` 的 JSON（`META` 加对应业务体），变更与删除由 `META` 的类型与子类型区分（`Q5`）；`schd` 把积累的变化按批发出，每批一条消息、内容是 `SCHD.FLTR` JSON 数组，单批条数上限见 `PARAM:msgx.schd.flush-limit`，超限部分留待后续轮次发出。空字段不输出，没变化不发；删航班只走 `msg`；不设 message key；可能重复投递。
|
||||
- **C-10** 静态参考数据写本系统库，admin-api 只读；本系统不调用 admin-api。
|
||||
- **C-11** 航班快照写 Redis；网页客户端经 `GET /all/flights` 读这一份，不直连 Redis；共享航班（`MAID` 非空）不单列返回（是否以主航班 `MAFL` 提供见 `Q6`）。成功 HTTP 200 返回裸 JSON 数组（不套旧 `ResponseDto`），不分页；失败 HTTP 503，JSON 对象 `{"error":"FLIGHT_PROJECTION_UNAVAILABLE","reason":"<细节>"}`；投影读失败不得返回 200 空数组（`INV-11`）。投影 value、`KAFKA:schd` 数组元素与查询成功体元素同形：都是日计划一条 `FLTR` 转成的 JSON（字段名与 [XSD](legacy/unisysaodbsis.xsd) `FLTR` 一致，空字段不输出，与 `C-9` 相同）；由自有 PG 当前态重建，不是另起包装。
|
||||
|
||||
@@ -112,7 +112,7 @@
|
||||
|
||||
| 编号 | 状态 | 事项 | 说明 |
|
||||
|---|---|---|---|
|
||||
| Q1 | 已定 | `schd` 一条 Kafka 消息里装多少变更 | 两轮发送之间积累的全部变更合成一条，条数不设上限（`C-9`） |
|
||||
| Q1 | 已定 | `schd` 一条 Kafka 消息里装多少变更 | 单批上限内的待发航班聚成一条 `SCHD.FLTR` JSON 数组，超限部分后续轮次发出（`C-9`） |
|
||||
| Q2 | 待对方 | FLOP 里 `SRVT`/`VIPF` 等集合段缺席是否等于删除 | 见 `G-SRVT-VIPF`；旧系统不解析这两段，无法对拍 |
|
||||
| Q3 | 待对方 | SIS 没写明的 FLOP 子类型怎么解析、处理完算啥 | 按现有逻辑处理（`US-05` AC1）；见 `G-FLOP-SEMANTICS`、`CLM-1`。旧系统找不到对应处理器时按失败兜底，仍写回处理时间且无业务改动 |
|
||||
| Q4 | 待对方 | 上游重发时会不会改正文 | 假定内容不变，只认身份；改正文会被当重复跳过。旧系统不比较正文 |
|
||||
|
||||
@@ -23,7 +23,7 @@ interface DeliveryPort {
|
||||
*/
|
||||
fun sendKafkaSchd(topic: String, payloadJson: String)
|
||||
|
||||
/** 发一条删除通知：key 是 FLID、value 为空；下游按"整态里这个键没了"理解成删除。 */
|
||||
/** 遗留的空 value 删除通知（tombstone）；当前删除只走 `KAFKA:msg` 的 UPSERT（`C-9`），生产路径不再调用。 */
|
||||
fun sendKafkaNull(topic: String, key: String)
|
||||
|
||||
/** 给健康检查用的连通性探测；默认返回 true，真实 Kafka 实现要覆写成向 broker 拉一次 metadata 来判断。 */
|
||||
@@ -156,8 +156,8 @@ class Dispatcher(
|
||||
}
|
||||
|
||||
/**
|
||||
* 发 `KAFKA:schd`：本 tick 窗口内待发 UPSERT 聚成一条 `SCHD.FLTR` JSON 数组、不设 key（`C-9`）。
|
||||
* 空窗口不发。schd 侧 TOMBSTONE 不发送（删航班只走 msg）。
|
||||
* 发 `KAFKA:schd`：本轮待发 UPSERT 取上限（`PARAM:msgx.schd.flush-limit`）聚成一条 `SCHD.FLTR` JSON 数组、不设 key（`C-9`）；
|
||||
* 超限剩余行留待后续轮次发出。空窗口不发。schd 侧 TOMBSTONE 不发送（删航班只走 msg）。
|
||||
*/
|
||||
internal fun flushSchd() {
|
||||
val batch = try {
|
||||
|
||||
@@ -2,7 +2,7 @@ package com.gzzn.omms.msgexchange.domain
|
||||
|
||||
/**
|
||||
* 事件形态。生产路径只写 [UPSERT]：删除通知也走 `KAFKA:msg` 的 UPSERT 变更（`C-9`）。
|
||||
* [TOMBSTONE] 仍保留在枚举与投递分支里，当前无生产写入方（待 `Q5` 定稿清理或保留）。
|
||||
* [TOMBSTONE] 仅枚举与写入守卫保留兼容语义，当前无生产写入方、不发送。
|
||||
*/
|
||||
enum class EventType { UPSERT, TOMBSTONE }
|
||||
|
||||
|
||||
@@ -2,7 +2,8 @@ package com.gzzn.omms.msgexchange.domain
|
||||
|
||||
/**
|
||||
* 事件往哪儿投。KAFKA_SCHD 按 FLID（航班实例 ID）发航班的完整状态，KAFKA_MSG 只发"变化了"
|
||||
* 的通知；两个主题之间不保证顺序，删除用 value 为空的 tombstone 消息表示。
|
||||
* 的通知；两个主题之间不保证顺序。删航班只走 `KAFKA:msg` 的 UPSERT 变更（`C-9`），
|
||||
* 不发空 value 的 tombstone。
|
||||
*
|
||||
* 常量值就是 MSG_EVENT.TARGET 落库的值，真正的 topic 名由投递适配层映射
|
||||
* （KAFKA:msg → "msg"，KAFKA:schd → "schd"）。ES 投影还没做，先不登记目标。
|
||||
|
||||
@@ -95,6 +95,35 @@ class DispatcherTickTest {
|
||||
assertEquals(0, repo.rows.values.count { it.state.name == "PENDING" })
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `schd overflow beyond flush-limit is sent in follow-up rounds`() {
|
||||
val repo = StubMsgEvents()
|
||||
val port = StubDeliveryPort()
|
||||
val props = PipelineProps().apply { schd.flushLimit = 2 }
|
||||
repo.insertAll(
|
||||
listOf(
|
||||
ev(1, Targets.KAFKA_SCHD, "F1", """{"v":"f1"}""", 1),
|
||||
ev(2, Targets.KAFKA_SCHD, "F2", """{"v":"f2"}""", 1),
|
||||
ev(3, Targets.KAFKA_SCHD, "F3", """{"v":"f3"}""", 1),
|
||||
),
|
||||
)
|
||||
val d = dispatcher(repo, port, props)
|
||||
|
||||
d.flushSchd()
|
||||
val first = port.sent.filter { it.topic == "schd" }
|
||||
assertEquals(1, first.size)
|
||||
assertNull(first.single().key)
|
||||
assertTrue(first.single().payload!!.contains("f1"))
|
||||
assertTrue(first.single().payload!!.contains("f2"))
|
||||
assertEquals(1, repo.rows.values.count { it.state == EventStatus.PENDING })
|
||||
|
||||
d.flushSchd()
|
||||
val all = port.sent.filter { it.topic == "schd" }
|
||||
assertEquals(2, all.size)
|
||||
assertTrue(all.last().payload!!.contains("f3"))
|
||||
assertEquals(0, repo.rows.values.count { it.state == EventStatus.PENDING })
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `schd retry is not selected before its backoff expires`() {
|
||||
val repo = StubMsgEvents()
|
||||
|
||||
Reference in New Issue
Block a user