Use a deep copy of the new best possible assignment for measuring baseline divergence. (#542)
The new assignment is really critical in waged rebalancer. If there is any potential changes in measure baseline divergence, waged rebalancer may not work correctly.
To avoid changes of the new assignment and make it safe when being used to measure baseline divergence, use a deep copy of the new assignment.
diff --git a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/WagedRebalancer.java b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/WagedRebalancer.java
index 7d52389..a88f809 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/WagedRebalancer.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/WagedRebalancer.java
@@ -401,11 +401,17 @@
// Asynchronously report baseline divergence metric before persisting to metadata store,
// just in case if persisting fails, we still have the metric.
+ // To avoid changes of the new assignment and make it safe when being used to measure baseline
+ // divergence, use a deep copy of the new assignment.
+ Map<String, ResourceAssignment> newAssignmentCopy = new HashMap<>();
+ for (Map.Entry<String, ResourceAssignment> entry : newAssignment.entrySet()) {
+ newAssignmentCopy.put(entry.getKey(), new ResourceAssignment(entry.getValue().getRecord()));
+ }
BaselineDivergenceGauge baselineDivergenceGauge = _metricCollector.getMetric(
WagedRebalancerMetricCollector.WagedRebalancerMetricNames.BaselineDivergenceGauge.name(),
BaselineDivergenceGauge.class);
baselineDivergenceGauge.asyncMeasureAndUpdateValue(clusterData.getAsyncTasksThreadPool(),
- currentBaseline, newAssignment);
+ currentBaseline, newAssignmentCopy);
if (_assignmentMetadataStore != null) {
try {
