Bug fix for not closing the RESTPubsub connections (#32)

Author: NathanSpeidel

src/main/java/com/yahoo/bullet/common/BulletConfig.java

@@ -126,7 +126,7 @@ public class BulletConfig extends Config {
     public static final int DEFAULT_WINDOW_MIN_EMIT_EVERY = 1000;
 
     public static final boolean DEFAULT_RATE_LIMIT_ENABLE = true;
-    public static final long DEFAULT_RATE_LIMIT_MAX_EMIT_COUNT = 100;
+    public static final long DEFAULT_RATE_LIMIT_MAX_EMIT_COUNT = 50;
     public static final long DEFAULT_RATE_LIMIT_TIME_INTERVAL = 100;
 
     public static final String DEFAULT_PUBSUB_CONTEXT_NAME = Context.QUERY_PROCESSING.name();
@@ -255,7 +255,7 @@ public class BulletConfig extends Config {
 
         VALIDATOR.define(PUBSUB_CONTEXT_NAME)
                  .defaultTo(DEFAULT_PUBSUB_CONTEXT_NAME)
-                 .checkIf(Validator.isIn(String.class, Context.QUERY_PROCESSING.name(), Context.QUERY_SUBMISSION.name()));
+                 .checkIf(Validator.isIn(Context.QUERY_PROCESSING.name(), Context.QUERY_SUBMISSION.name()));
         VALIDATOR.define(PUBSUB_CLASS_NAME)
                  .defaultTo(DEFAULT_PUBSUB_CLASS_NAME)
                  .checkIf(Validator::isString);

src/main/java/com/yahoo/bullet/pubsub/rest/RESTPublisher.java

@@ -9,8 +9,8 @@
 import com.yahoo.bullet.pubsub.Publisher;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.http.client.config.RequestConfig;
+import org.apache.http.client.methods.CloseableHttpResponse;
 import org.apache.http.client.methods.HttpPost;
-import org.apache.http.HttpResponse;
 import org.apache.http.entity.StringEntity;
 import org.apache.http.impl.client.CloseableHttpClient;
 import java.io.IOException;
@@ -51,13 +51,10 @@ public void close() {
      */
     protected void sendToURL(String url, PubSubMessage message) {
         log.debug("Sending message: {} to url: {}", message, url);
-        try {
-            HttpResponse response = client.execute(makeHttpPost(url, message));
+        try (CloseableHttpResponse response = client.execute(makeHttpPost(url, message))) {
             if (response == null || response.getStatusLine().getStatusCode() != RESTPubSub.OK_200) {
-                log.error("Couldn't reach REST pubsub server. Got response: {}", response);
-                return;
+                log.error("Couldn't POST to REST pubsub server. Got response: {}", response);
             }
-            log.debug("Successfully wrote message with status code {}. Response was: {}", response.getStatusLine().getStatusCode(), response);
         } catch (Exception e) {
             log.error("Error when trying to POST. Message was: {}. Error was: ", message.asJSON(), e);
         }
@@ -68,9 +65,7 @@ private HttpPost makeHttpPost(String url, PubSubMessage message) throws Unsuppor
         httpPost.setEntity(new StringEntity(message.asJSON()));
         httpPost.setHeader(CONTENT_TYPE, APPLICATION_JSON);
         RequestConfig requestConfig =
-                RequestConfig.custom().setConnectTimeout(connectTimeout)
-                        .setSocketTimeout(connectTimeout)
-                        .build();
+                RequestConfig.custom().setConnectTimeout(connectTimeout).setSocketTimeout(connectTimeout).build();
         httpPost.setConfig(requestConfig);
         return httpPost;
     }

src/main/java/com/yahoo/bullet/pubsub/rest/RESTSubscriber.java

@@ -11,8 +11,9 @@
 import lombok.Getter;
 import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.http.HttpResponse;
+import org.apache.http.HttpEntity;
 import org.apache.http.client.config.RequestConfig;
+import org.apache.http.client.methods.CloseableHttpResponse;
 import org.apache.http.client.methods.HttpGet;
 import org.apache.http.impl.client.CloseableHttpClient;
 import org.apache.http.util.EntityUtils;
@@ -58,20 +59,20 @@ public List<PubSubMessage> getMessages() {
         }
         lastRequest = currentTime;
         for (String url : urls) {
-            try {
-                log.debug("Getting messages from url: {}", url);
-                HttpResponse response = client.execute(makeHttpGet(url));
+            try (CloseableHttpResponse response = client.execute(makeHttpGet(url))) {
                 int statusCode = response.getStatusLine().getStatusCode();
                 if (statusCode == RESTPubSub.OK_200) {
-                    String message = EntityUtils.toString(response.getEntity(), RESTPubSub.UTF_8);
+                    HttpEntity httpEntity = response.getEntity();
+                    String message = EntityUtils.toString(httpEntity, RESTPubSub.UTF_8);
                     log.debug("Received message from url: {}. Message was {}", url, message);
                     messages.add(PubSubMessage.fromJSON(message));
+                    EntityUtils.consume(httpEntity);
                 } else if (statusCode != RESTPubSub.NO_CONTENT_204) {
                     // NO_CONTENT_204 indicates there are no new messages - anything else indicates a problem
-                    log.error("Http call failed with status code {} and response {}.", statusCode, response);
+                    log.error("HTTP call to {} failed with status code {} and response {}.", url, statusCode, response);
                 }
             } catch (Exception e) {
-                log.error("Http call to {} failed with error:", url, e);
+                log.error("HTTP call to {} failed with error:", url, e);
             }
         }
         return messages;
@@ -82,16 +83,14 @@ public void close() {
         try {
             client.close();
         } catch (IOException e) {
-            log.warn("Caught exception when closing AsyncHttpClient: ", e);
+            log.warn("Caught exception when closing HTTP client: ", e);
         }
     }
 
     private HttpGet makeHttpGet(String url) {
         HttpGet httpGet = new HttpGet(url);
         RequestConfig requestConfig =
-                RequestConfig.custom().setConnectTimeout(connectTimeout)
-                                      .setSocketTimeout(connectTimeout)
-                                      .build();
+                RequestConfig.custom().setConnectTimeout(connectTimeout).setSocketTimeout(connectTimeout).build();
         httpGet.setConfig(requestConfig);
         return httpGet;
     }

src/main/resources/bullet_defaults.yaml

@@ -214,7 +214,7 @@ bullet.query.window.min.emit.every.ms: 1000
 bullet.query.rate.limit.enable: true
 # This is the maximum amount of times data can be retrieved for a query in a given time interval (bullet.query.rate.limit.time.interval)
 # before it is considered as exceeding the rate limit.
-bullet.query.rate.limit.max.emit.count: 100
+bullet.query.rate.limit.max.emit.count: 50
 # This is the smallest interval in ms at which the check for whether the rate limit is being exceeded check is done if your backend uses rate limiting.
 bullet.query.rate.limit.time.interval: 100
 

src/test/java/com/yahoo/bullet/querying/RateLimitErrorTest.java

@@ -28,7 +28,7 @@ public void testMetaAndRateConversion() {
         Assert.assertTrue(((List<BulletError>) actual.get(Meta.ERROR_KEY)).get(0) == error);
 
         String asJSON = error.asJSON();
-        double defaultRate = (DEFAULT_RATE_LIMIT_MAX_EMIT_COUNT / DEFAULT_RATE_LIMIT_TIME_INTERVAL) * RateLimiter.SECOND;
+        double defaultRate = ((double) DEFAULT_RATE_LIMIT_MAX_EMIT_COUNT / DEFAULT_RATE_LIMIT_TIME_INTERVAL) * RateLimiter.SECOND;
         double actualRate = 19.34 * RateLimiter.SECOND;
         assertJSONEquals(asJSON, "{'error': '" + String.format(RateLimitError.ERROR_FORMAT, defaultRate, actualRate) + "', " +
                                   "'resolutions': ['" + RateLimitError.NARROW_FILTER + "', '" + RateLimitError.TIME_WINDOW + "']" +