Save changes. incomplete state with backoff timer changes not yet done because of the mixup in the Lru cache code logic, with expiry time.

Author: kannanjgithub

rls/src/main/java/io/grpc/rls/CachingRlsLbClient.java

@@ -363,13 +363,13 @@ final CachedRouteLookupResponse get(final RouteLookupRequest request) {
       final CacheEntry cacheEntry;
       cacheEntry = linkedHashLruCache.read(request);
       if (cacheEntry == null
-          || cacheEntry instanceof BackoffCacheEntry
-          && (((BackoffCacheEntry) cacheEntry).isBackoffTimeEnded)) {
+          || cacheEntry instanceof BackoffCacheEntry && cacheEntry.isExpired(ticker.read())) {
         PendingCacheEntry pendingEntry = pendingCallCache.get(request);
         if (pendingEntry != null) {
           return CachedRouteLookupResponse.pendingResponse(pendingEntry);
         }
-        return asyncRlsCall(request, /* backoffPolicy= */ null);
+        return asyncRlsCall(request, cacheEntry instanceof BackoffCacheEntry
+            ? ((BackoffCacheEntry) cacheEntry).backoffPolicy : null);
       }
 
       if (cacheEntry instanceof DataCacheEntry) {
@@ -463,7 +463,9 @@ private BackoffCacheEntry createBackOffEntry(
         request, status, delayNanos);
     BackoffCacheEntry entry = new BackoffCacheEntry(request, status, backoffPolicy);
     // Lock is held, so the task can't execute before the assignment
-    entry.scheduledFuture = scheduledExecutorService.schedule(
+    entry.backoffTimer = scheduledExecutorService.schedule(
+        () -> refreshBackoffEntry(entry), delayNanos, TimeUnit.NANOSECONDS);
+    entry.backoffTimer = scheduledExecutorService.schedule(
         () -> refreshBackoffEntry(entry), delayNanos, TimeUnit.NANOSECONDS);
     linkedHashLruCache.cacheAndClean(request, entry);
     return entry;
@@ -472,13 +474,12 @@ private BackoffCacheEntry createBackOffEntry(
   private void refreshBackoffEntry(BackoffCacheEntry entry) {
     synchronized (lock) {
       // This checks whether the task has been cancelled and prevents a second execution.
-      if (!entry.scheduledFuture.cancel(false)) {
+      if (!entry.backoffTimer.cancel(false)) {
         // Future was previously cancelled
         return;
       }
       logger.log(ChannelLogLevel.DEBUG,
           "[RLS Entry {0}] Calling RLS for transition to pending", entry.request);
-      entry.isBackoffTimeEnded = true;
       // Cache updated. updateBalancingState() to reattempt picks
       helper.triggerPendingRpcProcessing();
     }
@@ -785,8 +786,7 @@ private static final class BackoffCacheEntry extends CacheEntry {
 
     private final Status status;
     private final BackoffPolicy backoffPolicy;
-    private Future<?> scheduledFuture;
-    private boolean isBackoffTimeEnded;
+    private Future<?> backoffTimer;
 
     BackoffCacheEntry(RouteLookupRequest request, Status status, BackoffPolicy backoffPolicy) {
       super(request);
@@ -805,12 +805,12 @@ int getSizeBytes() {
 
     @Override
     boolean isExpired(long now) {
-      return scheduledFuture.isDone();
+      return backoffTimer.isDone();
     }
 
     @Override
     void cleanup() {
-      scheduledFuture.cancel(false);
+      backoffTimer.cancel(false);
     }
 
     @Override

rls/src/test/java/io/grpc/rls/CachingRlsLbClientTest.java

@@ -360,12 +360,14 @@ public void get_throttled_backoffBehavior() throws Exception {
     CachedRouteLookupResponse resp = getInSyncContext(routeLookupRequest);
     assertThat(resp.hasError()).isTrue();
 
-    // let it be throttled again
     fakeClock.forwardTime(10, TimeUnit.MILLISECONDS);
+    // Assert that the same backoff policy is still in effect for the cache entry.
+    // The below provider should not get used, so the back off time will still be set to 10ms.
+    fakeBackoffProvider.nextPolicy = createBackoffPolicy(20, TimeUnit.MILLISECONDS);
+    // let it be throttled again
     resp = getInSyncContext(routeLookupRequest);
     assertThat(resp.hasError()).isTrue();
 
-    // Assert that the backoff policy is still in effect for the cache entry.
     fakeClock.forwardTime(9, TimeUnit.MILLISECONDS);
     resp = getInSyncContext(routeLookupRequest);
     assertThat(resp.hasError()).isTrue();