Fix NullPointerException before resharding (close #156)

Author: chuwy

core/src/main/scala/com.snowplowanalytics.stream/loader/Emitter.scala

@@ -69,15 +69,16 @@ class Emitter(
   @throws[IOException]
   private def attemptEmit(records: List[EmitterJsonInput]): List[EmitterJsonInput] = {
     if (records.isEmpty) {
-      null
+      Nil
     } else {
       val (validRecords: List[EmitterJsonInput], invalidRecords: List[EmitterJsonInput]) =
         records.partition(_._2.isValid)
       // Send all valid records to stdout / Sink and return those rejected by it
       val rejects = goodSink match {
         case Some(s) =>
           validRecords.foreach {
-            case (_, record) => record.map(r => s.store(r.json.toString, None, true))
+            case (_, Validated.Valid(r)) => s.store(r.json.toString, None, true)
+            case _                       => ()
           }
           Nil
         case None if validRecords.isEmpty => Nil
@@ -153,7 +154,10 @@ class Emitter(
   /**
    * Closes the Sink client when the KinesisConnectorRecordProcessor is shut down
    */
-  override def shutdown(): Unit = bulkSender.close()
+  override def shutdown(): Unit = {
+    println("Shutting down emitter")
+    bulkSender.close()
+  }
 
   /**
    * Handles records rejected by the JsonTransformer or by Sink

core/src/main/scala/com.snowplowanalytics.stream/loader/clients/BulkSender.scala

@@ -62,6 +62,7 @@ trait BulkSender[A] {
    * Terminate the application in a way the KCL cannot stop, prevents shutdown hooks from running
    */
   protected def forceShutdown(): Unit = {
+    log.info("BulkSender force shutdown")
     tracker.foreach { t =>
       // TODO: Instead of waiting a fixed time, use synchronous tracking or futures (when the tracker supports futures)
       SnowplowTracking.trackApplicationShutdown(t)

elasticsearch/src/main/scala/com/snowplowanalytics/stream/loader/clients/ElasticsearchBulkSender.scala

@@ -95,7 +95,8 @@ class ElasticsearchBulkSender(
   }
 
   // do not close the es client, otherwise it will fail when resharding
-  override def close(): Unit = ()
+  override def close(): Unit =
+    log.info("Closing BulkSender")
 
   override def send(records: List[EmitterJsonInput]): List[EmitterJsonInput] = {
     val connectionAttemptStartTime = System.currentTimeMillis()
@@ -170,7 +171,7 @@ class ElasticsearchBulkSender(
 
   /** Logs the cluster health */
   override def logHealth(): Unit =
-    client.execute(clusterHealth) onComplete {
+    client.execute(clusterHealth).onComplete {
       case SSuccess(health) =>
         health match {
           case response =>