From e182b30c58824a4153cc2fe92db463028194eb80 Mon Sep 17 00:00:00 2001 From: "Zimin A.N." Date: Mon, 4 May 2026 15:43:56 +0300 Subject: [PATCH] =?UTF-8?q?feat(outbox):=20attempt=20cap=20=E2=86=92=20DLQ?= =?UTF-8?q?=20+=20admin=20endpoint=20+=206=20unit-=D1=82=D0=B5=D1=81=D1=82?= =?UTF-8?q?=D0=BE=D0=B2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit После N (default 1000) неудачных попыток публикации событие маркируется dlq_at и исключается из polling. Без cap poller бесконечно ретраил мёртвые сообщения (orphan topic, ACL deny), забивая логи и lag-метрику. - 0011-outbox-dlq.xml: dlq_at column + перестроенный idx_outbox_unpublished (исключает DLQ) + idx_outbox_dlq для admin-листинга. - OutboxPoller: ordinis.outbox.attempt-max (default 1000), markDlq() при достижении cap, новый counter ordinis_outbox_dlq_total. - OutboxLagGauge: ordinis_outbox_dlq_size gauge — алерт на > 0. - OutboxEventRepository: countPending() (без DLQ), findByDlqAtIsNotNull(Pageable). - OutboxAdminController: GET /admin/outbox/{stats,dlq} для UI. - 6 unit-тестов через mock-maker-subclass (JDK 25 ломает Mockito inline). --- .../src/main/resources/application.yml | 14 ++ .../ordinis/domain/outbox/OutboxEvent.java | 13 ++ .../domain/outbox/OutboxEventRepository.java | 30 +++- .../db/changelog/changes/0011-outbox-dlq.xml | 51 ++++++ .../main/resources/db/changelog/master.xml | 1 + .../ordinis/outbox/OutboxLagGauge.java | 18 +- .../ordinis/outbox/OutboxPoller.java | 29 ++- .../ordinis/outbox/OutboxPollerTest.java | 170 ++++++++++++++++++ .../org.mockito.plugins.MockMaker | 1 + .../restapi/web/OutboxAdminController.java | 73 ++++++++ 10 files changed, 391 insertions(+), 9 deletions(-) create mode 100644 ordinis-migrations/src/main/resources/db/changelog/changes/0011-outbox-dlq.xml create mode 100644 ordinis-outbox/src/test/java/cloud/nstart/terravault/ordinis/outbox/OutboxPollerTest.java create mode 100644 ordinis-outbox/src/test/resources/mockito-extensions/org.mockito.plugins.MockMaker create mode 100644 ordinis-rest-api/src/main/java/cloud/nstart/terravault/ordinis/restapi/web/OutboxAdminController.java diff --git a/ordinis-app/src/main/resources/application.yml b/ordinis-app/src/main/resources/application.yml index 954f913..b8a6bb7 100644 --- a/ordinis-app/src/main/resources/application.yml +++ b/ordinis-app/src/main/resources/application.yml @@ -57,6 +57,20 @@ ordinis: enabled: ${ORDINIS_OUTBOX_ENABLED:true} poll-interval-ms: ${ORDINIS_OUTBOX_POLL_INTERVAL_MS:1000} batch-size: ${ORDINIS_OUTBOX_BATCH_SIZE:100} + # Cap на ретраи: после N попыток событие → DLQ (dlq_at), poller его пропускает. + # 1000 при poll=1s ≈ ~17 минут активных ошибок (или быстрее на fail-fast топиках). + attempt-max: ${ORDINIS_OUTBOX_ATTEMPT_MAX:1000} + +springdoc: + api-docs: + path: /v3/api-docs + swagger-ui: + path: /swagger-ui.html + operations-sorter: method + tags-sorter: alpha + display-request-duration: true + show-actuator: false + packages-to-scan: cloud.nstart.terravault.ordinis.restapi.web --- # DEV profile: подключается к cluster PG+Kafka через kubectl port-forward. diff --git a/ordinis-domain/src/main/java/cloud/nstart/terravault/ordinis/domain/outbox/OutboxEvent.java b/ordinis-domain/src/main/java/cloud/nstart/terravault/ordinis/domain/outbox/OutboxEvent.java index fd74254..5ef7297 100644 --- a/ordinis-domain/src/main/java/cloud/nstart/terravault/ordinis/domain/outbox/OutboxEvent.java +++ b/ordinis-domain/src/main/java/cloud/nstart/terravault/ordinis/domain/outbox/OutboxEvent.java @@ -73,6 +73,9 @@ public class OutboxEvent { @Column(name = "last_error", columnDefinition = "TEXT") private String lastError; + @Column(name = "dlq_at") + private OffsetDateTime dlqAt; + protected OutboxEvent() {} public OutboxEvent( @@ -103,6 +106,15 @@ public class OutboxEvent { this.lastError = error; } + public void markDlq(String error) { + this.lastError = error; + this.dlqAt = OffsetDateTime.now(); + } + + public boolean isDlq() { + return dlqAt != null; + } + public Long getId() { return id; } public String getEventType() { return eventType; } public String getAggregateType() { return aggregateType; } @@ -118,6 +130,7 @@ public class OutboxEvent { public String getSpanId() { return spanId; } public Integer getAttemptCount() { return attemptCount; } public String getLastError() { return lastError; } + public OffsetDateTime getDlqAt() { return dlqAt; } public void setDictionaryName(String v) { this.dictionaryName = v; } public void setTraceId(String v) { this.traceId = v; } diff --git a/ordinis-domain/src/main/java/cloud/nstart/terravault/ordinis/domain/outbox/OutboxEventRepository.java b/ordinis-domain/src/main/java/cloud/nstart/terravault/ordinis/domain/outbox/OutboxEventRepository.java index 0fc740b..7211f74 100644 --- a/ordinis-domain/src/main/java/cloud/nstart/terravault/ordinis/domain/outbox/OutboxEventRepository.java +++ b/ordinis-domain/src/main/java/cloud/nstart/terravault/ordinis/domain/outbox/OutboxEventRepository.java @@ -1,5 +1,6 @@ package cloud.nstart.terravault.ordinis.domain.outbox; +import org.springframework.data.domain.Page; import org.springframework.data.domain.Pageable; import org.springframework.data.jpa.repository.JpaRepository; import org.springframework.data.jpa.repository.Query; @@ -9,10 +10,35 @@ import java.util.List; public interface OutboxEventRepository extends JpaRepository { /** - * Unpublished events для polling. Использует partial index {@code idx_outbox_unpublished}. + * Unpublished events для polling. Использует partial index + * {@code idx_outbox_unpublished} (where published_at IS NULL AND dlq_at IS NULL). */ - @Query("SELECT o FROM OutboxEvent o WHERE o.publishedAt IS NULL ORDER BY o.id ASC") + @Query(""" + SELECT o FROM OutboxEvent o + WHERE o.publishedAt IS NULL AND o.dlqAt IS NULL + ORDER BY o.id ASC + """) List findUnpublished(Pageable pageable); + /** + * Сколько ожидает публикации (без DLQ). Метрика + * {@code ordinis_outbox_pending_events}. + */ + @Query("SELECT COUNT(o) FROM OutboxEvent o WHERE o.publishedAt IS NULL AND o.dlqAt IS NULL") + long countPending(); + + /** + * Сколько в DLQ. Метрика {@code ordinis_outbox_dlq_size} — растёт ⇒ алерт. + */ + long countByDlqAtIsNotNull(); + + /** + * DLQ-листинг для admin UI. Партиальный индекс {@code idx_outbox_dlq} по + * {@code dlq_at DESC}. + */ + Page findByDlqAtIsNotNull(Pageable pageable); + + /** @deprecated используйте {@link #countPending()} (не считает DLQ). */ + @Deprecated long countByPublishedAtIsNull(); } diff --git a/ordinis-migrations/src/main/resources/db/changelog/changes/0011-outbox-dlq.xml b/ordinis-migrations/src/main/resources/db/changelog/changes/0011-outbox-dlq.xml new file mode 100644 index 0000000..6f2eb85 --- /dev/null +++ b/ordinis-migrations/src/main/resources/db/changelog/changes/0011-outbox-dlq.xml @@ -0,0 +1,51 @@ + + + + + + + + + + + + + + Pending index должен исключать dlq_at — иначе poller продолжит "видеть" + DLQ-записи и ordinis_outbox_pending_events будет показывать их в lag. + + + DROP INDEX IF EXISTS idx_outbox_unpublished; + CREATE INDEX idx_outbox_unpublished ON outbox_events (created_at) + WHERE published_at IS NULL AND dlq_at IS NULL; + + + + DROP INDEX IF EXISTS idx_outbox_unpublished; + CREATE INDEX idx_outbox_unpublished ON outbox_events (created_at) + WHERE published_at IS NULL; + + + + + + Партиальный индекс для быстрого листинга DLQ в admin UI. + + CREATE INDEX idx_outbox_dlq ON outbox_events (dlq_at DESC) + WHERE dlq_at IS NOT NULL; + + + DROP INDEX IF EXISTS idx_outbox_dlq; + + + + diff --git a/ordinis-migrations/src/main/resources/db/changelog/master.xml b/ordinis-migrations/src/main/resources/db/changelog/master.xml index c20282f..6bc0982 100644 --- a/ordinis-migrations/src/main/resources/db/changelog/master.xml +++ b/ordinis-migrations/src/main/resources/db/changelog/master.xml @@ -15,5 +15,6 @@ + diff --git a/ordinis-outbox/src/main/java/cloud/nstart/terravault/ordinis/outbox/OutboxLagGauge.java b/ordinis-outbox/src/main/java/cloud/nstart/terravault/ordinis/outbox/OutboxLagGauge.java index e523961..70891ac 100644 --- a/ordinis-outbox/src/main/java/cloud/nstart/terravault/ordinis/outbox/OutboxLagGauge.java +++ b/ordinis-outbox/src/main/java/cloud/nstart/terravault/ordinis/outbox/OutboxLagGauge.java @@ -8,8 +8,13 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.stereotype.Component; /** - * Gauge {@code ordinis_outbox_pending_events} — сколько unpublished. Алертится - * Prometheus'ом если растёт (poller не успевает или Kafka недоступна). + * Gauges: + *
    + *
  • {@code ordinis_outbox_pending_events} — unpublished без DLQ. Растёт ⇒ + * poller не успевает либо Kafka недоступна.
    • + *
  • {@code ordinis_outbox_dlq_size} — события в DLQ. Любое значение > 0 ⇒ + * требуется ручной разбор (orphan topic, ACL, broken payload).
    • + *
*/ @Component @ConditionalOnProperty(name = "ordinis.outbox.enabled", havingValue = "true", matchIfMissing = false) @@ -26,8 +31,13 @@ public class OutboxLagGauge { @PostConstruct void register() { Gauge.builder("ordinis_outbox_pending_events", repository, - r -> (double) r.countByPublishedAtIsNull()) - .description("Количество outbox events ожидающих публикацию в Kafka") + r -> (double) r.countPending()) + .description("Количество outbox events ожидающих публикацию (без DLQ)") + .register(meterRegistry); + + Gauge.builder("ordinis_outbox_dlq_size", repository, + r -> (double) r.countByDlqAtIsNotNull()) + .description("Количество outbox events в DLQ (требуют ручного разбора)") .register(meterRegistry); } } diff --git a/ordinis-outbox/src/main/java/cloud/nstart/terravault/ordinis/outbox/OutboxPoller.java b/ordinis-outbox/src/main/java/cloud/nstart/terravault/ordinis/outbox/OutboxPoller.java index 1f1542b..3f38d0b 100644 --- a/ordinis-outbox/src/main/java/cloud/nstart/terravault/ordinis/outbox/OutboxPoller.java +++ b/ordinis-outbox/src/main/java/cloud/nstart/terravault/ordinis/outbox/OutboxPoller.java @@ -25,6 +25,11 @@ import java.util.concurrent.TimeoutException; * @Scheduled poller для outbox. Sweep каждые {@code ordinis.outbox.poll-interval-ms}, * batch size {@code ordinis.outbox.batch-size}. Активируется только если * {@code ordinis.outbox.enabled=true} (отключаем в read-api / projection-writer). + * + *

После {@code ordinis.outbox.attempt-max} (default 1000) неудачных попыток + * событие маркируется {@code dlq_at} и исключается из дальнейшего polling. Без + * cap poller бесконечно ретраит мёртвые сообщения (orphan topic, ACL deny), + * забивая лог и lag-метрику. */ @Component @ConditionalOnProperty(name = "ordinis.outbox.enabled", havingValue = "true", matchIfMissing = false) @@ -37,6 +42,7 @@ public class OutboxPoller { private final com.fasterxml.jackson.databind.ObjectMapper objectMapper; private final Counter publishedCounter; private final Counter failedCounter; + private final Counter dlqCounter; private final Timer publishTimer; @Value("${ordinis.outbox.batch-size:100}") @@ -45,6 +51,9 @@ public class OutboxPoller { @Value("${ordinis.outbox.send-timeout-ms:10000}") private long sendTimeoutMs; + @Value("${ordinis.outbox.attempt-max:1000}") + private int attemptMax; + public OutboxPoller( OutboxEventRepository repository, KafkaTemplate kafkaTemplate, @@ -59,6 +68,9 @@ public class OutboxPoller { this.failedCounter = Counter.builder("ordinis_outbox_publish_errors_total") .description("Количество ошибок публикации в Kafka") .register(meterRegistry); + this.dlqCounter = Counter.builder("ordinis_outbox_dlq_total") + .description("Количество событий перенесённых в DLQ (attempt cap)") + .register(meterRegistry); this.publishTimer = Timer.builder("ordinis_kafka_publish_duration_seconds") .description("Время публикации одного события в Kafka") .register(meterRegistry); @@ -103,10 +115,21 @@ public class OutboxPoller { } private void markFailed(OutboxEvent event, String error) { - log.warn("Failed to publish outbox event id={} topic={}: {}", - event.getId(), event.getKafkaTopic(), error); event.markFailed(error); - repository.save(event); failedCounter.increment(); + + if (event.getAttemptCount() != null && event.getAttemptCount() >= attemptMax) { + event.markDlq(error); + dlqCounter.increment(); + log.error( + "Outbox event id={} topic={} → DLQ after {} attempts. Last error: {}", + event.getId(), event.getKafkaTopic(), event.getAttemptCount(), error); + } else { + log.warn( + "Failed to publish outbox event id={} topic={} attempt={}/{}: {}", + event.getId(), event.getKafkaTopic(), + event.getAttemptCount(), attemptMax, error); + } + repository.save(event); } } diff --git a/ordinis-outbox/src/test/java/cloud/nstart/terravault/ordinis/outbox/OutboxPollerTest.java b/ordinis-outbox/src/test/java/cloud/nstart/terravault/ordinis/outbox/OutboxPollerTest.java new file mode 100644 index 0000000..72da08e --- /dev/null +++ b/ordinis-outbox/src/test/java/cloud/nstart/terravault/ordinis/outbox/OutboxPollerTest.java @@ -0,0 +1,170 @@ +package cloud.nstart.terravault.ordinis.outbox; + +import cloud.nstart.terravault.ordinis.domain.DataScope; +import cloud.nstart.terravault.ordinis.domain.outbox.OutboxEvent; +import cloud.nstart.terravault.ordinis.domain.outbox.OutboxEventRepository; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.JsonNodeFactory; +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; +import org.apache.kafka.clients.producer.RecordMetadata; +import org.apache.kafka.common.TopicPartition; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.data.domain.Pageable; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.support.SendResult; +import org.springframework.test.util.ReflectionTestUtils; + +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeoutException; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +class OutboxPollerTest { + + private OutboxEventRepository repo; + private KafkaTemplate kafka; + private OutboxPoller poller; + + @BeforeEach + void setUp() { + repo = mock(OutboxEventRepository.class); + kafka = mock(KafkaTemplate.class); + poller = new OutboxPoller(repo, kafka, new ObjectMapper(), new SimpleMeterRegistry()); + ReflectionTestUtils.setField(poller, "batchSize", 100); + ReflectionTestUtils.setField(poller, "sendTimeoutMs", 1_000L); + ReflectionTestUtils.setField(poller, "attemptMax", 3); + } + + private OutboxEvent newEvent() { + return new OutboxEvent( + "DICT_RECORD_CREATED", + "DictionaryRecord", + "spacecraft:RESURS-P-3", + DataScope.PUBLIC, + JsonNodeFactory.instance.objectNode().put("foo", "bar"), + "ordinis.cuod.events.public", + "spacecraft:RESURS-P-3"); + } + + @Test + void empty_batch_does_nothing() { + when(repo.findUnpublished(any(Pageable.class))).thenReturn(List.of()); + + poller.poll(); + + verify(kafka, never()).send(anyString(), anyString(), anyString()); + verify(repo, never()).save(any()); + } + + @Test + void successful_publish_marks_published() { + OutboxEvent event = newEvent(); + when(repo.findUnpublished(any(Pageable.class))).thenReturn(List.of(event)); + + SendResult result = mock(SendResult.class); + when(result.getRecordMetadata()).thenReturn( + new RecordMetadata(new TopicPartition("ordinis.cuod.events.public", 0), 0L, 0, 0L, 0, 0)); + when(kafka.send(eq("ordinis.cuod.events.public"), anyString(), anyString())) + .thenReturn(CompletableFuture.completedFuture(result)); + + poller.poll(); + + assertThat(event.getPublishedAt()).isNotNull(); + assertThat(event.getDlqAt()).isNull(); + verify(repo).save(event); + } + + @Test + void failure_below_cap_marks_failed_not_dlq() { + OutboxEvent event = newEvent(); + when(repo.findUnpublished(any(Pageable.class))).thenReturn(List.of(event)); + + CompletableFuture> failed = new CompletableFuture<>(); + failed.completeExceptionally(new TimeoutException("kafka down")); + when(kafka.send(anyString(), anyString(), anyString())).thenReturn(failed); + + poller.poll(); + + assertThat(event.getAttemptCount()).isEqualTo(1); + assertThat(event.getDlqAt()).isNull(); + assertThat(event.getLastError()).contains("kafka down"); + verify(repo).save(event); + } + + @Test + void failure_at_cap_marks_dlq() { + OutboxEvent event = newEvent(); + // Симулируем что 2 предыдущих попытки уже были. + event.markFailed("prev 1"); + event.markFailed("prev 2"); + assertThat(event.getAttemptCount()).isEqualTo(2); + + when(repo.findUnpublished(any(Pageable.class))).thenReturn(List.of(event)); + + CompletableFuture> failed = new CompletableFuture<>(); + failed.completeExceptionally(new TimeoutException("still down")); + when(kafka.send(anyString(), anyString(), anyString())).thenReturn(failed); + + poller.poll(); + + assertThat(event.getAttemptCount()).isEqualTo(3); + assertThat(event.getDlqAt()).isNotNull(); + assertThat(event.isDlq()).isTrue(); + assertThat(event.getLastError()).contains("still down"); + verify(repo).save(event); + } + + @Test + void failure_above_cap_stays_dlq_not_re_marked() { + OutboxEvent event = newEvent(); + for (int i = 0; i < 3; i++) event.markFailed("err " + i); + event.markDlq("first dlq"); + java.time.OffsetDateTime firstDlqAt = event.getDlqAt(); + + when(repo.findUnpublished(any(Pageable.class))).thenReturn(List.of()); + + poller.poll(); + + // Poller не должен подобрать DLQ событие — findUnpublished фильтрует dlq_at IS NULL. + assertThat(event.getDlqAt()).isEqualTo(firstDlqAt); + } + + @Test + void mixed_batch_publishes_some_dlqs_others() { + OutboxEvent ok = newEvent(); + OutboxEvent doomed = newEvent(); + doomed.markFailed("p1"); + doomed.markFailed("p2"); + + when(repo.findUnpublished(any(Pageable.class))).thenReturn(List.of(ok, doomed)); + + SendResult result = mock(SendResult.class); + when(result.getRecordMetadata()).thenReturn( + new RecordMetadata(new TopicPartition("ordinis.cuod.events.public", 0), 0L, 0, 0L, 0, 0)); + CompletableFuture> okFut = CompletableFuture.completedFuture(result); + CompletableFuture> badFut = new CompletableFuture<>(); + badFut.completeExceptionally(new TimeoutException("doomed")); + + // Первый event получает success, второй — failure. + when(kafka.send(anyString(), anyString(), anyString())) + .thenReturn(okFut) + .thenReturn(badFut); + + poller.poll(); + + assertThat(ok.getPublishedAt()).isNotNull(); + assertThat(ok.getDlqAt()).isNull(); + assertThat(doomed.getDlqAt()).isNotNull(); + verify(repo, times(2)).save(any()); + } +} diff --git a/ordinis-outbox/src/test/resources/mockito-extensions/org.mockito.plugins.MockMaker b/ordinis-outbox/src/test/resources/mockito-extensions/org.mockito.plugins.MockMaker new file mode 100644 index 0000000..fdbd0b1 --- /dev/null +++ b/ordinis-outbox/src/test/resources/mockito-extensions/org.mockito.plugins.MockMaker @@ -0,0 +1 @@ +mock-maker-subclass diff --git a/ordinis-rest-api/src/main/java/cloud/nstart/terravault/ordinis/restapi/web/OutboxAdminController.java b/ordinis-rest-api/src/main/java/cloud/nstart/terravault/ordinis/restapi/web/OutboxAdminController.java new file mode 100644 index 0000000..145999b --- /dev/null +++ b/ordinis-rest-api/src/main/java/cloud/nstart/terravault/ordinis/restapi/web/OutboxAdminController.java @@ -0,0 +1,73 @@ +package cloud.nstart.terravault.ordinis.restapi.web; + +import cloud.nstart.terravault.ordinis.domain.outbox.OutboxEvent; +import cloud.nstart.terravault.ordinis.domain.outbox.OutboxEventRepository; +import org.springframework.data.domain.Page; +import org.springframework.data.domain.PageRequest; +import org.springframework.data.domain.Sort; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RequestParam; +import org.springframework.web.bind.annotation.RestController; + +import java.time.OffsetDateTime; +import java.util.List; +import java.util.Map; + +/** + * Admin endpoints для outbox observability. Только чтение — replay делается + * вручную через SQL (UPDATE outbox_events SET dlq_at=NULL, attempt_count=0 WHERE id=...). + * Авто-replay сделаем когда появится первый реальный кейс DLQ. + */ +@RestController +@RequestMapping("/api/v1/admin/outbox") +public class OutboxAdminController { + + private final OutboxEventRepository repository; + + public OutboxAdminController(OutboxEventRepository repository) { + this.repository = repository; + } + + @GetMapping("/stats") + public Map stats() { + return Map.of( + "pending", repository.countPending(), + "dlq", repository.countByDlqAtIsNotNull() + ); + } + + @GetMapping("/dlq") + public Page dlq( + @RequestParam(defaultValue = "0") int page, + @RequestParam(defaultValue = "50") int size) { + Page events = repository.findByDlqAtIsNotNull( + PageRequest.of(page, Math.min(size, 200), Sort.by(Sort.Direction.DESC, "dlqAt"))); + return events.map(DlqEntry::from); + } + + public record DlqEntry( + Long id, + String eventType, + String dictionaryName, + String aggregateId, + String kafkaTopic, + Integer attemptCount, + String lastError, + OffsetDateTime createdAt, + OffsetDateTime dlqAt) { + + static DlqEntry from(OutboxEvent e) { + return new DlqEntry( + e.getId(), + e.getEventType(), + e.getDictionaryName(), + e.getAggregateId(), + e.getKafkaTopic(), + e.getAttemptCount(), + e.getLastError(), + e.getCreatedAt(), + e.getDlqAt()); + } + } +}