fix(delivery): drop redundant schd event index
V9 追加删除已无消费者的 idx_evt_flid(读端不再按 (PARTITION_KEY, STATE_VERSION) 合并),并在最终 schema 断言其不存在;新增用例覆盖 schd DEAD 行被更高合法代次替换为单行 PENDING、错误字段清空、旧代次不得复活。 验证:./gradlew test 130 tests / 0 fail。
This commit is contained in:
@@ -22,3 +22,6 @@ DELETE FROM msg_event a
|
||||
CREATE UNIQUE INDEX uq_schd_event
|
||||
ON msg_event (target, partition_key)
|
||||
WHERE target = 'KAFKA:schd';
|
||||
|
||||
-- 读端不再按 (PARTITION_KEY, STATE_VERSION) 合并(单行化 + 条件确认),idx_evt_flid 已无消费者。
|
||||
DROP INDEX IF EXISTS idx_evt_flid;
|
||||
|
||||
@@ -2,6 +2,7 @@ package com.gzzn.omms.msgexchange.delivery
|
||||
|
||||
import com.gzzn.omms.msgexchange.MutableClock
|
||||
import com.gzzn.omms.msgexchange.config.PipelineProps
|
||||
import com.gzzn.omms.msgexchange.domain.ErrorClass
|
||||
import com.gzzn.omms.msgexchange.domain.EventStatus
|
||||
import com.gzzn.omms.msgexchange.domain.EventType
|
||||
import com.gzzn.omms.msgexchange.domain.MsgEvent
|
||||
@@ -181,6 +182,27 @@ class DispatcherTickTest {
|
||||
assertEquals(EventStatus.PENDING, row.state) // 旧代次的确认影响 0 行,不把新内容带走
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a higher generation replaces a dead schd row and clears the old error`() {
|
||||
val repo = StubMsgEvents()
|
||||
repo.insertAll(listOf(ev(1, Targets.KAFKA_SCHD, "F1", """{"v":"v1"}""", 1)))
|
||||
val deadId = repo.rows.values.single().eventId!!
|
||||
repo.markDead(deadId, ErrorClass.EXHAUSTED, "broker-down", attempts = 5)
|
||||
|
||||
repo.insertAll(listOf(ev(2, Targets.KAFKA_SCHD, "F1", """{"v":"v2"}""", 2)))
|
||||
|
||||
val row = repo.rows.values.single() // 仍是一行;DEAD 只保留到下一合法代次
|
||||
assertEquals(EventStatus.PENDING, row.state)
|
||||
assertEquals(2, row.stateVersion)
|
||||
assertEquals(0, row.attempts)
|
||||
assertNull(row.errorClass)
|
||||
assertNull(row.lastError)
|
||||
assertTrue(row.eventId != deadId, "接受新代次必须以新的 EVENT_ID 替换行主键")
|
||||
|
||||
repo.insertAll(listOf(ev(3, Targets.KAFKA_SCHD, "F1", """{"v":"v1-late"}""", 1)))
|
||||
assertEquals(2, repo.rows.values.single().stateVersion) // 不被接受的旧代次不得复活该行
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `KAFKA msg delivery unaffected while schd batch is retrying`() {
|
||||
val repo = StubMsgEvents()
|
||||
|
||||
+6
@@ -177,6 +177,12 @@ class FlywayMigrationTest {
|
||||
"INSERT INTO msg_event (target, partition_key, event_type, state_version, payload_json, state, attempts, created_at) " +
|
||||
"VALUES ('KAFKA:msg', 'F1', 'UPSERT', 2, '{}', 'PENDING', 0, now())",
|
||||
)
|
||||
stmt.executeQuery(
|
||||
"SELECT count(*) FROM pg_indexes WHERE tablename = 'msg_event' AND indexname = 'idx_evt_flid'",
|
||||
).use { rs ->
|
||||
assertTrue(rs.next())
|
||||
assertEquals(0, rs.getInt(1), "V9 必须删除已无消费者的 idx_evt_flid")
|
||||
}
|
||||
|
||||
// V6:入队时间是超期判据 R 的比较对象,必须非空(received_at 则允许为 NULL)
|
||||
stmt.executeQuery(
|
||||
|
||||
Reference in New Issue
Block a user