Fix applying client side service config override in INDIS flow (#995)

* fix applying client side service config override in INDIS flow

* bump minor version and add null guard

* fix SimpleLoadBalancerTest flaky failed issue

* tiny cleanup

---------

Co-authored-by: brycezhongqing <zhonchen+1@linkedin.com>

CHANGELOG.md

@@ -14,6 +14,9 @@ and what APIs have changed, if applicable.
 
 ## [Unreleased]
 
+## [29.52.0] - 2024-04-01
+- fix applying client side service config override in INDIS flow
+
 ## [29.51.14] - 2024-03-27
 - Support translating default values for optional non-record/union fields to Avro (when TRANSLATE_DEFAULT is enabled). 
 
@@ -5674,7 +5677,8 @@ patch operations can re-use these classes for generating patch messages.
 
 ## [0.14.1]
 
-[Unreleased]: https://github.com/linkedin/rest.li/compare/v29.51.14...master
+[Unreleased]: https://github.com/linkedin/rest.li/compare/v29.52.0...master
+[29.52.0]: https://github.com/linkedin/rest.li/compare/v29.51.14...v29.52.0
 [29.51.14]: https://github.com/linkedin/rest.li/compare/v29.51.13...v29.51.14
 [29.51.13]: https://github.com/linkedin/rest.li/compare/v29.51.12...v29.51.13
 [29.51.12]: https://github.com/linkedin/rest.li/compare/v29.51.11...v29.51.12

d2/src/main/java/com/linkedin/d2/balancer/properties/ServicePropertiesJsonSerializer.java

@@ -70,6 +70,10 @@ public ServicePropertiesJsonSerializer()
 
   public ServicePropertiesJsonSerializer(Map<String, Map<String, Object>> clientServicesConfig)
   {
+    if (clientServicesConfig == null)
+    {
+      clientServicesConfig = Collections.emptyMap();
+    }
     _clientServicesConfig = validateClientServicesConfig(clientServicesConfig);
   }
 

d2/src/main/java/com/linkedin/d2/xds/XdsToD2PropertiesAdaptor.java

@@ -61,7 +61,7 @@ public class XdsToD2PropertiesAdaptor
   private final XdsClient _xdsClient;
   private final List<XdsConnectionListener> _xdsConnectionListeners = Collections.synchronizedList(new ArrayList<>());
 
-  private final ServicePropertiesJsonSerializer _servicePropertiesJsonSerializer = new ServicePropertiesJsonSerializer();
+  private final ServicePropertiesJsonSerializer _servicePropertiesJsonSerializer;
   private final ClusterPropertiesJsonSerializer _clusterPropertiesJsonSerializer = new ClusterPropertiesJsonSerializer();
   private final UriPropertiesJsonSerializer _uriPropertiesJsonSerializer = new UriPropertiesJsonSerializer();
   private final UriPropertiesMerger _uriPropertiesMerger = new UriPropertiesMerger();
@@ -94,10 +94,17 @@ public class XdsToD2PropertiesAdaptor
 
   public XdsToD2PropertiesAdaptor(XdsClient xdsClient, DualReadStateManager dualReadStateManager,
       ServiceDiscoveryEventEmitter eventEmitter)
+  {
+    this(xdsClient, dualReadStateManager, eventEmitter, Collections.emptyMap());
+  }
+
+  public XdsToD2PropertiesAdaptor(XdsClient xdsClient, DualReadStateManager dualReadStateManager,
+      ServiceDiscoveryEventEmitter eventEmitter, Map<String, Map<String, Object>> clientServicesConfig)
   {
     _xdsClient = xdsClient;
     _dualReadStateManager = dualReadStateManager;
     _eventEmitter = eventEmitter;
+    _servicePropertiesJsonSerializer = new ServicePropertiesJsonSerializer(clientServicesConfig);
   }
 
   public void start()

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

@@ -62,7 +62,7 @@ public LoadBalancerWithFacilities create(D2ClientConfig config)
     d2ClientJmxManager.registerXdsClientJmx(xdsClient.getXdsClientJmx());
 
     XdsToD2PropertiesAdaptor adaptor = new XdsToD2PropertiesAdaptor(xdsClient, config.dualReadStateManager,
-        config.serviceDiscoveryEventEmitter);
+        config.serviceDiscoveryEventEmitter, config.clientServicesConfig);
 
     XdsLoadBalancer xdsLoadBalancer = new XdsLoadBalancer(adaptor, executorService,
         new XdsFsTogglingLoadBalancerFactory(config.lbWaitTimeout, config.lbWaitUnit, config.indisFsBasePath,

d2/src/test/java/com/linkedin/d2/xds/TestXdsToD2PropertiesAdaptor.java

@@ -37,11 +37,15 @@
 import indis.XdsD2;
 import java.net.URI;
 import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
 import org.mockito.Mock;
 import org.mockito.MockitoAnnotations;
 import org.testng.Assert;
+import org.testng.annotations.DataProvider;
 import org.testng.annotations.Test;
 
+import static com.linkedin.d2.balancer.properties.PropertyKeys.*;
 import static org.mockito.Mockito.*;
 
 
@@ -70,12 +74,39 @@ public class TestXdsToD2PropertiesAdaptor {
   private static final XdsClient.NodeUpdate EMPTY_NODE_DATA = new XdsClient.NodeUpdate(null);
   private static final XdsClient.D2URIMapUpdate EMPTY_DATA_URI_MAP = new XdsClient.D2URIMapUpdate(null);
 
-  @Test
-  public void testListenToService()
+  /* Provide {
+   * @clientOverride transport port client properties set on client override
+   * @original original transport client properties fetched from SD backend
+   * @expected overridden transport client properties after applying client override
+   * }
+   */
+  @DataProvider
+  public Object[][] provideTransportClientProperties()
+  {
+
+    Map<String, Object> original = new HashMap<>();
+    original.put(HTTP_REQUEST_TIMEOUT, "1000");
+    original.put(ALLOWED_CLIENT_OVERRIDE_KEYS,
+        Collections.singletonList(HTTP_REQUEST_TIMEOUT));
+
+    Map<String, Object> overridden = new HashMap<>();
+    overridden.put(HTTP_REQUEST_TIMEOUT, "20000");
+    overridden.put(ALLOWED_CLIENT_OVERRIDE_KEYS,
+        Collections.singletonList(HTTP_REQUEST_TIMEOUT));
+
+    return new Object[][]{
+        {Collections.emptyMap(), original, original},
+        {Collections.singletonMap(HTTP_REQUEST_TIMEOUT, "20000"), original, overridden}
+    };
+  }
+  @Test(dataProvider = "provideTransportClientProperties")
+  public void testListenToService(Map<String, Object> clientOverride, Map<String, Object> original,
+      Map<String, Object> overridden)
   {
     XdsToD2PropertiesAdaptorFixture fixture = new XdsToD2PropertiesAdaptorFixture();
     String serviceName = "FooService";
-    fixture.getSpiedAdaptor().listenToService(serviceName);
+    fixture.getSpiedAdaptor(Collections.singletonMap(serviceName, clientOverride))
+        .listenToService(serviceName);
 
     verify(fixture._xdsClient).watchXdsResource(eq("/d2/services/" + serviceName), anyNodeWatcher());
 
@@ -88,7 +119,10 @@ public void testListenToService()
                         serviceName,
                         PRIMARY_CLUSTER_NAME,
                         "",
-                        Collections.singletonList("relative")
+                        Collections.singletonList("relative"),
+                        Collections.emptyMap(),
+                        original,
+                        Collections.emptyMap(), Collections.emptyList(), Collections.emptySet()
                     )
                 )
             )
@@ -98,7 +132,10 @@ public void testListenToService()
     );
     verify(fixture._serviceEventBus).publishInitialize(serviceName,
         new ServiceStoreProperties(serviceName, PRIMARY_CLUSTER_NAME, "",
-            Collections.singletonList("relative"))
+            Collections.singletonList("relative"),
+            Collections.emptyMap(),
+            overridden,
+            Collections.<String, String>emptyMap(), Collections.emptyList(), Collections.emptySet())
     );
   }
 
@@ -375,7 +412,13 @@ private static class XdsToD2PropertiesAdaptorFixture
 
     XdsToD2PropertiesAdaptor getSpiedAdaptor()
     {
-      _adaptor = spy(new XdsToD2PropertiesAdaptor(_xdsClient, null, _eventEmitter));
+      return getSpiedAdaptor(Collections.emptyMap());
+    }
+
+    XdsToD2PropertiesAdaptor getSpiedAdaptor(Map<String, Map<String, Object>> clientServicesConfig)
+    {
+      _adaptor = spy(new XdsToD2PropertiesAdaptor(_xdsClient, null,
+          _eventEmitter, clientServicesConfig));
       _adaptor.setClusterEventBus(_clusterEventBus);
       _adaptor.setServiceEventBus(_serviceEventBus);
       _adaptor.setUriEventBus(_uriEventBus);

d2/src/test/java/com/linkedin/d2/xds/XdsToD2SampleClient.java

@@ -89,7 +89,7 @@ public static void main(String[] args) throws Exception
 
     XdsChannelFactory xdsChannelFactory = new XdsChannelFactory(sslContext, xdsServer);
     XdsClient xdsClient = new XdsClientImpl(node, xdsChannelFactory.createChannel(),
-        Executors.newSingleThreadScheduledExecutor(), XdsClientImpl.DEFAULT_READY_TIMEOUT_MILLIS);
+        Executors.newSingleThreadScheduledExecutor(), XdsClientImpl.DEFAULT_READY_TIMEOUT_MILLIS, false);
 
     DualReadStateManager dualReadStateManager = new DualReadStateManager(
         () -> DualReadModeProvider.DualReadMode.DUAL_READ,

gradle.properties

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