feat(infra): add Kafka Dead Letter Queue config to platform-infra (STA-218) (#193)

* feat(infra): add Kafka Dead Letter Queue config to platform-infra (STA-218)

Shared KafkaDlqConfig with DefaultErrorHandler + DeadLetterPublishingRecoverer.
Failed messages retry 3x with exponential backoff (1s, 2s, 4s) then route
to <topic>.dlt. Auto-activates via @ConditionalOnClass for all Kafka consumers.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix(infra): apply coding standards to KafkaDlqConfig (STA-218)

- Remove verbose Javadoc from config class
- Replace inline mock() with @ExtendWith(MockitoExtension.class) + @mock
- Remove @SuppressWarnings("unchecked") no longer needed

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* refactor(infra): simplify KafkaDlqConfig with ConditionalOnMissingBean (STA-218)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* test(infra): add behavioral backoff assertions to KafkaDlqConfigTest (STA-218)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: Puneethkumarck

platform-infra/src/main/java/com/stablecoin/payments/platform/infrastructure/messaging/KafkaDlqConfig.java

@@ -0,0 +1,45 @@
+package com.stablecoin.payments.platform.infrastructure.messaging;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.core.KafkaOperations;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.listener.CommonErrorHandler;
+import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
+import org.springframework.kafka.listener.DefaultErrorHandler;
+import org.springframework.kafka.support.ExponentialBackOffWithMaxRetries;
+
+@Slf4j
+@Configuration
+@ConditionalOnClass(KafkaTemplate.class)
+public class KafkaDlqConfig {
+
+    private static final int MAX_RETRIES = 3;
+    private static final long INITIAL_INTERVAL_MS = 1_000L;
+    private static final double MULTIPLIER = 2.0;
+    private static final long MAX_INTERVAL_MS = 10_000L;
+
+    @Bean
+    @ConditionalOnMissingBean(DeadLetterPublishingRecoverer.class)
+    public DeadLetterPublishingRecoverer deadLetterPublishingRecoverer(
+            KafkaOperations<String, Object> kafkaOperations) {
+        return new DeadLetterPublishingRecoverer(kafkaOperations);
+    }
+
+    @Bean
+    @ConditionalOnMissingBean(CommonErrorHandler.class)
+    public DefaultErrorHandler kafkaErrorHandler(DeadLetterPublishingRecoverer recoverer) {
+        var backOff = new ExponentialBackOffWithMaxRetries(MAX_RETRIES);
+        backOff.setInitialInterval(INITIAL_INTERVAL_MS);
+        backOff.setMultiplier(MULTIPLIER);
+        backOff.setMaxInterval(MAX_INTERVAL_MS);
+
+        log.info("Configured Kafka DLQ error handler: maxRetries={}, initialInterval={}ms, multiplier={}",
+                MAX_RETRIES, INITIAL_INTERVAL_MS, MULTIPLIER);
+
+        return new DefaultErrorHandler(recoverer, backOff);
+    }
+}

platform-infra/src/test/java/com/stablecoin/payments/platform/infrastructure/messaging/KafkaDlqConfigTest.java

@@ -0,0 +1,54 @@
+package com.stablecoin.payments.platform.infrastructure.messaging;
+
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.kafka.core.KafkaOperations;
+import org.springframework.kafka.support.ExponentialBackOffWithMaxRetries;
+import org.springframework.test.util.ReflectionTestUtils;
+import org.springframework.util.backoff.BackOff;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@ExtendWith(MockitoExtension.class)
+@DisplayName("KafkaDlqConfig")
+class KafkaDlqConfigTest {
+
+    private final KafkaDlqConfig config = new KafkaDlqConfig();
+
+    @Mock
+    private KafkaOperations<String, Object> kafkaOperations;
+
+    @Test
+    @DisplayName("should create DeadLetterPublishingRecoverer wired with KafkaOperations")
+    void shouldCreateDeadLetterPublishingRecoverer() {
+        // when
+        var recoverer = config.deadLetterPublishingRecoverer(kafkaOperations);
+
+        // then
+        assertThat(recoverer).isNotNull();
+    }
+
+    @Test
+    @DisplayName("should configure error handler with exponential backoff: 3 retries, 1s initial, 2x multiplier")
+    void shouldConfigureErrorHandlerWithExponentialBackoff() {
+        // given
+        var recoverer = config.deadLetterPublishingRecoverer(kafkaOperations);
+        var expectedBackOff = new ExponentialBackOffWithMaxRetries(3);
+        expectedBackOff.setInitialInterval(1_000L);
+        expectedBackOff.setMultiplier(2.0);
+        expectedBackOff.setMaxInterval(10_000L);
+
+        // when
+        var errorHandler = config.kafkaErrorHandler(recoverer);
+
+        // then
+        var failureTracker = ReflectionTestUtils.getField(errorHandler, "failureTracker");
+        var actualBackOff = (BackOff) ReflectionTestUtils.getField(failureTracker, "backOff");
+        assertThat(actualBackOff)
+                .usingRecursiveComparison()
+                .isEqualTo(expectedBackOff);
+    }
+}