code: private void updateClusterMetrics(String topologyId, TopologyMetric tpMetric) {
          if (tpMetric.get_topologyMetric().get_metrics_size() == 0) {
              return;
          }
  
          MetricInfo topologyMetrics = tpMetric.get_topologyMetric();
          // make a new MetricInfo to save the topologyId's metric
          MetricInfo clusterMetrics = MetricUtils.mkMetricInfo();
          Set<String> metricNames = new HashSet<>();
  
          for (Map.Entry<String, Map<Integer, MetricSnapshot>> entry : topologyMetrics.get_metrics().entrySet()) {
              String metricName = MetricUtils.topo2clusterName(entry.getKey());
              MetricType metricType = MetricUtils.metricType(metricName);
              Map<Integer, MetricSnapshot> winData = new HashMap<>();
  
              for (Map.Entry<Integer, MetricSnapshot> entryData : entry.getValue().entrySet()) {
                  MetricSnapshot snapshot = entryData.getValue().deepCopy();
                  winData.put(entryData.getKey(), snapshot);
                  if (metricType == MetricType.HISTOGRAM) {
                      // reset topology metric points
                      entryData.getValue().set_points(new byte[0]);
                      entryData.getValue().set_pointSize(0);
                  }
              }
              clusterMetrics.put_to_metrics(metricName, winData);
              metricNames.add(metricName);
          }
  
          // save to local cache, waiting for merging
          TopologyMetricContext clusterTpMetricContext = context.getClusterTopologyMetricContext();
          clusterTpMetricContext.addToMemCache(topologyId, clusterMetrics);
          context.registerMetrics(JStormMetrics.CLUSTER_METRIC_KEY, metricNames);
  
      }
docs: update cluster metrics local cache