LEGLINK-186: Restrict MeasureEval cache cleanup to terminal outcomes

With the retry ladder in place, cleanup-on-any-failure became a bug: a
transient failure deleted the {correlationId} cache entry in the finally
block, so the redelivered record read an empty cache and the empty-bundle
branch emitted a silent false not-reportable report.

The consumer now cleans up only when processing succeeds (keeping the
INITIAL+reportable entry for the SUPPLEMENTAL pass, as before). Failure-
path cleanup moves to the one place that knows the routing decision:
RetryTopicRecoverer gains an optional onDeadLetter hook, invoked solely
on terminal decisions (poison / exhausted / retries disabled) and only
after the dead letter is durably published. Hook failures are logged and
swallowed - the recovery already succeeded, and propagating would
redeliver the record into a duplicate dead letter.

MeasureEval wires the hook through a new ResourceCacheCleanup component
(Redis + optional ABS, best-effort, never throws) shared by the success
path and both recoverers; a record whose value never deserialized is
skipped and left to the cache expiration policy. Validation keeps the
existing factory overload unchanged.

Verified live in docker-compose across seven scenarios: poison, Mongo/
Redis/ABS outages run to exhaustion (cache kept through every retry,
cleaned or gracefully leaked at the dead letter), and Mongo/Redis/ABS
restored mid-ladder (redelivery completes a real NHSN evaluation and
cleans up on success). Known follow-up: the Azure SDK's internal retry
policy makes each ABS call block the single-threaded consumer executor
for minutes during an ABS outage; tighten RequestRetryOptions so the
retry-topic ladder is the single source of retry truth.

Claude-Session: https://claude.ai/code/session_01KN565tFAuJUAEkK2DdRF1e

Author: arianamihailescu

Java/measureeval/src/main/java/com/lantanagroup/link/measureeval/configs/KafkaConfig.java

@@ -4,6 +4,9 @@
 import com.fasterxml.jackson.databind.JavaType;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.lantanagroup.link.measureeval.records.*;
+import com.lantanagroup.link.measureeval.services.AbsResourceService;
+import com.lantanagroup.link.measureeval.services.RedisResourceService;
+import com.lantanagroup.link.measureeval.services.ResourceCacheCleanup;
 import com.lantanagroup.link.shared.kafka.RetryTopicRecoverer;
 import com.lantanagroup.link.shared.kafka.RetryTopicRecovererFactory;
 import com.lantanagroup.link.shared.config.KafkaRetryConfig;
@@ -285,35 +288,57 @@ private RetryTopicRecoverer createRetryTopicRecoverer(
             KafkaTemplate<?, ?> kafkaTemplate,
             String retryTopic,
             String errorTopic,
-            KafkaRetryConfig retryConfig) {
-        return RetryTopicRecovererFactory.create(kafkaTemplate, retryTopic, errorTopic, retryConfig, NON_RETRYABLE);
+            KafkaRetryConfig retryConfig,
+            ResourceCacheCleanup cacheCleanup) {
+        // The consumer no longer cleans up the resource cache on failure — a record routed to -Retry
+        // is redelivered and still needs its cached resources. This hook is the failure-path cleanup:
+        // it runs only on terminal decisions (poison / exhausted / retries disabled), after the dead
+        // letter is durably published. A record whose value never deserialized is skipped; the cache
+        // expiration policy reclaims whatever it left behind.
+        return RetryTopicRecovererFactory.create(kafkaTemplate, retryTopic, errorTopic, retryConfig, NON_RETRYABLE,
+                (record, exception) -> {
+                    if (record.value() instanceof AbstractResourceRecord resourceRecord) {
+                        cacheCleanup.cleanup(resourceRecord.getCacheKey(), resourceRecord.getCacheType());
+                    }
+                });
+    }
+
+    @Bean
+    public ResourceCacheCleanup resourceCacheCleanup(
+            RedisResourceService redisResourceService,
+            ObjectProvider<AbsResourceService> absResourceService) {
+        return new ResourceCacheCleanup(redisResourceService, absResourceService.getIfAvailable());
     }
 
     @Bean
     public ConsumerRecordRecoverer resourceNormalizedRecoverer(
             @Qualifier("compressedKafkaTemplate")
             KafkaTemplate<String, ResourcesNormalized> kafkaTemplate,
-            KafkaRetryConfig retryConfig) {
+            KafkaRetryConfig retryConfig,
+            ResourceCacheCleanup resourceCacheCleanup) {
 
         return createRetryTopicRecoverer(
                 kafkaTemplate,
                 "ResourcesNormalized-Retry",
                 "ResourcesNormalized-Error",
-                retryConfig
+                retryConfig,
+                resourceCacheCleanup
         );
     }
 
     @Bean
     public ConsumerRecordRecoverer evaluationRequestedRecoverer(
             @Qualifier("compressedKafkaTemplate")
             KafkaTemplate<String, EvaluationRequested> kafkaTemplate,
-            KafkaRetryConfig retryConfig) {
+            KafkaRetryConfig retryConfig,
+            ResourceCacheCleanup resourceCacheCleanup) {
 
         return createRetryTopicRecoverer(
                 kafkaTemplate,
                 "EvaluationRequested-Retry",
                 "EvaluationRequested-Error",
-                retryConfig
+                retryConfig,
+                resourceCacheCleanup
         );
     }
 

Java/measureeval/src/main/java/com/lantanagroup/link/measureeval/services/AbstractResourceConsumer.java

@@ -13,7 +13,6 @@
 import com.lantanagroup.link.shared.kafka.Topics;
 import com.lantanagroup.link.shared.kafka.records.ResourceKey;
 import com.lantanagroup.link.shared.utils.DiagnosticNames;
-import com.lantanagroup.link.shared.utils.LogUtils;
 import io.opentelemetry.api.common.Attributes;
 import io.opentelemetry.api.trace.Span;
 import org.apache.commons.collections4.map.PassiveExpiringMap;
@@ -54,6 +53,7 @@ public abstract class AbstractResourceConsumer<T extends AbstractResourceRecord>
     private final MeasureReportGeneratedProducer measureReportGeneratedProducer;
     private final RedisResourceService redisResourceService;
     private final AbsResourceService absResourceService;
+    private final ResourceCacheCleanup cacheCleanup;
     private final MongoOperations mongoOperations;
 
     public AbstractResourceConsumer (
@@ -80,6 +80,7 @@ public AbstractResourceConsumer (
         this.blobStorageService = blobStorageService;
         this.redisResourceService = redisResourceService;
         this.absResourceService = absResourceService;
+        this.cacheCleanup = new ResourceCacheCleanup(redisResourceService, absResourceService);
         this.mongoOperations = mongoOperations;
     }
 
@@ -119,9 +120,6 @@ protected void process(ConsumerRecord<ResourceKey, T> record) {
             if (value.getCacheType() == null) {
                 throw new ValidationException("Cache Type is null.");
             }
-            // Captured here, next to its validation, rather than after the metrics/logging below: the
-            // finally-block cleanup is guarded on both correlationId and cacheType, so anything that
-            // throws between the two assignments would silently strand a valid cache entry.
             cacheType = value.getCacheType();
             correlationId = value.getCacheKey();
             if (correlationId == null || correlationId.isEmpty()) {
@@ -211,7 +209,7 @@ protected void process(ConsumerRecord<ResourceKey, T> record) {
                     value.getQueryType() == QueryType.INITIAL && reportablePatient;
 
             // INITIAL + reportable keeps the cache for the SUPPLEMENTAL pass to reuse; every other
-            // outcome, including failure, is cleaned up by the finally block below.
+            // successful outcome cleans it up below.
             keepCacheForSupplemental = initialReportable;
 
             if (initialReportable) {
@@ -225,39 +223,28 @@ protected void process(ConsumerRecord<ResourceKey, T> record) {
                         resources.size(), correlationId);
             }
 
-        } finally {
-            // A task may still be running if we got here by way of an exception; stop it so that its
-            // elapsed time is reported and so that starting the cleanup task below cannot throw.
-            if (perf && taskStopWatch.isRunning()) {
-                taskStopWatch.stop();
-            }
-
-            // Clean up cache after SUPPLEMENTAL, after INITIAL if patient is not reportable, and after
-            // any failure. INITIAL + reportable keeps the cache for the SUPPLEMENTAL pass to reuse.
+            // Cleanup runs ONLY when processing succeeded. A thrown exception must leave the cache
+            // intact: the recoverer may route the record to -Retry, and the redelivered record needs
+            // its cached resources. Terminal (dead-letter) cleanup is the recoverer's job — it is the
+            // only place the routing decision is known (see the terminal-failure hook in KafkaConfig).
             if (keepCacheForSupplemental) {
                 logger.debug("Keeping cache for SUPPLEMENTAL pass, correlationId={}", correlationId);
-            } else if (correlationId != null && cacheType != null) {
+            } else {
                 if (perf) taskStopWatch.start("cleanupCache");
                 try {
-                    switch (cacheType) {
-                        case REDIS -> redisResourceService.cleanup(correlationId);
-                        case ABS -> {
-                            if (absResourceService != null) {
-                                absResourceService.cleanup(correlationId);
-                            }
-                        }
-                    }
-                    logger.debug("Cache cleanup complete for correlationId={}, cacheType={}",
-                            LogUtils.sanitize(correlationId), LogUtils.sanitize(cacheType));
-                } catch (Exception e) {
-                    // Never mask the exception that brought us here, if any.
-                    logger.error("Cache cleanup failed for correlationId={}, cacheType={}",
-                            LogUtils.sanitize(correlationId), LogUtils.sanitize(cacheType), e);
+                    cacheCleanup.cleanup(correlationId, cacheType);
                 } finally {
                     if (perf) taskStopWatch.stop();
                 }
             }
 
+        } finally {
+            // A task may still be running if we got here by way of an exception; stop it so that its
+            // elapsed time is reported.
+            if (perf && taskStopWatch.isRunning()) {
+                taskStopWatch.stop();
+            }
+
             if (perf) {
                 totalStopWatch.stop();
                 for (StopWatch.TaskInfo task : taskStopWatch.getTaskInfo()) {

Java/measureeval/src/main/java/com/lantanagroup/link/measureeval/services/ResourceCacheCleanup.java

@@ -0,0 +1,49 @@
+package com.lantanagroup.link.measureeval.services;
+
+import com.lantanagroup.link.measureeval.entities.CacheType;
+import com.lantanagroup.link.shared.utils.LogUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Releases the resource cache entry for a correlation id. Shared by the two places allowed to
+ * delete it: the consumer's success path (the record was fully evaluated) and the recoverer's
+ * terminal-failure hook (the record was durably dead-lettered and will never be redelivered).
+ *
+ * <p>Never call this on a failure that may still be retried — the redelivered record needs its
+ * cached resources. Best-effort: failures are logged, never thrown, so cleanup can never turn a
+ * handled record back into a failure; the cache expiration policy is the backstop.</p>
+ */
+public class ResourceCacheCleanup {
+
+    private static final Logger logger = LoggerFactory.getLogger(ResourceCacheCleanup.class);
+
+    private final RedisResourceService redisResourceService;
+    private final AbsResourceService absResourceService;
+
+    public ResourceCacheCleanup(RedisResourceService redisResourceService, AbsResourceService absResourceService) {
+        this.redisResourceService = redisResourceService;
+        this.absResourceService = absResourceService;
+    }
+
+    public void cleanup(String correlationId, CacheType cacheType) {
+        if (correlationId == null || correlationId.isEmpty() || cacheType == null) {
+            return;
+        }
+        try {
+            switch (cacheType) {
+                case REDIS -> redisResourceService.cleanup(correlationId);
+                case ABS -> {
+                    if (absResourceService != null) {
+                        absResourceService.cleanup(correlationId);
+                    }
+                }
+            }
+            logger.debug("Cache cleanup complete for correlationId={}, cacheType={}",
+                    LogUtils.sanitize(correlationId), LogUtils.sanitize(cacheType));
+        } catch (Exception e) {
+            logger.error("Cache cleanup failed for correlationId={}, cacheType={}",
+                    LogUtils.sanitize(correlationId), LogUtils.sanitize(cacheType), e);
+        }
+    }
+}

Java/measureeval/src/test/java/com/lantanagroup/link/measureeval/configs/RetryTopicRecovererDecisionTest.java

@@ -238,4 +238,119 @@ void deadLetteredRecord_stillRecordsHowManyAttemptsItGot() {
         assertEquals(1, ByteBuffer.wrap(attempts.value()).getInt(),
                 "poison failed once and was never retried, so it must record exactly one attempt");
     }
+
+    // ---- Terminal-failure hook. A record routed to -Error is finished — it will never be redelivered —
+    //      so this is the one moment its resources (e.g. the cached FHIR bundle) may be released. The
+    //      hook must fire exactly on dead-letter decisions, only after the dead letter is durably
+    //      published, and must never turn a completed recovery back into a failure.
+
+    /** Drives the record through the 6-arg factory wiring, recording publish/hook ordering. */
+    @SuppressWarnings("unchecked")
+    private void runWithHook(KafkaRetryConfig config, ConsumerRecord<String, String> record,
+                             Exception exception,
+                             java.util.function.BiConsumer<ConsumerRecord<?, ?>, Exception> onDeadLetter,
+                             List<String> events) {
+        KafkaTemplate<Object, Object> template = mock(KafkaTemplate.class);
+        when(template.partitionsFor(anyString())).thenAnswer(invocation ->
+                List.of(new PartitionInfo(invocation.getArgument(0), 0, null, new Node[0], new Node[0])));
+        when(template.send(any(ProducerRecord.class))).thenAnswer(invocation -> {
+            events.add("published");
+            return CompletableFuture.completedFuture(null);
+        });
+
+        RetryTopicRecoverer recoverer = RetryTopicRecovererFactory.create(
+                template, RETRY_TOPIC, ERROR_TOPIC, config, KafkaConfig.NON_RETRYABLE, onDeadLetter);
+        recoverer.accept(record, exception);
+    }
+
+    @Test
+    void terminalHook_firesForPoison_afterTheDeadLetterIsPublished() {
+        List<String> events = new java.util.ArrayList<>();
+
+        runWithHook(retryConfig(3, false), record(), new ValidationException("Query Type is null."),
+                (r, e) -> events.add("hook"), events);
+
+        assertEquals(List.of("published", "hook"), events,
+                "the hook releases resources the record still needs until it is durably dead-lettered, "
+                        + "so it must run exactly once, after the publish");
+    }
+
+    @Test
+    void terminalHook_firesForExhaustedAttempts() {
+        ConsumerRecord<String, String> record = record();
+        stampAttempts(record, 2);
+        List<String> events = new java.util.ArrayList<>();
+
+        runWithHook(retryConfig(3, false), record, new RuntimeException("still failing"),
+                (r, e) -> events.add("hook"), events);
+
+        assertEquals(List.of("published", "hook"), events);
+    }
+
+    @Test
+    void terminalHook_doesNotFireForARetriedRecord() {
+        List<String> events = new java.util.ArrayList<>();
+
+        runWithHook(retryConfig(3, false), record(), new RuntimeException("transient boom"),
+                (r, e) -> events.add("hook"), events);
+
+        assertEquals(List.of("published"), events,
+                "a record routed to -Retry will be redelivered and still needs its resources — "
+                        + "the hook must not fire");
+    }
+
+    /** Drives a record through the REAL KafkaConfig recoverer bean, so the wiring itself is pinned. */
+    @SuppressWarnings("unchecked")
+    private void runThroughKafkaConfig(ConsumerRecord<String, Object> record, Exception exception,
+                                       com.lantanagroup.link.measureeval.services.RedisResourceService redis) {
+        KafkaTemplate<String, com.lantanagroup.link.measureeval.records.ResourcesNormalized> template =
+                mock(KafkaTemplate.class);
+        when(template.partitionsFor(anyString())).thenAnswer(invocation ->
+                List.of(new PartitionInfo(invocation.getArgument(0), 0, null, new Node[0], new Node[0])));
+        when(template.send(any(ProducerRecord.class)))
+                .thenReturn(CompletableFuture.completedFuture(null));
+
+        var recoverer = new KafkaConfig().resourceNormalizedRecoverer(
+                template, retryConfig(3, false),
+                new com.lantanagroup.link.measureeval.services.ResourceCacheCleanup(redis, null));
+        recoverer.accept(record, exception);
+    }
+
+    private static ConsumerRecord<String, Object> resourceRecord(String cacheKey) {
+        var value = new com.lantanagroup.link.measureeval.records.ResourcesNormalized();
+        value.setCacheKey(cacheKey);
+        value.setCacheType(com.lantanagroup.link.measureeval.entities.CacheType.REDIS);
+        return new ConsumerRecord<>("ResourcesNormalized", 0, 0L, "key", value);
+    }
+
+    @Test
+    void kafkaConfigHook_cleansTheResourceCache_whenARecordIsDeadLettered() {
+        var redis = mock(com.lantanagroup.link.measureeval.services.RedisResourceService.class);
+
+        runThroughKafkaConfig(resourceRecord("corr-1"), new ValidationException("poison"), redis);
+
+        verify(redis).cleanup("corr-1");
+    }
+
+    @Test
+    void kafkaConfigHook_leavesTheResourceCache_whenARecordIsRetried() {
+        var redis = mock(com.lantanagroup.link.measureeval.services.RedisResourceService.class);
+
+        runThroughKafkaConfig(resourceRecord("corr-1"), new RuntimeException("transient boom"), redis);
+
+        verify(redis, never()).cleanup(anyString());
+    }
+
+    @Test
+    void terminalHook_throwing_doesNotFailTheRecovery() {
+        List<String> events = new java.util.ArrayList<>();
+
+        // The dead letter is already published when the hook runs; a hook failure must not propagate,
+        // or the offset stays uncommitted and the record is redelivered into a duplicate dead letter.
+        assertDoesNotThrow(() -> runWithHook(
+                retryConfig(3, false), record(), new ValidationException("Query Type is null."),
+                (r, e) -> { throw new RuntimeException("hook failed"); }, events));
+
+        assertEquals(List.of("published"), events);
+    }
 }