fix(auto_balancer): fix potential memory leak (#1612) (#1613)

Signed-off-by: Shichao Nie <niesc@automq.com>

core/src/main/java/kafka/autobalancer/model/Snapshot.java

@@ -16,9 +16,9 @@
 
 public class Snapshot {
     public static final double INVALID = -1;
-    private final Snapshot prev;
     private final double[] values;
     private final double latest;
+    private Snapshot prev;
 
     public Snapshot(Snapshot prev, Collection<Double> values) {
         this.prev = prev;
@@ -68,6 +68,10 @@ public Snapshot getPrev() {
         return this.prev;
     }
 
+    public void setPrev(Snapshot prev) {
+        this.prev = prev;
+    }
+
     public int size() {
         return this.values.length;
     }

core/src/main/java/kafka/autobalancer/model/samples/SnapshotSamples.java

@@ -45,6 +45,9 @@ public void append(double value) {
     }
 
     public Snapshot snapshot() {
+        if (this.prev != null) {
+            this.prev.setPrev(null);
+        }
         Snapshot snapshot = new Snapshot(prev, values);
         this.prev = snapshot;
         return snapshot;

core/src/test/java/kafka/autobalancer/model/samples/SnapshotSamplesTest.java

@@ -42,4 +42,19 @@ public void testAppendAndSnapshot() {
         Assertions.assertEquals(1948, snapshot.get90thPercentile(), 1);
         Assertions.assertEquals(1744, snapshot.getValue(0.5), 1);
     }
+
+    @Test
+    public void testSnapshot() {
+        SnapshotSamples sequence = new SnapshotSamples(512);
+        for (int i = 0; i < 1000; i++) {
+            sequence.append(i);
+        }
+        for (int i = 0; i < 100; i++) {
+            sequence.snapshot();
+        }
+        Snapshot snapshot = sequence.snapshot();
+        Assertions.assertNotNull(snapshot);
+        Assertions.assertNotNull(snapshot.getPrev());
+        Assertions.assertNull(snapshot.getPrev().getPrev());
+    }
 }