Put all config in separate files and introduce ResourceContext

Author: blootsvoets

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

@@ -36,14 +36,14 @@ import org.radarbase.output.worker.RadarKafkaRestructure
 import org.slf4j.LoggerFactory
 import redis.clients.jedis.JedisPool
 import java.io.IOException
-import java.nio.file.Files
 import java.text.NumberFormat
 import java.time.LocalDateTime
 import java.time.format.DateTimeFormatter
 import java.util.concurrent.Executors
 import java.util.concurrent.TimeUnit
 import java.util.concurrent.atomic.LongAdder
 import kotlin.Long.Companion.MAX_VALUE
+import kotlin.io.path.createDirectories
 import kotlin.system.exitProcess
 
 /** Main application.  */
@@ -84,7 +84,7 @@ class Application(
                 (config.worker.numThreads - 1).toString())
 
         try {
-            Files.createDirectories(config.paths.temp)
+            config.paths.temp.createDirectories()
         } catch (ex: IOException) {
             logger.error("Failed to create temporary directory")
             return

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

@@ -23,9 +23,10 @@ import org.slf4j.LoggerFactory
 import java.io.Closeable
 import java.io.Flushable
 import java.io.IOException
-import java.nio.file.Files
 import java.nio.file.Paths
 import java.time.Instant
+import kotlin.io.path.deleteExisting
+import kotlin.io.path.exists
 
 open class Accountant @Throws(IOException::class)
 constructor(factory: FileStoreFactory, topic: String) : Flushable, Closeable {
@@ -53,9 +54,9 @@ constructor(factory: FileStoreFactory, topic: String) : Flushable, Closeable {
                 .resolve(OFFSETS_FILE_NAME)
                 .resolve("$topic.csv")
 
-        return if (Files.exists(offsetsPath)) {
+        return if (offsetsPath.exists()) {
             OffsetFilePersistence(factory.targetStorage).read(offsetsPath)
-                    .also { Files.delete(offsetsPath) }
+                    .also { offsetsPath.deleteExisting() }
         } else null
     }
 

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

@@ -21,11 +21,12 @@ import org.radarbase.output.util.PostponedWriter
 import org.radarbase.output.util.Timer.time
 import org.slf4j.LoggerFactory
 import java.io.IOException
-import java.nio.file.Files
 import java.nio.file.Path
 import java.time.Instant
 import java.util.concurrent.TimeUnit
 import java.util.regex.Pattern
+import kotlin.io.path.bufferedWriter
+import kotlin.io.path.createTempFile
 
 /**
  * Accesses a OffsetRange file using the CSV format. On writing, this will create the file if
@@ -38,13 +39,11 @@ class OffsetFilePersistence(
         return try {
             if (targetStorage.status(path) != null) {
                 OffsetRangeSet().also { set ->
-                    targetStorage.newBufferedReader(path).use { br ->
-                        // ignore header
-                        br.readLine() ?: return@use
-
-                        generateSequence { br.readLine() }
-                                .map(::parseLine)
-                                .forEach(set::add)
+                    targetStorage.newBufferedReader(path).useLines { lines ->
+                        lines
+                            .drop(1)  // ignore header
+                            .map(::parseLine)
+                            .forEach(set::add)
                     }
                 }
             } else null
@@ -94,9 +93,9 @@ class OffsetFilePersistence(
 
         override fun doWrite() = time("accounting.offsets") {
             try {
-                val tmpPath = Files.createTempFile("offsets", ".csv")
+                val tmpPath = createTempFile("offsets", ".csv")
 
-                Files.newBufferedWriter(tmpPath).use { writer ->
+                tmpPath.bufferedWriter().use { writer ->
                     writer.append("offsetFrom,offsetTo,partition,topic\n")
                     offsets.forEach { topicPartition, offsetIntervals ->
                         offsetIntervals.forEach { offsetFrom, offsetTo, lastModified ->

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

@@ -2,6 +2,7 @@ package org.radarbase.output.cleaner
 
 import org.radarbase.output.FileStoreFactory
 import org.radarbase.output.accounting.Accountant
+import org.radarbase.output.util.ResourceContext.Companion.resourceContext
 import org.radarbase.output.util.Timer
 import org.slf4j.LoggerFactory
 import java.io.Closeable
@@ -61,10 +62,10 @@ class SourceDataCleaner(
 
         return try {
             lockManager.tryRunLocked(topic) {
-                Accountant(fileStoreFactory, topic).use { accountant ->
-                    TimestampExtractionCheck(sourceStorage, fileStoreFactory).use { extractionCheck ->
-                        deleteOldFiles(accountant, extractionCheck, topic, topicPath).toLong()
-                    }
+                resourceContext {
+                    val accountant = createResource { Accountant(fileStoreFactory, topic) }
+                    val extractionCheck = createResource { TimestampExtractionCheck(sourceStorage, fileStoreFactory) }
+                    deleteOldFiles(accountant, extractionCheck, topic, topicPath).toLong()
                 }
             }
         } catch (ex: IOException) {

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

@@ -23,7 +23,6 @@ import org.slf4j.LoggerFactory
 import java.io.FileNotFoundException
 import java.io.IOException
 import java.nio.file.Path
-import java.util.*
 
 /**
  * Caches open file handles. If more than the limit is cached, the half of the files that were used

src/main/java/org/radarbase/output/config/AzureConfig.kt

@@ -0,0 +1,50 @@
+package org.radarbase.output.config
+
+import com.azure.core.credential.BasicAuthenticationCredential
+import com.azure.storage.blob.BlobServiceClient
+import com.azure.storage.blob.BlobServiceClientBuilder
+import com.azure.storage.common.StorageSharedKeyCredential
+import org.radarbase.output.config.RestructureConfig.Companion.copyEnv
+import org.slf4j.LoggerFactory
+
+data class AzureConfig(
+    /** URL to reach object store at. */
+    val endpoint: String,
+    /** Name of the Azure Blob Storage container. */
+    val container: String,
+    /** If no endOffset is in the filename, read it from object metadata. */
+    val endOffsetFromMetadata: Boolean = false,
+    /** Azure username. */
+    val username: String?,
+    /** Azure password. */
+    val password: String?,
+    /** Shared Azure Blob Storage account name. */
+    val accountName: String?,
+    /** Shared Azure Blob Storage account key. */
+    val accountKey: String?,
+    /** Azure SAS token for a configured service. */
+    val sasToken: String?,
+) {
+    fun createAzureClient(): BlobServiceClient = BlobServiceClientBuilder().apply {
+        endpoint(endpoint)
+        when {
+            !username.isNullOrEmpty() && !password.isNullOrEmpty() -> credential(
+                BasicAuthenticationCredential(username, password))
+            !accountName.isNullOrEmpty() && !accountKey.isNullOrEmpty() -> credential(
+                StorageSharedKeyCredential(accountName, accountKey))
+            !sasToken.isNullOrEmpty() -> sasToken(sasToken)
+            else -> logger.warn("No Azure credentials supplied. Assuming a public blob storage.")
+        }
+    }.buildClient()
+
+    fun withEnv(prefix: String): AzureConfig = this
+        .copyEnv("${prefix}AZURE_USERNAME") { copy(username = it) }
+        .copyEnv("${prefix}AZURE_PASSWORD") { copy(password = it) }
+        .copyEnv("${prefix}AZURE_ACCOUNT_NAME") { copy(accountName = it) }
+        .copyEnv("${prefix}AZURE_ACCOUNT_KEY") { copy(accountKey = it) }
+        .copyEnv("${prefix}AZURE_SAS_TOKEN") { copy(sasToken = it) }
+
+    companion object {
+        private val logger = LoggerFactory.getLogger(AzureConfig::class.java)
+    }
+}

src/main/java/org/radarbase/output/config/CleanerConfig.kt

@@ -0,0 +1,18 @@
+package org.radarbase.output.config
+
+data class CleanerConfig(
+    /** Whether to enable the cleaner. */
+    val enable: Boolean = false,
+    /** How often to run the cleaner in seconds. */
+    val interval: Long = 1260L,
+    /** Age in days after an avro file can be removed. Must be strictly positive. */
+    val age: Int = 7,
+    /** Maximum number of files to clean in a given topic. */
+    val maxFilesPerTopic: Int? = null,
+) {
+    fun validate() {
+        check(age > 0) { "Cleaner file age must be strictly positive" }
+        check(interval > 0) { "Cleaner interval must be strictly positive" }
+        if (maxFilesPerTopic != null) check(maxFilesPerTopic > 0) { "Maximum files per topic must be strictly positive" }
+    }
+}

src/main/java/org/radarbase/output/config/CompressionConfig.kt

@@ -0,0 +1,14 @@
+package org.radarbase.output.config
+
+import org.radarbase.output.compression.Compression
+import org.radarbase.output.compression.CompressionFactory
+
+data class CompressionConfig(
+    override val factory: String = CompressionFactory::class.qualifiedName!!,
+    override val properties: Map<String, String> = emptyMap(),
+    /** Compression type. Currently one of gzip, zip or none. */
+    val type: String = "none",
+) : PluginConfig {
+    fun createFactory(): CompressionFactory = factory.toPluginInstance(properties)
+    fun createCompression(): Compression = createFactory()[type]
+}

src/main/java/org/radarbase/output/config/DeduplicationConfig.kt

@@ -0,0 +1,27 @@
+package org.radarbase.output.config
+
+import org.radarbase.output.config.RestructureConfig.Companion.copyOnChange
+
+data class DeduplicationConfig(
+    /** Whether to enable deduplication. */
+    val enable: Boolean? = null,
+    /**
+     * Only deduplicate using given fields. Fields not specified here are ignored
+     * for determining duplication.
+     */
+    val distinctFields: Set<String>? = null,
+    /**
+     * Ignore given fields for determining whether a row is identical to another.
+     */
+    val ignoreFields: Set<String>? = null,
+) {
+    fun withDefaults(deduplicationDefaults: DeduplicationConfig): DeduplicationConfig =
+        deduplicationDefaults
+            .copyOnChange<DeduplicationConfig, Boolean?>(null, { enable }) { copy(enable = it) }
+            .copyOnChange<DeduplicationConfig, Set<String>?>(null, { distinctFields }) {
+                copy(distinctFields = it)
+            }
+            .copyOnChange<DeduplicationConfig, Set<String>?>(null, { ignoreFields }) {
+                copy(ignoreFields = it)
+            }
+}

src/main/java/org/radarbase/output/config/Extensions.kt

@@ -0,0 +1,12 @@
+package org.radarbase.output.config
+
+import org.radarbase.output.Plugin
+
+internal inline fun <reified T : Plugin> String.toPluginInstance(properties: Map<String, String>): T {
+    return try {
+        (Class.forName(this).getConstructor().newInstance() as T)
+            .also { it.init(properties) }
+    } catch (ex: ReflectiveOperationException) {
+        throw IllegalStateException("Cannot map class $this to ${T::class.java.name}")
+    }
+}