From 11af873a67666845dd75830e330d74eb4a4a0316 Mon Sep 17 00:00:00 2001
From: Henning Andersen <33268011+henningandersen@users.noreply.github.com>
Date: Mon, 21 Sep 2020 20:55:47 +0200
Subject: [PATCH] Cardinality request breaker leak (#62685) (#62718)

If HyperLogLogPlusPlus failed during construction, it would
not release already allocated resources, causing the request
circuit breaker to not be adjusted down.

Closes #62439
---
 .../metrics/HyperLogLogPlusPlus.java          |  7 ++-
 .../metrics/HyperLogLogPlusPlusTests.java     | 45 +++++++++++++++++++
 2 files changed, 51 insertions(+), 1 deletion(-)

diff --git a/server/src/main/java/org/elasticsearch/search/aggregations/metrics/HyperLogLogPlusPlus.java b/server/src/main/java/org/elasticsearch/search/aggregations/metrics/HyperLogLogPlusPlus.java
index 9c3caccf9ef83..2540300af2599 100644
--- a/server/src/main/java/org/elasticsearch/search/aggregations/metrics/HyperLogLogPlusPlus.java
+++ b/server/src/main/java/org/elasticsearch/search/aggregations/metrics/HyperLogLogPlusPlus.java
@@ -784,7 +784,12 @@ public HyperLogLogPlusPlus(int precision, BigArrays bigArrays, long initialBucke
         this.bigArrays = bigArrays;
         algorithm = new OpenBitSet();
         runLens = bigArrays.newByteArray(initialBucketCount << p);
-        hashSet = new Hashset(initialBucketCount);
+        try {
+            hashSet = new Hashset(initialBucketCount);
+        } catch (RuntimeException e) {
+            runLens.close();
+            throw e;
+        }
         final double alpha;
         switch (p) {
             case 4:
diff --git a/server/src/test/java/org/elasticsearch/search/aggregations/metrics/HyperLogLogPlusPlusTests.java b/server/src/test/java/org/elasticsearch/search/aggregations/metrics/HyperLogLogPlusPlusTests.java
index 514af2a67667d..e018d00b23a5b 100644
--- a/server/src/test/java/org/elasticsearch/search/aggregations/metrics/HyperLogLogPlusPlusTests.java
+++ b/server/src/test/java/org/elasticsearch/search/aggregations/metrics/HyperLogLogPlusPlusTests.java
@@ -21,13 +21,22 @@
 
 import com.carrotsearch.hppc.BitMixer;
 import com.carrotsearch.hppc.IntHashSet;
+import org.elasticsearch.common.breaker.CircuitBreaker;
+import org.elasticsearch.common.breaker.CircuitBreakingException;
+import org.elasticsearch.common.breaker.NoopCircuitBreaker;
 import org.elasticsearch.common.util.BigArrays;
 import org.elasticsearch.search.aggregations.metrics.HyperLogLogPlusPlus;
+import org.elasticsearch.indices.breaker.CircuitBreakerService;
 import org.elasticsearch.test.ESTestCase;
 
+import java.util.concurrent.atomic.AtomicLong;
+
 import static org.elasticsearch.search.aggregations.metrics.HyperLogLogPlusPlus.MAX_PRECISION;
 import static org.elasticsearch.search.aggregations.metrics.HyperLogLogPlusPlus.MIN_PRECISION;
 import static org.hamcrest.Matchers.closeTo;
+import static org.hamcrest.Matchers.equalTo;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
 public class HyperLogLogPlusPlusTests extends ESTestCase {
     public void testEncodeDecode() {
@@ -127,4 +136,40 @@ public void testPrecisionFromThreshold() {
         assertEquals(18, HyperLogLogPlusPlus.precisionFromThreshold(100000));
         assertEquals(18, HyperLogLogPlusPlus.precisionFromThreshold(1000000));
     }
+
+    public void testCircuitBreakerOnConstruction() {
+        int whenToBreak = randomInt(10);
+        AtomicLong total = new AtomicLong();
+        CircuitBreakerService breakerService = mock(CircuitBreakerService.class);
+        when(breakerService.getBreaker(CircuitBreaker.REQUEST)).thenReturn(new NoopCircuitBreaker(CircuitBreaker.REQUEST) {
+            private int countDown = whenToBreak;
+            @Override
+            public double addEstimateBytesAndMaybeBreak(long bytes, String label) throws CircuitBreakingException {
+                if (countDown-- == 0) {
+                    throw new CircuitBreakingException("test error", bytes, Long.MAX_VALUE, Durability.TRANSIENT);
+                }
+                total.addAndGet(bytes);
+                return total.get();
+            }
+
+            @Override
+            public long addWithoutBreaking(long bytes) {
+                total.addAndGet(bytes);
+                return total.get();
+            }
+        });
+        BigArrays bigArrays = new BigArrays(null, breakerService, CircuitBreaker.REQUEST).withCircuitBreaking();
+        final int p = randomIntBetween(HyperLogLogPlusPlus.MIN_PRECISION, HyperLogLogPlusPlus.MAX_PRECISION);
+        try {
+            for (int i = 0; i < whenToBreak + 1; ++i) {
+                final HyperLogLogPlusPlus subject = new HyperLogLogPlusPlus(p, bigArrays, 0);
+                subject.close();
+            }
+            fail("Must fail");
+        } catch (CircuitBreakingException e) {
+            // OK
+        }
+
+        assertThat(total.get(), equalTo(0L));
+    }
 }