Allow Retrying QueueAttributes Fetching Failures

[tomazfernandes]

Discussed in #1234

^{Originally posted by jtglass1 September 25, 2024}
Hi guys. I've a strange issue which is intermittent and I wanted to see if you had come across this before or might know what is causing it. I've two apps - [a] Admin - which runs in one JVM in ECS which allows the user to create queues dynamically via your framework [b] Reception - server which runs in a many JVM cluster also in ECS - it takes messages inbound from a REST API uses headers to derive the SQS queue the message will be stored on, and writes to that queue. 99% of the time this all works - and when a new queue is created by Admin, reception is able to use it immediately. However, sometimes, when reception tries to write to a new queue, it cannot find it. All other queues continue to work and we've never seen the look-up of a queue start to fail when it worked previously - the issue only happens when trying to write to a new queue for the first time. Only by restarting the reception server does the look-up start to work. It feels like there is some sort of cache here which is - for some reason - not getting refreshed reliably. I don't know if the 'can't find queue' is coming from the Spring layers or the AWS layers. It is also possible that there is an attribute I can use when creating the queue which will solve this issue - something like 'force cache reset'. I'm using v 3.1.1. I've put this in discussion as I honestly don't know where to start - AWS issue or Spring Cloud issue. Hoping you might have a view, thanks!

[LeeHyungGeol]

Hello @tomazfernandes.

Would it be okay if i give it a try on this issue?

[tomazfernandes]

Hey @LeeHyungGeol, sure, give it a go!

Let me know if you'd like to discuss the overall direction first.

[LeeHyungGeol]

Hi @tomazfernandes , thank you for the green light! I'd like to discuss the overall direction before implementing.

Problem Understanding

Based on my analysis of the issue and the discussion in #1234, the core problem is:

• Failed CompletableFuture<QueueAttributes> objects are cached indefinitely in queueAttributesCache
• When queue creation hasn't propagated to AWS SQS yet, the initial failure gets cached
• All subsequent requests return the cached failed future, requiring a service restart

Proposed Solution

I'm considering a three-pronged approach to make the cache more resilient:

1. Immediate Failure Removal (Priority 1)

private CompletableFuture<QueueAttributes> getQueueAttributes(String endpointName) {
    CompletableFuture<QueueAttributes> future = this.queueAttributesCache.computeIfAbsent(
        endpointName, newName -> doGetQueueAttributes(endpointName, newName)
    );
    
    // Remove failed futures from cache
    future.whenComplete((result, throwable) -> {
        if (throwable != null) {
            this.queueAttributesCache.remove(endpointName);
            logger.debug("Removed failed queue attributes from cache for: {}", endpointName);
        }
    });
    
    return future;
}

2. TTL-based Cache Expiration (Optional enhancement)

// Cache entry with timestamp
private static class CacheEntry {
    final CompletableFuture<QueueAttributes> future;
    final long createdAt;
    
    CacheEntry(CompletableFuture<QueueAttributes> future) {
        this.future = future;
        this.createdAt = System.currentTimeMillis();
    }
    
    boolean isExpired(long ttlMillis) {
        return System.currentTimeMillis() - createdAt > ttlMillis;
    }
}

private final Map<String, CacheEntry> queueAttributesCache = new ConcurrentHashMap<>();
private final long cacheTtl = Duration.ofMinutes(5).toMillis(); // Configurable

private CompletableFuture<QueueAttributes> getQueueAttributes(String endpointName) {
    CacheEntry cached = queueAttributesCache.get(endpointName);
    
    // Check if cache is valid
    if (cached != null && !cached.isExpired(cacheTtl)) {
        if (!cached.future.isCompletedExceptionally()) {
            return cached.future;
        }
        // Remove failed entries immediately
        queueAttributesCache.remove(endpointName);
    } else if (cached != null) {
        // Remove expired entry
        queueAttributesCache.remove(endpointName);
    }
    
    // Create new future
    CompletableFuture<QueueAttributes> future = doGetQueueAttributes(endpointName, endpointName);
    queueAttributesCache.put(endpointName, new CacheEntry(future));
    
    // Remove on failure
    future.whenComplete((result, throwable) -> {
        if (throwable != null) {
            queueAttributesCache.remove(endpointName);
        }
    });
    
    return future;
}

3. Rate Limiting for Failed Attempts (Optional enhancement)

private final Map<String, Long> lastFailureTime = new ConcurrentHashMap<>();
private final long failureRetryDelay = Duration.ofSeconds(1).toMillis(); // Configurable

private CompletableFuture<QueueAttributes> getQueueAttributes(String endpointName) {
    // Check if we should skip due to recent failure
    Long lastFailure = lastFailureTime.get(endpointName);
    if (lastFailure != null && 
        System.currentTimeMillis() - lastFailure < failureRetryDelay) {
        return CompletableFuture.failedFuture(
            new QueueAttributesResolvingException(
                "Queue attributes resolution recently failed for: " + endpointName)
        );
    }
    
    CacheEntry cached = queueAttributesCache.get(endpointName);
    // ... cache checking logic ...
    
    CompletableFuture<QueueAttributes> future = doGetQueueAttributes(endpointName, endpointName);
    
    future.whenComplete((result, throwable) -> {
        if (throwable != null) {
            queueAttributesCache.remove(endpointName);
            lastFailureTime.put(endpointName, System.currentTimeMillis());
        } else {
            lastFailureTime.remove(endpointName); // Clear on success
        }
    });
    
    return future;
}

Implementation Strategy

I see two possible approaches:

Option A: Minimal Change

• Just implement failure removal (Immediate Failure Removal (Priority 1))
• ~10 lines of code change
• Solves the immediate problem

Option B: Comprehensive Solution

• All three improvements with opt-in configuration

// In SqsTemplateOptions
SqsTemplateOptions cacheStrategy(CacheStrategy strategy);
SqsTemplateOptions cacheTtl(Duration ttl);
SqsTemplateOptions cacheRetryDelay(Duration delay);

public enum CacheStrategy {
    LEGACY,          // Current behavior
    REMOVE_ON_FAILURE,  // Option A
    SMART_CACHE      // Full solution with TTL + rate limiting
}

Questions

1. Which approach would you prefer for the initial PR?
2. Should this be configurable, or should we change the default behavior?
3. Any concerns about the performance impact of whenComplete on every cache access?
4. Would you prefer a phased approach (starting with Option A, then enhancing)?

I'm leaning toward Option A first (minimal change), then potentially following up with enhancements based on community feedback. What are your thoughts?

[tomazfernandes]

Hey @LeeHyungGeol, thanks for the detailed proposal.

I'm leaning toward Option A first (minimal change), then potentially following up with enhancements based on community feedback. What are your thoughts?

Agreed, let's make the minimal change, and gather feedback before iterating.

Also noting that this will mostly affect the use case mentioned in the original discussion of handling failures within SqsTemplate, since the SqsMessageListenerContainer will still fail at startup if it can't find the queue. Perhaps we can add some kind of retrial mechanism there in the future.

Thanks!

[LeeHyungGeol]

Hi @tomazfernandes,

Following our discussion, I've created the PR #1491 with Option A (minimal change) as agreed.

The implementation:

• Removes failed futures from cache using whenComplete
• ~5 lines of code change
• Includes tests to verify the fix

Ready for your review when you have time. Thanks!