Fix Dualread monitoring log (#967)

* fix dualread monitoring log

* add retries to some tests for stability

CHANGELOG.md

@@ -14,6 +14,9 @@ and what APIs have changed, if applicable.
 
 ## [Unreleased]
 
+## [29.49.6] - 2024-01-12
+- fix dualread monitoring log
+
 ## [29.49.5] - 2024-01-11
 - Added KQueue support for domain sockets.
 
@@ -5609,7 +5612,8 @@ patch operations can re-use these classes for generating patch messages.
 
 ## [0.14.1]
 
-[Unreleased]: https://github.com/linkedin/rest.li/compare/v29.49.5...master
+[Unreleased]: https://github.com/linkedin/rest.li/compare/v29.49.6...master
+[29.49.6]: https://github.com/linkedin/rest.li/compare/v29.49.5...v29.49.6
 [29.49.5]: https://github.com/linkedin/rest.li/compare/v29.49.4...v29.49.5
 [29.49.4]: https://github.com/linkedin/rest.li/compare/v29.49.3...v29.49.4
 [29.49.3]: https://github.com/linkedin/rest.li/compare/v29.49.2...v29.49.3

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

@@ -16,6 +16,7 @@
 
 package com.linkedin.d2.balancer.dualread;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.cache.Cache;
 import com.google.common.cache.CacheBuilder;
 import com.google.common.cache.RemovalCause;
@@ -146,9 +147,10 @@ private Cache<String, CacheEntry<T>> buildCache()
         .build();
   }
 
-  private String getEntriesMessage(boolean fromNewLb, CacheEntry<T> oldE, CacheEntry<T> newE)
+  @VisibleForTesting
+  String getEntriesMessage(boolean fromNewLb, CacheEntry<T> oldE, CacheEntry<T> newE)
   {
-    return String.format("Old LB: {}, New LB: {}.",
+    return String.format("\nOld LB: %s\nNew LB: %s",
         fromNewLb? oldE : newE, fromNewLb? newE : oldE);
   }
 
@@ -157,7 +159,8 @@ private String getTimestamp() {
         .format(_format);
   }
 
-  private static final class CacheEntry<T>
+  @VisibleForTesting
+  static final class CacheEntry<T>
   {
     final String _version;
     final String _timeStamp;

d2/src/test/java/com/linkedin/d2/balancer/dualread/DualReadLoadBalancerMonitorTest.java

@@ -0,0 +1,57 @@
+package com.linkedin.d2.balancer.dualread;
+
+import com.linkedin.d2.balancer.properties.UriProperties;
+import com.linkedin.d2.util.TestDataHelper;
+import org.testng.Assert;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+import static com.linkedin.d2.util.TestDataHelper.*;
+
+
+public class DualReadLoadBalancerMonitorTest {
+
+  @DataProvider
+  public Object[][] getEntriesMessageDataProvider()
+  {
+    return new Object[][] {
+        {
+          true, new DualReadLoadBalancerMonitor.CacheEntry<>("", "", PROPERTIES_1),
+            new DualReadLoadBalancerMonitor.CacheEntry<>("", "", PROPERTIES_2),
+            "\nOld LB: CacheEntry{_version=, _timeStamp='', _data=UriProperties [_clusterName=TestCluster,"
+                + " _urisBySchemeAndPartition={http={0=[http://google.com:1], 1=[http://google.com:1]}},"
+                + " _partitions={http://google.com:1={0=[ weight =1.0 ], 1=[ weight =2.0 ]}},"
+                + " _uriSpecificProperties={}]}\n"
+                + "New LB: CacheEntry{_version=, _timeStamp='', "
+                + "_data=UriProperties [_clusterName=TestCluster, "
+                + "_urisBySchemeAndPartition={http={1=[http://linkedin.com:2]}}, "
+                + "_partitions={http://linkedin.com:2={1=[ weight =0.5 ]}}, _uriSpecificProperties={}]}"
+        },
+        {
+          false, new DualReadLoadBalancerMonitor.CacheEntry<>("", "", PROPERTIES_1),
+            new DualReadLoadBalancerMonitor.CacheEntry<>("", "", PROPERTIES_2),
+            "\nOld LB: CacheEntry{_version=, _timeStamp='', _data=UriProperties [_clusterName=TestCluster,"
+                + " _urisBySchemeAndPartition={http={1=[http://linkedin.com:2]}}, "
+                + "_partitions={http://linkedin.com:2={1=[ weight =0.5 ]}}, _uriSpecificProperties={}]}\n"
+                + "New LB: CacheEntry{_version=, _timeStamp='', _data=UriProperties [_clusterName=TestCluster,"
+                + " _urisBySchemeAndPartition={http={0=[http://google.com:1], 1=[http://google.com:1]}},"
+                + " _partitions={http://google.com:1={0=[ weight =1.0 ], 1=[ weight =2.0 ]}},"
+                + " _uriSpecificProperties={}]}"
+        }
+    };
+  }
+  @Test(dataProvider = "getEntriesMessageDataProvider")
+  public void testGetEntriesMessage(Boolean isFromNewLb,
+      DualReadLoadBalancerMonitor.CacheEntry<UriProperties> oldE,
+      DualReadLoadBalancerMonitor.CacheEntry<UriProperties> newE,
+      String expected)
+  {
+    DualReadLoadBalancerMonitor.UriPropertiesDualReadMonitor monitor =
+        new DualReadLoadBalancerMonitor.UriPropertiesDualReadMonitor(
+            new DualReadLoadBalancerJmx(), TestDataHelper.getClock());
+
+    Assert.assertEquals(monitor.getEntriesMessage(isFromNewLb, oldE, newE),
+        expected,
+        "entry message is not the same");
+  }
+}

d2/src/test/java/com/linkedin/d2/balancer/util/WarmUpLoadBalancerTest.java

@@ -108,7 +108,7 @@ public void afterTest() throws IOException
     _dualReadStateManager = null;
   }
 
-  @Test
+  @Test(retryAnalyzer = ThreeRetries.class)
   public void testMakingWarmUpRequests() throws URISyntaxException, InterruptedException, ExecutionException, TimeoutException
   {
     createDefaultServicesIniFiles();
@@ -122,12 +122,12 @@ public void testMakingWarmUpRequests() throws URISyntaxException, InterruptedExc
 
     FutureCallback<None> callback = new FutureCallback<>();
     warmUpLoadBalancer.start(callback);
-    callback.get(30, TimeUnit.MILLISECONDS); // 3 services should take at most 3 * 5ms
+    callback.get(50, TimeUnit.MILLISECONDS); // 3 services should take at most 3 * 5ms
 
     Assert.assertEquals(VALID_FILES.size(), requestCount.get());
   }
 
-  @Test(timeOut = 10000, groups = { "ci-flaky" })
+  @Test(timeOut = 10000, retryAnalyzer = ThreeRetries.class)
   public void testDeletingFilesAfterShutdown() throws InterruptedException, ExecutionException, TimeoutException
   {
     createDefaultServicesIniFiles();
@@ -167,7 +167,7 @@ public void testDeletingFilesAfterShutdown() throws InterruptedException, Execut
    * require additional services at runtime, we have to check that those services are not cleared from the cache
    * otherwise it would incur in a penalty at the next deployment
    */
-  @Test(timeOut = 10000)
+  @Test(timeOut = 10000, retryAnalyzer = ThreeRetries.class)
   public void testNotDeletingFilesGetClient() throws InterruptedException, ExecutionException, TimeoutException, ServiceUnavailableException
   {
     createDefaultServicesIniFiles();
@@ -225,7 +225,7 @@ private List<String> getPartialDownstreams() throws InterruptedException, Execut
   /**
    * If there are 0 valid files, no requests should be triggered
    */
-  @Test
+  @Test(retryAnalyzer = ThreeRetries.class)
   public void testNoMakingWarmUpRequestsWithoutValidFiles() throws URISyntaxException, InterruptedException, ExecutionException, TimeoutException
   {
     createServicesIniFiles(UNVALID_FILES);
@@ -247,7 +247,7 @@ public void testNoMakingWarmUpRequestsWithoutValidFiles() throws URISyntaxExcept
   /**
    * Should not send warm up requests if we are NOT using the WarmUpLoadBalancer
    */
-  @Test
+  @Test(retryAnalyzer = ThreeRetries.class)
   public void testNoMakingWarmUpRequestsWithoutWarmUp() throws URISyntaxException, InterruptedException, ExecutionException, TimeoutException
   {
     createDefaultServicesIniFiles();
@@ -262,7 +262,7 @@ public void testNoMakingWarmUpRequestsWithoutWarmUp() throws URISyntaxException,
     Assert.assertEquals(0, requestCount.get());
   }
 
-  @Test(timeOut = 10000)
+  @Test(timeOut = 10000, retryAnalyzer = ThreeRetries.class)
   public void testThrottling() throws InterruptedException
   {
     int NRequests = 100;
@@ -298,7 +298,7 @@ public void testThrottling() throws InterruptedException
   /**
    * Tests that if the requests are not throttled it makes a large amount of concurrent calls
    */
-  @Test(timeOut = 10000)
+  @Test(timeOut = 10000, retryAnalyzer = ThreeRetries.class)
   public void testThrottlingUnlimitedRequests() throws URISyntaxException, InterruptedException, ExecutionException, TimeoutException
   {
     int NRequests = 500;
@@ -332,7 +332,7 @@ public void testThrottlingUnlimitedRequests() throws URISyntaxException, Interru
     Assert.assertEquals(NRequests, requestCount.get());
   }
 
-  @Test(timeOut = 10000)
+  @Test(timeOut = 10000, retryAnalyzer = ThreeRetries.class)
   public void testHitTimeout() throws URISyntaxException, InterruptedException, ExecutionException, TimeoutException
   {
     int NRequests = 5000;

d2/src/test/java/com/linkedin/d2/util/TestDataHelper.java

@@ -5,6 +5,7 @@
 import com.linkedin.d2.balancer.properties.UriProperties;
 import com.linkedin.d2.discovery.event.D2ServiceDiscoveryEventHelper;
 import com.linkedin.d2.discovery.event.ServiceDiscoveryEventEmitter;
+import com.linkedin.util.clock.Clock;
 import java.net.URI;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -240,4 +241,10 @@ public Long get() {
       }
     };
   }
+
+  public static Clock getClock()
+  {
+    Supplier<Long> timeSupplier = TestDataHelper.getTimeSupplier(100);
+    return () -> timeSupplier.get();
+  }
 }

gradle.properties

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