Remove stream parallel setting and optimize imports

Author: blootsvoets

src/main/java/org/radarbase/output/Application.kt

@@ -18,10 +18,10 @@ package org.radarbase.output
 
 import com.beust.jcommander.JCommander
 import com.beust.jcommander.ParameterException
-import kotlinx.coroutines.*
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.Semaphore
-import kotlinx.coroutines.sync.withLock
 import org.radarbase.output.accounting.*
 import org.radarbase.output.cleaner.SourceDataCleaner
 import org.radarbase.output.compression.Compression
@@ -33,7 +33,6 @@ import org.radarbase.output.source.SourceStorage
 import org.radarbase.output.source.SourceStorageFactory
 import org.radarbase.output.target.TargetStorage
 import org.radarbase.output.target.TargetStorageFactory
-import org.radarbase.output.util.SuspendedCloseable.Companion.useSuspended
 import org.radarbase.output.util.Timer
 import org.radarbase.output.worker.FileCacheStore
 import org.radarbase.output.worker.Job
@@ -89,8 +88,6 @@ class Application(
     override fun newFileCacheStore(accountant: Accountant) = FileCacheStore(this, accountant)
 
     fun start() {
-        System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism",
-            (config.worker.numThreads - 1).toString())
         System.setProperty("kotlinx.coroutines.scheduler.max.pool.size",
             config.worker.numThreads.toString())
         System.setProperty("kotlinx.coroutines.scheduler.core.pool.size",

src/main/java/org/radarbase/output/accounting/OffsetPersistenceFactory.kt

@@ -18,8 +18,6 @@ package org.radarbase.output.accounting
 
 import kotlinx.coroutines.CoroutineScope
 import org.radarbase.output.util.SuspendedCloseable
-import java.io.Closeable
-import java.io.Flushable
 import java.nio.file.Path
 
 /**

src/main/java/org/radarbase/output/cleaner/SourceDataCleaner.kt

@@ -1,8 +1,11 @@
 package org.radarbase.output.cleaner
 
-import kotlinx.coroutines.*
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.launch
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withPermit
+import kotlinx.coroutines.withContext
 import org.radarbase.output.Application.Companion.format
 import org.radarbase.output.FileStoreFactory
 import org.radarbase.output.accounting.Accountant

src/main/java/org/radarbase/output/format/CsvAvroConverterFactory.kt

@@ -11,7 +11,10 @@ import org.radarbase.output.util.TimeUtil.parseDateTime
 import org.radarbase.output.util.TimeUtil.parseTime
 import org.radarbase.output.util.TimeUtil.toDouble
 import org.slf4j.LoggerFactory
-import java.io.*
+import java.io.IOException
+import java.io.InputStream
+import java.io.Reader
+import java.io.Writer
 import java.nio.file.Path
 import kotlin.io.path.inputStream
 import kotlin.io.path.outputStream

src/main/java/org/radarbase/output/format/RecordConverterFactory.kt

@@ -23,7 +23,6 @@ import org.radarbase.output.compression.Compression
 import org.radarbase.output.util.ResourceContext.Companion.resourceContext
 import java.io.*
 import java.nio.file.Path
-import java.util.*
 import java.util.regex.Pattern
 import kotlin.collections.component1
 import kotlin.collections.component2

src/main/java/org/radarbase/output/source/S3SourceStorage.kt

@@ -3,8 +3,13 @@ package org.radarbase.output.source
 import io.minio.*
 import io.minio.errors.ErrorResponseException
 import io.minio.messages.Tags
-import kotlinx.coroutines.*
-import kotlinx.coroutines.flow.*
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.flowOn
+import kotlinx.coroutines.flow.retryWhen
+import kotlinx.coroutines.withContext
 import org.apache.avro.file.SeekableFileInput
 import org.apache.avro.file.SeekableInput
 import org.radarbase.output.config.S3Config

src/main/java/org/radarbase/output/source/SourceStorage.kt

@@ -1,10 +1,10 @@
 package org.radarbase.output.source
 
 import org.apache.avro.file.SeekableInput
-import org.radarbase.output.accounting.TopicPartitionOffsetRange
-import org.radarbase.output.util.*
 import org.radarbase.output.util.AvroFileLister.Companion.avroFileTreeLister
 import org.radarbase.output.util.AvroTopicLister.Companion.avroTopicTreeLister
+import org.radarbase.output.util.SuspendedCloseable
+import org.radarbase.output.util.TopicPath
 import java.nio.file.Path
 import java.time.Instant
 

src/main/java/org/radarbase/output/target/S3TargetStorage.kt

@@ -17,7 +17,6 @@
 package org.radarbase.output.target
 
 import io.minio.*
-import io.minio.errors.ErrorResponseException
 import org.radarbase.output.config.S3Config
 import org.radarbase.output.source.S3SourceStorage.Companion.faultTolerant
 import org.radarbase.output.util.bucketBuild

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

@@ -19,10 +19,8 @@ package org.radarbase.output.util
 import kotlinx.coroutines.*
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
-import org.slf4j.LoggerFactory
 import java.io.IOException
-import java.util.concurrent.*
-import java.util.concurrent.CancellationException
+import java.util.concurrent.TimeUnit
 
 /**
  * File writer where data is written in a separate thread with a timeout.

src/main/java/org/radarbase/output/util/Timer.kt

@@ -18,7 +18,6 @@ package org.radarbase.output.util
 
 import org.radarbase.output.util.ProgressBar.Companion.appendTime
 import java.time.Duration
-import java.util.*
 import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.ConcurrentMap
 import java.util.concurrent.atomic.LongAdder