Fix dual-read potential risk by using ThreadPool

CHANGELOG.md

@@ -14,6 +14,9 @@ and what APIs have changed, if applicable.
 
 ## [Unreleased]
 
+## [29.46.10] - 2023-11-13
+Fix dual-read potential risk that newLb may impact oldLb
+
 ## [29.46.9] - 2023-11-02
 - Update FieldDef so that it will lazily cache the hashCode.
 
@@ -5557,7 +5560,8 @@ patch operations can re-use these classes for generating patch messages.
 
 ## [0.14.1]
 
-[Unreleased]: https://github.com/linkedin/rest.li/compare/v29.46.9...master
+[Unreleased]: https://github.com/linkedin/rest.li/compare/v29.46.10...master
+[29.46.10]: https://github.com/linkedin/rest.li/compare/v29.46.9...v29.46.10
 [29.46.9]: https://github.com/linkedin/rest.li/compare/v29.46.8...v29.46.9
 [29.46.8]: https://github.com/linkedin/rest.li/compare/v29.46.7...v29.46.8
 [29.46.7]: https://github.com/linkedin/rest.li/compare/v29.46.6...v29.46.7

d2/src/main/java/com/linkedin/d2/balancer/D2ClientBuilder.java

@@ -201,7 +201,8 @@ public D2Client build()
                   _config.serviceDiscoveryEventEmitter,
                   _config.dualReadStateManager,
                   _config.xdsExecutorService,
-                  _config.xdsStreamReadyTimeout
+                  _config.xdsStreamReadyTimeout,
+                  _config.loadBalancerExecutor
     );
 
     final LoadBalancerWithFacilitiesFactory loadBalancerFactory = (_config.lbWithFacilitiesFactory == null) ?
@@ -643,6 +644,11 @@ public D2ClientBuilder setDualReadStateManager(DualReadStateManager dualReadStat
     return this;
   }
 
+  public D2ClientBuilder setLoadBalancerExecutor(ExecutorService loadBalancerExecutor) {
+    _config.loadBalancerExecutor = loadBalancerExecutor;
+    return this;
+  }
+
   /**
    * Single-threaded executor service for xDS communication.
    */

d2/src/main/java/com/linkedin/d2/balancer/D2ClientConfig.java

@@ -43,6 +43,7 @@
 import io.grpc.netty.shaded.io.netty.handler.ssl.SslContext;
 import java.util.Collections;
 import java.util.Map;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
@@ -127,6 +128,7 @@ public class D2ClientConfig
 
   public ScheduledExecutorService xdsExecutorService = null;
   public Long xdsStreamReadyTimeout = null;
+  public ExecutorService loadBalancerExecutor = null;
 
   public D2ClientConfig()
   {
@@ -195,7 +197,8 @@ public D2ClientConfig()
                  ServiceDiscoveryEventEmitter serviceDiscoveryEventEmitter,
                  DualReadStateManager dualReadStateManager,
                  ScheduledExecutorService xdsExecutorService,
-                 Long xdsStreamReadyTimeout)
+                 Long xdsStreamReadyTimeout,
+                 ExecutorService loadBalancerExecutor)
   {
     this.zkHosts = zkHosts;
     this.xdsServer = xdsServer;
@@ -261,5 +264,6 @@ public D2ClientConfig()
     this.dualReadStateManager = dualReadStateManager;
     this.xdsExecutorService = xdsExecutorService;
     this.xdsStreamReadyTimeout = xdsStreamReadyTimeout;
+    this.loadBalancerExecutor = loadBalancerExecutor;
   }
 }

d2/src/main/java/com/linkedin/d2/balancer/dualread/DualReadLoadBalancer.java

@@ -16,6 +16,7 @@
 
 package com.linkedin.d2.balancer.dualread;
 
+import com.google.common.util.concurrent.MoreExecutors;
 import com.linkedin.common.callback.Callback;
 import com.linkedin.common.callback.Callbacks;
 import com.linkedin.common.util.None;
@@ -34,6 +35,7 @@
 import com.linkedin.r2.message.RequestContext;
 import com.linkedin.r2.transport.common.TransportClientFactory;
 import com.linkedin.r2.transport.common.bridge.client.TransportClient;
+import java.util.concurrent.ExecutorService;
 import javax.annotation.Nonnull;
 import org.apache.commons.lang3.tuple.Pair;
 import org.slf4j.Logger;
@@ -59,16 +61,35 @@ public class DualReadLoadBalancer implements LoadBalancerWithFacilities
   private final LoadBalancerWithFacilities _oldLb;
   private final LoadBalancerWithFacilities _newLb;
   private final DualReadStateManager _dualReadStateManager;
-
+  private ExecutorService _executor;
   private boolean _isNewLbReady;
 
+  @Deprecated
   public DualReadLoadBalancer(LoadBalancerWithFacilities oldLb, LoadBalancerWithFacilities newLb,
       @Nonnull DualReadStateManager dualReadStateManager)
+  {
+    this(oldLb, newLb, dualReadStateManager, null);
+    LOG.warn("Deprecated DualReadLoadBalancer constructor used without a threadpool");
+  }
+
+  public DualReadLoadBalancer(LoadBalancerWithFacilities oldLb, LoadBalancerWithFacilities newLb,
+      @Nonnull DualReadStateManager dualReadStateManager, ExecutorService executor)
   {
     _oldLb = oldLb;
     _newLb = newLb;
     _dualReadStateManager = dualReadStateManager;
     _isNewLbReady = false;
+    if(executor == null)
+    {
+      // Using a direct executor here means the code is executed directly,
+      // blocking the caller. This means the old behavior is preserved.
+      _executor = MoreExecutors.newDirectExecutorService();
+      LOG.warn("Deprecated DualReadLoadBalancer constructor used without a threadpool executor");
+    }
+    else
+    {
+      _executor = executor;
+    }
   }
 
   @Override
@@ -109,35 +130,36 @@ public void getClient(Request request, RequestContext requestContext, Callback<T
         _newLb.getClient(request, requestContext, clientCallback);
         break;
       case DUAL_READ:
-        _newLb.getLoadBalancedServiceProperties(serviceName, new Callback<ServiceProperties>()
-        {
-          @Override
-          public void onError(Throwable e)
-          {
-            LOG.error("Double read failure. Unable to read service properties from: " + serviceName, e);
-          }
-
-          @Override
-          public void onSuccess(ServiceProperties result)
-          {
-            String clusterName = result.getClusterName();
-            _dualReadStateManager.updateCluster(clusterName, DualReadModeProvider.DualReadMode.DUAL_READ);
-            _newLb.getLoadBalancedClusterAndUriProperties(clusterName, new Callback<Pair<ClusterProperties, UriProperties>>()
+        _executor.execute(
+            () -> _newLb.getLoadBalancedServiceProperties(serviceName, new Callback<ServiceProperties>()
             {
               @Override
               public void onError(Throwable e)
               {
-                LOG.error("Dual read failure. Unable to read cluster properties from: " + clusterName, e);
+                LOG.error("Double read failure. Unable to read service properties from: {}", serviceName, e);
               }
 
               @Override
-              public void onSuccess(Pair<ClusterProperties, UriProperties> result)
+              public void onSuccess(ServiceProperties result)
               {
-                LOG.debug("Dual read is successful. Get cluster and uri properties: " + result);
+                String clusterName = result.getClusterName();
+                _dualReadStateManager.updateCluster(clusterName, DualReadModeProvider.DualReadMode.DUAL_READ);
+                _newLb.getLoadBalancedClusterAndUriProperties(clusterName, new Callback<Pair<ClusterProperties, UriProperties>>()
+                    {
+                      @Override
+                      public void onError(Throwable e)
+                      {
+                        LOG.error("Dual read failure. Unable to read cluster properties from: {}", clusterName, e);
+                      }
+
+                      @Override
+                      public void onSuccess(Pair<ClusterProperties, UriProperties> result)
+                      {
+                        LOG.debug("Dual read is successful. Get cluster and uri properties: {}", result);
+                      }
+                    });
               }
-            });
-          }
-        });
+            }));
         _oldLb.getClient(request, requestContext, clientCallback);
         break;
       case OLD_LB_ONLY:
@@ -155,7 +177,7 @@ public void getLoadBalancedServiceProperties(String serviceName, Callback<Servic
         _newLb.getLoadBalancedServiceProperties(serviceName, clientCallback);
         break;
       case DUAL_READ:
-        _newLb.getLoadBalancedServiceProperties(serviceName, Callbacks.empty());
+        _executor.execute(() -> _newLb.getLoadBalancedServiceProperties(serviceName, Callbacks.empty()));
         _oldLb.getLoadBalancedServiceProperties(serviceName, clientCallback);
         break;
       case OLD_LB_ONLY:
@@ -174,7 +196,7 @@ public void getLoadBalancedClusterAndUriProperties(String clusterName,
         _newLb.getLoadBalancedClusterAndUriProperties(clusterName, callback);
         break;
       case DUAL_READ:
-        _newLb.getLoadBalancedClusterAndUriProperties(clusterName, Callbacks.empty());
+        _executor.execute(() -> _newLb.getLoadBalancedClusterAndUriProperties(clusterName, Callbacks.empty()));
         _oldLb.getLoadBalancedClusterAndUriProperties(clusterName, callback);
         break;
       case OLD_LB_ONLY:

d2/src/main/java/com/linkedin/d2/xds/balancer/DualReadZkAndXdsLoadBalancerFactory.java

@@ -44,6 +44,6 @@ public DualReadZkAndXdsLoadBalancerFactory(@Nonnull DualReadStateManager dualRea
   @Override
   public LoadBalancerWithFacilities create(D2ClientConfig config)
   {
-    return new DualReadLoadBalancer(_zkLbFactory.create(config), _xdsLbFactory.create(config), _dualReadStateManager);
+    return new DualReadLoadBalancer(_zkLbFactory.create(config), _xdsLbFactory.create(config), _dualReadStateManager, config.loadBalancerExecutor);
   }
 }

gradle.properties

@@ -1,4 +1,4 @@
-version=29.46.9
+version=29.46.10
 group=com.linkedin.pegasus
 org.gradle.configureondemand=true
 org.gradle.parallel=true