LEGLINK-186: Address code review findings on the retry work

- Tolerate legacy and malformed attempts headers (the one real bug):
  currentAttempts() threw BufferUnderflowException on anything but a
  4-byte value; a record carrying spring-kafka's legacy 1-byte encoding
  (mixed-version rollouts, old in-flight retry records) escaped the
  recoverer, skipped the ack, and stalled the partition. Now decodes
  both supported encodings and reads unsupported lengths as 0 - worst
  case one extra ladder pass instead of a wedged partition. Tests for
  both cases, watched red first.

- Pin the awaited dead-letter publish: spring-kafka 3.1.4's
  DeadLetterPublishingRecoverer defaults failIfSendResultIsError=true
  (verified in the library bytecode), so accept() already awaits the
  bounded send result and a failed publish propagates - keeping the
  source offset uncommitted and the onDeadLetter hook unfired. No code
  change needed; a test now pins the invariant against future
  spring-kafka upgrades, with a comment at the delegate.

- Correct the async-ack comments in AbstractAsyncConsumer: under
  MANUAL_IMMEDIATE + asyncAcks an executor-thread acknowledge() is
  queued for in-order commit, and a missing ack pauses the partition
  until restart/rebalance rather than triggering live redelivery.
  Comments only; behavior unchanged; nack() noted as unsupported.

- Branch coverage: the KafkaConfig dead-letter hook's non-resource
  value path (skip cleanup, no throw) and ResourceCacheCleanup's
  ABS-with-null-service guard, which no longer had indirect coverage
  once cleanup became success-only.

- Sanitize logging arguments (LogUtils.sanitize) in
  CacheBlobStorageConfig, AbsResourceService, AbstractResourceConsumer,
  and RedisResourceService.

- Record shipped defaults in app-config.yaml: PT3S,PT3S and false for
  the spring.kafka.retry keys in both service sections.

151 measureeval-side tests green; validation compiles against the
shared changes.

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

Author: arianamihailescu

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

@@ -5,6 +5,7 @@
 import com.azure.storage.common.policy.RequestRetryOptions;
 import com.azure.storage.common.policy.RetryPolicyType;
 import com.lantanagroup.link.measureeval.services.AbsResourceService;
+import com.lantanagroup.link.shared.utils.LogUtils;
 import lombok.Getter;
 import lombok.Setter;
 import org.apache.commons.lang3.StringUtils;
@@ -47,7 +48,8 @@ public AbsResourceService absResourceService() {
             return null;
         }
         logger.info("Creating AbsResourceService: container={}, blobRoot={}, maxTries={}, tryTimeout={}s",
-                blobContainerName, blobRoot, maxTries, tryTimeoutSeconds);
+                LogUtils.sanitize(blobContainerName), LogUtils.sanitize(blobRoot),
+                LogUtils.sanitize(maxTries), LogUtils.sanitize(tryTimeoutSeconds));
         BlobContainerClient client = new BlobServiceClientBuilder()
                 .connectionString(connectionString)
                 .retryOptions(buildRetryOptions())

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

@@ -7,6 +7,7 @@
 import com.azure.storage.blob.models.BlobStorageException;
 import com.azure.storage.blob.models.ListBlobsOptions;
 import com.lantanagroup.link.measureeval.entities.Resource;
+import com.lantanagroup.link.shared.utils.LogUtils;
 import org.hl7.fhir.r4.model.ResourceType;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -74,11 +75,12 @@ private List<Resource> readBlobResources(String blobName, String facilityId, Str
             // silently producing a not-reportable report and acking. Connectivity/timeout failures are
             // not BlobStorageException, so they are not caught here and propagate for the same reason.
             if (BlobErrorCode.BLOB_NOT_FOUND.equals(e.getErrorCode())) {
-                logger.debug("Blob '{}' not found for correlationId='{}'; treating as empty cache", blobName, correlationId);
+                logger.debug("Blob '{}' not found for correlationId='{}'; treating as empty cache",
+                        LogUtils.sanitize(blobName), LogUtils.sanitize(correlationId));
                 return resources;
             }
             logger.error("Failed to download blob '{}' for correlationId='{}' (errorCode={}); propagating for retry.",
-                    blobName, correlationId, e.getErrorCode());
+                    LogUtils.sanitize(blobName), LogUtils.sanitize(correlationId), LogUtils.sanitize(e.getErrorCode()));
             throw e;
         }
 

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

@@ -13,6 +13,7 @@
 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;
@@ -228,7 +229,7 @@ protected void process(ConsumerRecord<ResourceKey, T> record) {
             // 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);
+                logger.debug("Keeping cache for SUPPLEMENTAL pass, correlationId={}", LogUtils.sanitize(correlationId));
             } else {
                 if (perf) taskStopWatch.start("cleanupCache");
                 try {

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

@@ -2,6 +2,7 @@
 
 import com.lantanagroup.link.measureeval.entities.Resource;
 import com.lantanagroup.link.measureeval.exceptions.ResourceCacheUnavailableException;
+import com.lantanagroup.link.shared.utils.LogUtils;
 import org.hl7.fhir.r4.model.ResourceType;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -47,7 +48,7 @@ public List<Resource> readResources(String facilityId, String correlationId, Str
 
         if (fields.isEmpty()) {
             // Reached only when the cache was reachable: the key genuinely has no fields.
-            logger.debug("No Redis entries for correlationId='{}' (cache reachable, key absent)", correlationId);
+            logger.debug("No Redis entries for correlationId='{}' (cache reachable, key absent)", LogUtils.sanitize(correlationId));
             return resources;
         }
 

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

@@ -332,6 +332,21 @@ void kafkaConfigHook_cleansTheResourceCache_whenARecordIsDeadLettered() {
         verify(redis).cleanup("corr-1");
     }
 
+    @Test
+    void kafkaConfigHook_skipsCleanup_whenADeadLetteredValueIsNotAResourceRecord() {
+        // The other side of the hook's instanceof guard: a terminal record whose value never
+        // deserialized into a resource record (e.g. container-thread poison republished as raw
+        // content) has no cache key to read - the hook must skip, not throw, and leave whatever
+        // the record left behind to the cache expiration policy.
+        var redis = mock(com.lantanagroup.link.measureeval.services.RedisResourceService.class);
+
+        assertDoesNotThrow(() -> runThroughKafkaConfig(
+                new ConsumerRecord<>("ResourcesNormalized", 0, 0L, "key", "raw non-record value"),
+                new ValidationException("poison"), redis));
+
+        verifyNoInteractions(redis);
+    }
+
     @Test
     void kafkaConfigHook_leavesTheResourceCache_whenARecordIsRetried() {
         var redis = mock(com.lantanagroup.link.measureeval.services.RedisResourceService.class);
@@ -341,6 +356,30 @@ void kafkaConfigHook_leavesTheResourceCache_whenARecordIsRetried() {
         verify(redis, never()).cleanup(anyString());
     }
 
+    @Test
+    void failedDeadLetterPublish_propagates_andTheHookDoesNotFire() {
+        // The recoverer must await the send result. If accept() returned after merely enqueueing the
+        // send, a broker failure would be invisible: the async consumer would ack the source record
+        // (losing it - neither retried nor dead-lettered) and the hook would release resources for a
+        // dead letter that never landed.
+        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)))
+                .thenReturn(CompletableFuture.failedFuture(new RuntimeException("broker unavailable")));
+
+        List<String> events = new java.util.ArrayList<>();
+        RetryTopicRecoverer recoverer = RetryTopicRecovererFactory.create(
+                template, RETRY_TOPIC, ERROR_TOPIC, retryConfig(3, false), KafkaConfig.NON_RETRYABLE,
+                (r, e) -> events.add("hook"));
+
+        assertThrows(Exception.class,
+                () -> recoverer.accept(record(), new ValidationException("poison")),
+                "a failed dead-letter publish must propagate so the source offset stays uncommitted");
+        assertEquals(List.of(), events,
+                "resources must not be released for a dead letter that was never published");
+    }
+
     @Test
     void terminalHook_throwing_doesNotFailTheRecovery() {
         List<String> events = new java.util.ArrayList<>();

Java/measureeval/src/test/java/com/lantanagroup/link/measureeval/services/ResourceCacheCleanupTest.java

@@ -0,0 +1,37 @@
+package com.lantanagroup.link.measureeval.services;
+
+import com.lantanagroup.link.measureeval.entities.CacheType;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+
+class ResourceCacheCleanupTest {
+
+    @Test
+    void cleanup_absCacheTypeWithoutConfiguredService_doesNothing() {
+        // The ABS branch's null-guard: cache-blob-storage may be unconfigured while a record still
+        // arrives claiming CacheType.ABS. Cleanup must no-op (not NPE), and must not fall through
+        // to Redis - the entry is left to the cache expiration policy.
+        RedisResourceService redis = mock(RedisResourceService.class);
+        ResourceCacheCleanup cleanup = new ResourceCacheCleanup(redis, null);
+
+        assertDoesNotThrow(() -> cleanup.cleanup("corr-1", CacheType.ABS));
+
+        verifyNoInteractions(redis);
+    }
+
+    @Test
+    void cleanup_absCacheTypeWithConfiguredService_cleansAbsOnly() {
+        RedisResourceService redis = mock(RedisResourceService.class);
+        AbsResourceService abs = mock(AbsResourceService.class);
+        ResourceCacheCleanup cleanup = new ResourceCacheCleanup(redis, abs);
+
+        cleanup.cleanup("corr-1", CacheType.ABS);
+
+        verify(abs).cleanup("corr-1");
+        verifyNoInteractions(redis);
+    }
+}

Java/measureeval/src/test/java/com/lantanagroup/link/measureeval/services/RetryTopicRecovererTest.java

@@ -50,6 +50,31 @@ private static long decodeAsSpring(byte[] value) {
         return new BigInteger(value).longValue();
     }
 
+    private static RecordHeaders attemptsHeader(byte[] value) {
+        RecordHeaders headers = new RecordHeaders();
+        headers.add(RetryTopicHeaders.DEFAULT_HEADER_ATTEMPTS, value);
+        return headers;
+    }
+
+    // ---- Attempts-header tolerance. spring-kafka accepts both the legacy one-byte and the current
+    //      four-byte encoding, and records can arrive carrying either (mixed-version rollouts, old
+    //      in-flight retry records). A decode failure here would propagate out of the recoverer and
+    //      stall the partition (the async consumer skips the ack), so unsupported lengths must read
+    //      as 0 - "no attempts yet" - rather than throw.
+
+    @Test
+    void currentAttempts_decodesLegacySingleByteHeader() {
+        assertEquals(3, RetryTopicRecoverer.currentAttempts(attemptsHeader(new byte[]{3})));
+    }
+
+    @Test
+    void currentAttempts_returnsZeroForUnsupportedLengths() {
+        assertEquals(0, RetryTopicRecoverer.currentAttempts(attemptsHeader(new byte[]{0, 1})),
+                "a two-byte value is neither encoding; it must read as zero, not throw");
+        assertEquals(0, RetryTopicRecoverer.currentAttempts(attemptsHeader(new byte[]{})),
+                "an empty value must read as zero, not throw");
+    }
+
     @Test
     void backoffTimestamp_decodesAsFutureDueTime_viaSpringReader() {
         RetryTopicRecoverer recoverer = newRecoverer(mock(DeadLetterPublishingRecoverer.class));

Java/shared/src/main/java/com/lantanagroup/link/shared/kafka/AbstractAsyncConsumer.java

@@ -47,18 +47,23 @@ protected void doConsume(ConsumerRecord<K, T> record, Acknowledgment ack) {
                                 record.topic(), record.partition(), record.offset(), processError);
                     } else {
                         // Route to -Retry/-Error. If this throws (e.g. the broker is unavailable), let it
-                        // propagate so the ack below is skipped — an uncommitted offset makes Kafka
-                        // redeliver the record instead of silently losing a failure that was neither
-                        // retried nor dead-lettered.
+                        // propagate so the ack below is skipped. Under MANUAL_IMMEDIATE + asyncAcks the
+                        // missing acknowledgment stalls this partition (no further records are committed
+                        // past the gap) rather than silently losing a failure that was neither retried nor
+                        // dead-lettered; the record is redelivered after a restart or rebalance.
                         recoverer.accept(record, processError);
                     }
                 }
                 // Reached only when process() succeeded, recovery published successfully, or we
-                // intentionally dropped (no recoverer): the record is accounted for, so commit the offset.
+                // intentionally dropped (no recoverer): the record is accounted for. With asyncAcks the
+                // acknowledgment is queued from this executor thread and the container commits it in
+                // offset order, not immediately. (Do not use nack() here — unsupported with asyncAcks.)
                 ack.acknowledge();
             } catch (Exception unrecovered) {
-                // process() failed AND recovery (or the ack itself) failed. Do NOT commit the offset:
-                // leaving it uncommitted preserves at-least-once delivery so the record is redelivered.
+                // process() failed AND recovery (or the ack itself) failed. Do NOT acknowledge: the
+                // missing ack pauses this partition, preserving the record (at-least-once, no loss) at
+                // the cost of a stalled partition until a restart or rebalance redelivers it — this log
+                // line is the operational signal for that condition.
                 logger.error("Failed to recover or acknowledge record from topic={} partition={} offset={}; leaving offset uncommitted for redelivery",
                         record.topic(), record.partition(), record.offset(), unrecovered);
             } finally {

Java/shared/src/main/java/com/lantanagroup/link/shared/kafka/RetryTopicRecoverer.java

@@ -146,10 +146,24 @@ public void accept(ConsumerRecord<?, ?> record, Exception exception) {
     /**
      * Retry attempts recorded in the attempts header, or 0 if absent. Shared with KafkaConfig's
      * destination resolver so routing and stamping use the same count.
+     *
+     * <p>Tolerates both encodings spring-kafka accepts — the current four-byte int and the legacy
+     * single byte — since records can arrive carrying either (mixed-version rollouts, old in-flight
+     * retry records). Any other length reads as 0 ("no attempts yet") rather than throwing: a decode
+     * failure would propagate out of {@link #accept} and stall the partition, and the worst case of
+     * reading 0 is one extra pass through the retry ladder before dead-lettering.</p>
      */
     public static int currentAttempts(Headers headers) {
         var header = headers.lastHeader(RetryTopicHeaders.DEFAULT_HEADER_ATTEMPTS);
-        if (header == null) return 0;
-        return ByteBuffer.wrap(header.value()).getInt();
+        if (header == null || header.value() == null) return 0;
+        byte[] value = header.value();
+        if (value.length == Integer.BYTES) {
+            return ByteBuffer.wrap(value).getInt();
+        }
+        if (value.length == 1) {
+            return value[0];
+        }
+        logger.warn("Unsupported attempts-header length {}; treating as 0 attempts.", value.length);
+        return 0;
     }
 }

Java/shared/src/main/java/com/lantanagroup/link/shared/kafka/RetryTopicRecovererFactory.java

@@ -77,6 +77,10 @@ public static RetryTopicRecoverer create(
                     return new TopicPartition(target, record.partition());
                 };
 
+        // Relies on spring-kafka's failIfSendResultIsError=true default (since 2.7): accept() awaits
+        // the send result (bounded by the producer's delivery.timeout.ms + buffer) and throws on
+        // failure, so the source offset stays uncommitted and the onDeadLetter hook only runs after a
+        // durable publish. Pinned by RetryTopicRecovererDecisionTest.failedDeadLetterPublish_*.
         DeadLetterPublishingRecoverer delegate = new DeadLetterPublishingRecoverer(kafkaTemplate, resolver);
 
         return new RetryTopicRecoverer(