Save in-progress changes.

Author: kannanjgithub

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

@@ -134,6 +134,7 @@ final class CachingRlsLbClient {
   private final ChannelLogger logger;
   private final ChildPolicyWrapper fallbackChildPolicyWrapper;
   private ConnectivityState lastRlsServerConnectivityState;
+  private boolean wasInTransientFailure;
 
   static {
     MetricInstrumentRegistry metricInstrumentRegistry
@@ -228,8 +229,7 @@ private CachingRlsLbClient(Builder builder) {
     refCountedChildPolicyWrapperFactory =
         new RefCountedChildPolicyWrapperFactory(
             lbPolicyConfig.getLoadBalancingPolicy(), childLbResolvedAddressFactory,
-            childLbHelperProvider,
-            new BackoffRefreshListener());
+            childLbHelperProvider);
     // TODO(creamsoup) wait until lb is ready
     String defaultTarget = lbPolicyConfig.getRouteLookupConfig().defaultTarget();
     if (defaultTarget != null && !defaultTarget.isEmpty()) {
@@ -261,12 +261,20 @@ public void accept(BatchRecorder recorder) {
   }
 
   private void rlsServerConnectionStateChanged() {
-    ConnectivityState currentState = rlsChannel.getState(false);
-    if (lastRlsServerConnectivityState == ConnectivityState.TRANSIENT_FAILURE
-      && currentState == ConnectivityState.READY) {
-
+    lastRlsServerConnectivityState = rlsChannel.getState(false);
+    if (wasInTransientFailure && lastRlsServerConnectivityState == ConnectivityState.READY) {
+      wasInTransientFailure = false;
+      synchronized (lock) {
+        for (CacheEntry value : linkedHashLruCache.values()) {
+          if (value instanceof BackoffCacheEntry) {
+            refreshBackoffEntry((BackoffCacheEntry) value);
+          }
+        }
+      }
+    } else if (!wasInTransientFailure && lastRlsServerConnectivityState == ConnectivityState.TRANSIENT_FAILURE) {
+      wasInTransientFailure = true;
     }
-    lastRlsServerConnectivityState = currentState;
+    rlsChannel.notifyWhenStateChanged(lastRlsServerConnectivityState, () -> rlsServerConnectionStateChanged());
   }
 
   void init() {
@@ -469,7 +477,8 @@ private void refreshBackoffEntry(BackoffCacheEntry entry) {
       logger.log(ChannelLogLevel.DEBUG,
           "[RLS Entry {0}] Calling RLS for transition to pending", entry.request);
       linkedHashLruCache.invalidate(entry.request);
-      asyncRlsCall(entry.request, entry.backoffPolicy);
+      // Cache updated. updateBalancingState() to reattempt picks
+      helper.triggerPendingRpcProcessing();
     }
   }
 

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

@@ -210,20 +210,17 @@ static final class RefCountedChildPolicyWrapperFactory {
         new HashMap<>();
 
     private final ChildLoadBalancerHelperProvider childLbHelperProvider;
-    private final ChildLbStatusListener childLbStatusListener;
     private final ChildLoadBalancingPolicy childPolicy;
     private ResolvedAddressFactory childLbResolvedAddressFactory;
 
     public RefCountedChildPolicyWrapperFactory(
         ChildLoadBalancingPolicy childPolicy,
         ResolvedAddressFactory childLbResolvedAddressFactory,
-        ChildLoadBalancerHelperProvider childLbHelperProvider,
-        ChildLbStatusListener childLbStatusListener) {
+        ChildLoadBalancerHelperProvider childLbHelperProvider) {
       this.childPolicy = checkNotNull(childPolicy, "childPolicy");
       this.childLbResolvedAddressFactory =
           checkNotNull(childLbResolvedAddressFactory, "childLbResolvedAddressFactory");
       this.childLbHelperProvider = checkNotNull(childLbHelperProvider, "childLbHelperProvider");
-      this.childLbStatusListener = checkNotNull(childLbStatusListener, "childLbStatusListener");
     }
 
     void init() {
@@ -248,8 +245,7 @@ ChildPolicyWrapper createOrGet(String target) {
       RefCountedChildPolicyWrapper pooledChildPolicyWrapper = childPolicyMap.get(target);
       if (pooledChildPolicyWrapper == null) {
         ChildPolicyWrapper childPolicyWrapper = new ChildPolicyWrapper(
-            target, childPolicy, childLbResolvedAddressFactory, childLbHelperProvider,
-            childLbStatusListener);
+            target, childPolicy, childLbResolvedAddressFactory, childLbHelperProvider);
         pooledChildPolicyWrapper = RefCountedChildPolicyWrapper.of(childPolicyWrapper);
         childPolicyMap.put(target, pooledChildPolicyWrapper);
         return pooledChildPolicyWrapper.getObject();
@@ -299,11 +295,9 @@ public ChildPolicyWrapper(
         String target,
         ChildLoadBalancingPolicy childPolicy,
         final ResolvedAddressFactory childLbResolvedAddressFactory,
-        ChildLoadBalancerHelperProvider childLbHelperProvider,
-        ChildLbStatusListener childLbStatusListener) {
+        ChildLoadBalancerHelperProvider childLbHelperProvider) {
       this.target = target;
-      this.helper =
-          new ChildPolicyReportingHelper(childLbHelperProvider, childLbStatusListener);
+      this.helper = new ChildPolicyReportingHelper(childLbHelperProvider);
       LoadBalancerProvider lbProvider = childPolicy.getEffectiveLbProvider();
       final ConfigOrError lbConfig =
           lbProvider
@@ -386,14 +380,11 @@ public String toString() {
     final class ChildPolicyReportingHelper extends ForwardingLoadBalancerHelper {
 
       private final ChildLoadBalancerHelper delegate;
-      private final ChildLbStatusListener listener;
 
       ChildPolicyReportingHelper(
-          ChildLoadBalancerHelperProvider childHelperProvider,
-          ChildLbStatusListener listener) {
+          ChildLoadBalancerHelperProvider childHelperProvider) {
         checkNotNull(childHelperProvider, "childHelperProvider");
         this.delegate = childHelperProvider.forTarget(getTarget());
-        this.listener = checkNotNull(listener, "listener");
       }
 
       @Override
@@ -406,7 +397,6 @@ public void updateBalancingState(ConnectivityState newState, SubchannelPicker ne
         picker = newPicker;
         state = newState;
         super.updateBalancingState(newState, newPicker);
-        listener.onStatusChanged(newState);
       }
     }
   }

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

@@ -78,8 +78,7 @@ public ResolvedAddresses create(Object childLbConfig) {
               ImmutableMap.<String, Object>of("foo", "bar"),
               lbProvider),
           resolvedAddressFactory,
-          new ChildLoadBalancerHelperProvider(helper, subchannelStateManager, picker),
-          childLbStatusListener);
+          new ChildLoadBalancerHelperProvider(helper, subchannelStateManager, picker));
 
   @Before
   public void setUp() {