Fix potential dataloss (#56)

* Fix dataloss, do not ignore interrupt exceptions while commit thread is running

* Disable vertica integration tests due to missing image

Author: accountas

stream-loader-core/src/main/scala/com/adform/streamloader/sink/batch/RecordBatchingSinker.scala

@@ -73,7 +73,7 @@ class RecordBatchingSinker[B <: RecordBatch](
 
         log.info(s"Committing batch $batch to storage")
         Metrics.commitDuration.recordCallable(() =>
-          retryOnFailureIf(retryPolicy)(!batchCommittedAfterFailure(batch)) {
+          retryOnFailureIf(retryPolicy)(isRunning.get() && !batchCommittedAfterFailure(batch)) {
             batchStorage.commitBatch(batch)
           }
         )
@@ -86,7 +86,7 @@ class RecordBatchingSinker[B <: RecordBatch](
           log.warn("Failed discarding batch")
         }
       } catch {
-        case e if isInterruptionException(e) =>
+        case e if isInterruptionException(e) && !isRunning.get() =>
           log.debug("Batch commit thread interrupted")
       }
     },

stream-loader-core/src/main/scala/com/adform/streamloader/util/Retry.scala

@@ -46,7 +46,6 @@ object Retry extends Logging {
   def isInterruptionException(e: Throwable): Boolean = e match {
     case _: InterruptedException | _: InterruptedIOException | _: ClosedByInterruptException | _: InterruptException =>
       true
-    case e: Throwable if e.getCause != null && isInterruptionException(e.getCause) => true
     case _ => false
   }