Fix bug: When sending event to Kafka fails for the first time, audit event needs to be inserted at the beginning of the queue. (#37)

bump version 0.8.0.1

Author: zzhhhzz

pom.xml

@@ -6,7 +6,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.pinterest.singer</groupId>
     <artifactId>singer-package</artifactId>
-    <version>0.8.0</version>
+    <version>0.8.0.1</version>
     <packaging>pom</packaging>
     <description>Singer Logging Agent modules</description>
     <inceptionYear>2013</inceptionYear>

singer-commons/pom.xml

@@ -20,7 +20,7 @@
   <parent>
     <groupId>com.pinterest.singer</groupId>
     <artifactId>singer-package</artifactId>
-    <version>0.8.0</version>
+    <version>0.8.0.1</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
   <developers>

singer-commons/src/main/java/com/pinterest/singer/loggingaudit/client/AuditEventKafkaSender.java

@@ -288,6 +288,7 @@ public void checkAndEnqueueWhenSendFailed(RecordMetadata recordMetadata, Excepti
       Integer count = eventTriedCount.get(event.getLoggingAuditHeaders());
       if (count == null){
         eventTriedCount.put(event.getLoggingAuditHeaders(), 1);
+        insertEvent(event);
         OpenTsdbMetricConverter
             .gauge(LoggingAuditClientMetrics.AUDIT_CLIENT_SENDER_KAFKA_EVENTS_RETRIED,
                 eventTriedCount.size(), "host=" + host, "stage=" + stage.toString(),
@@ -301,19 +302,23 @@ public void checkAndEnqueueWhenSendFailed(RecordMetadata recordMetadata, Excepti
           eventTriedCount.remove(event.getLoggingAuditHeaders());
       } else {
           eventTriedCount.put(event.getLoggingAuditHeaders(), count + 1);
-          try {
-            boolean success = queue.offerFirst(event, 3, TimeUnit.SECONDS);
-            if (!success) {
-              LOG.debug("Failed to enqueue LoggingAuditEvent at head of the queue when executing "
-                      + "producer send callback. Drop this event.");
-              eventTriedCount.remove(event.getLoggingAuditHeaders());
-            }
-          } catch (InterruptedException ex) {
-            LOG.debug(
-                "Enqueuing LoggingAuditEvent at head of the queue was interrupted in callback. "
-                    + "Drop this event");
-            eventTriedCount.remove(event.getLoggingAuditHeaders());
-          }
+          insertEvent(event);
+      }
+    }
+
+    public void insertEvent(LoggingAuditEvent event){
+      try {
+        boolean success = queue.offerFirst(event, 3, TimeUnit.SECONDS);
+        if (!success) {
+          LOG.debug("Failed to enqueue LoggingAuditEvent at head of the queue when executing "
+              + "producer send callback. Drop this event.");
+          eventTriedCount.remove(event.getLoggingAuditHeaders());
+        }
+      } catch (InterruptedException ex) {
+        LOG.debug(
+            "Enqueuing LoggingAuditEvent at head of the queue was interrupted in callback. "
+                + "Drop this event");
+        eventTriedCount.remove(event.getLoggingAuditHeaders());
       }
     }
 

singer/deb.version

@@ -1 +1 @@
-0.8.0
+0.8.0.1

singer/pom.xml

@@ -7,7 +7,7 @@
     <parent>
         <groupId>com.pinterest.singer</groupId>
         <artifactId>singer-package</artifactId>
-        <version>0.8.0</version>
+        <version>0.8.0.1</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <licenses>

thrift-logger/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>com.pinterest.singer</groupId>
     <artifactId>singer-package</artifactId>
-    <version>0.8.0</version>
+    <version>0.8.0.1</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
   <artifactId>thrift-logger</artifactId>