SUMOK-15 Modified the way throttling gets handled

Author: Juan Pablo Diaz-Vaz

src/main/java/com/mcplusa/kinesis/KinesisConnectorExecutor.java

@@ -128,34 +128,4 @@ private void setupInputStream() {
     private static boolean parseBoolean(String property, boolean defaultValue, Properties properties) {
         return Boolean.parseBoolean(properties.getProperty(property, Boolean.toString(defaultValue)));
     }
-
-    /**
-     * Helper method used to parse long properties.
-     * 
-     * @param property
-     *        The String key for the property
-     * @param defaultValue
-     *        The default value for the long property
-     * @param properties
-     *        The properties file to get property from
-     * @return property from property file, or if it is not specified, the default value
-     */
-    private static long parseLong(String property, long defaultValue, Properties properties) {
-        return Long.parseLong(properties.getProperty(property, Long.toString(defaultValue)));
-    }
-
-    /**
-     * Helper method used to parse integer properties.
-     * 
-     * @param property
-     *        The String key for the property
-     * @param defaultValue
-     *        The default value for the integer property
-     * @param properties
-     *        The properties file to get property from
-     * @return property from property file, or if it is not specified, the default value
-     */
-    private static int parseInt(String property, int defaultValue, Properties properties) {
-        return Integer.parseInt(properties.getProperty(property, Integer.toString(defaultValue)));
-    }
-}
+}

src/main/java/com/mcplusa/kinesis/KinesisConnectorExecutorBase.java

@@ -36,7 +36,7 @@ public abstract class KinesisConnectorExecutorBase<T, U> implements Runnable {
      * @param kinesisConnectorConfiguration Amazon Kinesis connector configuration
      */
     protected void initialize(KinesisConnectorConfiguration kinesisConnectorConfiguration) {
-        initialize(kinesisConnectorConfiguration, null);
+        initialize(kinesisConnectorConfiguration, new NullMetricsFactory());
     }
 
     /**
@@ -77,8 +77,6 @@ protected void initialize(KinesisConnectorConfiguration kinesisConnectorConfigur
         if (kinesisConnectorConfiguration.IDLE_TIME_BETWEEN_READS > kinesisConnectorConfiguration.BUFFER_MILLISECONDS_LIMIT) {
             LOG.warn("idleTimeBetweenReads is greater than bufferTimeMillisecondsLimit. For best results, ensure that bufferTimeMillisecondsLimit is more than or equal to idleTimeBetweenReads ");
         }
-
-        metricFactory = new NullMetricsFactory(); // TODO move higher in the hierarchy
 
         // If a metrics factory was specified, use it.
         if (metricFactory != null) {

src/main/java/com/mcplusa/sumologic/KinesisMessageModelSumologicTransformer.java

@@ -15,7 +15,7 @@ public class KinesisMessageModelSumologicTransformer extends
         SumologicTransformer<KinesisMessageModel> {
 
     /**
-     * Creates a new KinesisMessageModelDynamoDBTransformer.
+     * Creates a new KinesisMessageModelSumologicTransformer.
      */
     public KinesisMessageModelSumologicTransformer() {
         super(KinesisMessageModel.class);

src/main/java/com/mcplusa/sumologic/SumologicExecutor.java

@@ -27,7 +27,6 @@ public SumologicExecutor(String configFile) {
 
     /**
      * Main method starts and runs the SumologicExecutor.
-     * 
      * @param args
      */
     public static void main(String[] args) {  

src/main/java/com/mcplusa/sumologic/SumologicSender.java

@@ -14,7 +14,6 @@
 import org.apache.http.protocol.HTTP;
 import org.apache.http.util.EntityUtils;
 import org.apache.http.impl.client.DefaultHttpClient;
-
 import org.apache.http.impl.conn.tsccm.ThreadSafeClientConnManager;
 
 public class SumologicSender {
@@ -25,7 +24,9 @@ public class SumologicSender {
 
   private int connectionTimeout = 1000;
   private int socketTimeout = 60000;
-
+  private static final int RETRIES = 3;
+  private static final int SLEEP_TIME = 1000;
+  
 
 	public SumologicSender(String url) {
 		  this.url = url;
@@ -37,18 +38,31 @@ public SumologicSender(String url) {
 	}
 
 	public boolean sendToSumologic(String data) throws IOException{
-  	  HttpPost post = null;
-    post = new HttpPost(url);
-    post.setEntity(new StringEntity(data, HTTP.PLAIN_TEXT_TYPE, HTTP.UTF_8));
-    HttpResponse response = httpClient.execute(post);
-    int statusCode = response.getStatusLine().getStatusCode();
+	  int retries = RETRIES;
+	  int sleep_time = SLEEP_TIME;
+	  int statusCode;
+	  
+	  do {
+    	  HttpPost post = null;
+      post = new HttpPost(url);
+      post.setEntity(new StringEntity(data, HTTP.PLAIN_TEXT_TYPE, HTTP.UTF_8));
+      HttpResponse response = httpClient.execute(post);
+      statusCode = response.getStatusLine().getStatusCode();
+      
+      //need to consume the body if you want to re-use the connection.
+      EntityUtils.consume(response.getEntity());
+      try { 
+        post.abort(); 
+      } catch (Exception ignore) {}
+      if (statusCode == 429) {
+        LOG.warn("Got TOO MANY REQUESTS from Sumologic");
+        retries--;
+        try {
+          Thread.sleep(sleep_time);
+        } catch (InterruptedException ignore) {}
+      }
+	  } while (statusCode == 429 && retries > 0);
 
-    //need to consume the body if you want to re-use the connection.
-    EntityUtils.consume(response.getEntity());
-    try { 
-      post.abort(); 
-    } catch (Exception ignore) {}
- 
     // Check if the request was successful;
     if (statusCode != 200) {
       LOG.warn(String.format("Received HTTP error from Sumo Service: %d", statusCode));