Improve client partition table update push mechanism

When a partition table update is detected, a member pushes
updated partition table to its clients.

When there are many clients (hundreds or more), partition table updates
cause big latencies in migration system. Reason is, partition service's lock must be acquired
to create partition table's latest view and this is called on every partition update.

To fix that, we can skip some intermediate partition table updates. There's no need to push every update,
because once partition table updates begin, generally there'll be many. Most of them will be stale
in a short time. It's fine to skip some to reduce push frequency and lock contention.

hazelcast/src/main/java/com/hazelcast/client/impl/ClusterViewListenerService.java

@@ -29,6 +29,7 @@
 import com.hazelcast.internal.partition.PartitionReplica;
 import com.hazelcast.internal.partition.PartitionTableView;
 import com.hazelcast.internal.util.EmptyStatement;
+import com.hazelcast.internal.util.scheduler.CoalescingDelayedTrigger;
 import com.hazelcast.spi.impl.NodeEngine;
 import com.hazelcast.spi.impl.NodeEngineImpl;
 import com.hazelcast.spi.impl.executionservice.ExecutionService;
@@ -48,14 +49,20 @@
 
 public class ClusterViewListenerService {
     private static final int PUSH_PERIOD_IN_SECONDS = 30;
-    private final Map<ClientEndpoint, Long> clusterListeningEndpoints = new ConcurrentHashMap<ClientEndpoint, Long>();
+    private static final long PARTITION_UPDATE_DELAY_MS = 100;
+    private static final long PARTITION_UPDATE_MAX_DELAY_MS = 500;
+
+    private final Map<ClientEndpoint, Long> clusterListeningEndpoints = new ConcurrentHashMap<>();
     private final NodeEngine nodeEngine;
     private final boolean advancedNetworkConfigEnabled;
-    private AtomicBoolean pushScheduled = new AtomicBoolean();
+    private final AtomicBoolean pushScheduled = new AtomicBoolean();
+    private final CoalescingDelayedTrigger delayedPartitionUpdateTrigger;
 
     ClusterViewListenerService(NodeEngineImpl nodeEngine) {
         this.nodeEngine = nodeEngine;
         this.advancedNetworkConfigEnabled = nodeEngine.getConfig().getAdvancedNetworkConfig().isEnabled();
+        this.delayedPartitionUpdateTrigger = new CoalescingDelayedTrigger(nodeEngine.getExecutionService(),
+                PARTITION_UPDATE_DELAY_MS, PARTITION_UPDATE_MAX_DELAY_MS, this::pushPartitionTableView);
     }
 
     private void schedulePeriodicPush() {
@@ -64,20 +71,21 @@ private void schedulePeriodicPush() {
     }
 
     private void pushView() {
-        ClientMessage partitionViewMessage = getPartitionViewMessageOrNull();
-        if (partitionViewMessage != null) {
-            sendToListeningEndpoints(partitionViewMessage);
-        }
+        pushPartitionTableView();
         sendToListeningEndpoints(getMemberListViewMessage());
     }
 
-    public void onPartitionStateChange() {
+    private void pushPartitionTableView() {
         ClientMessage partitionViewMessage = getPartitionViewMessageOrNull();
         if (partitionViewMessage != null) {
             sendToListeningEndpoints(partitionViewMessage);
         }
     }
 
+    public void onPartitionStateChange() {
+        delayedPartitionUpdateTrigger.executeWithDelay();
+    }
+
     public void onMemberListChange() {
         sendToListeningEndpoints(getMemberListViewMessage());
     }