Null safety fixes

Author: blootsvoets

src/main/java/org/radarbase/hdfs/util/PostponedWriter.kt

@@ -58,17 +58,18 @@ abstract class PostponedWriter
      * not yet taken place, the write will occur earlier.
      */
     fun triggerWrite() {
-        var localWriteFuture: Future<*>? = writeFuture.get()
-        if (localWriteFuture == null) {
-            localWriteFuture = executor.schedule({ this.startWrite() }, timeout, timeoutUnit)
-            if (!writeFuture.compareAndSet(null, localWriteFuture)) {
-                localWriteFuture!!.cancel(false)
-            }
+        if (writeFuture.get() == null) {
+            executor.schedule(::startWrite, timeout, timeoutUnit)
+                    .also { newWriteFuture ->
+                        if (!writeFuture.compareAndSet(null, newWriteFuture)) {
+                            newWriteFuture.cancel(false)
+                        }
+                    }
         }
     }
 
     /** Start the write in the writer thread.  */
-    protected fun startWrite() {
+    private fun startWrite() {
         writeFuture.set(null)
         doWrite()
     }
@@ -89,8 +90,9 @@ abstract class PostponedWriter
 
     @Throws(IOException::class)
     private fun doFlush(shutdown: Boolean) {
-        val localFuture = executor.submit { this.startWrite() }
-        writeFuture.set(localFuture)
+        val localFuture = executor.submit(::startWrite)
+                .also { writeFuture.set(it) }
+
         if (shutdown) {
             executor.shutdown()
         }

src/main/java/org/radarbase/hdfs/worker/FileCache.kt

@@ -128,7 +128,7 @@ class FileCache(
         writer.close()
 
         if (!hasError.get()) {
-            if (deduplicate.enable!!) {
+            if (deduplicate.enable == true) {
                 time("close.deduplicate") {
                     val dedupTmp = tmpPath.resolveSibling("${tmpPath.fileName}.dedup")
                     converterFactory.deduplicate(fileName, tmpPath, dedupTmp, compression, deduplicate.distinctFields ?: emptySet(), deduplicate.ignoreFields ?: emptySet())

src/main/java/org/radarbase/hdfs/worker/RestructureWorker.kt

@@ -130,11 +130,11 @@ internal class RestructureWorker(
     }
 
     @Throws(IOException::class)
-    private fun writeRecord(topicPartition: TopicPartition, record: GenericRecord?,
+    private fun writeRecord(topicPartition: TopicPartition, record: GenericRecord,
                             offset: Long, suffix: Int = 0) {
         var currentSuffix = suffix
         val (path) = pathFactory.getRecordOrganization(
-                topicPartition.topic, record!!, currentSuffix)
+                topicPartition.topic, record, currentSuffix)
 
         val transaction = time("accounting.create") {
             Accountant.Transaction(topicPartition, offset)