Merge pull request #533 from RADAR-base/release-2.3.1

Release 2.3.1

Author: blootsvoets

.github/workflows/scheduled_snyk.yaml

@@ -11,7 +11,7 @@ jobs:
       - uses: actions/checkout@v3
       - uses: snyk/actions/setup@master
         with:
-          snyk-version: v1.931.0
+          snyk-version: v1.1032.0
 
       - uses: actions/setup-java@v3
         with:

.github/workflows/snyk.yaml

@@ -2,15 +2,15 @@ name: Snyk test
 on:
   pull_request:
     branches:
-      - master
+      - main
 jobs:
   security:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v3
       - uses: snyk/actions/setup@master
         with:
-          snyk-version: v1.931.0
+          snyk-version: v1.1032.0
 
       - uses: actions/setup-java@v3
         with:

README.md

@@ -70,7 +70,7 @@ When upgrading to version 0.6.0 from version 0.5.x or earlier, please follow the
 This package is available as docker image [`radarbase/radar-output-restructure`](https://hub.docker.com/r/radarbase/radar-output-restructure). The entrypoint of the image is the current application. So in all the commands listed in usage, replace `radar-output-restructure` with for example:
 
 ```shell
-docker run --rm -t --network s3 -v "$PWD/output:/output" radarbase/radar-output-restructure:2.3.0 -o /output /myTopic
+docker run --rm -t --network s3 -v "$PWD/output:/output" radarbase/radar-output-restructure:2.3.1 -o /output /myTopic
 ```
 
 ## Command line usage
@@ -228,7 +228,7 @@ This package requires at least Java JDK 8. Build the distribution with
 and install the package into `/usr/local` with for example
 ```shell
 sudo mkdir -p /usr/local
-sudo tar -xzf build/distributions/radar-output-restructure-2.3.0.tar.gz -C /usr/local --strip-components=1
+sudo tar -xzf build/distributions/radar-output-restructure-2.3.1.tar.gz -C /usr/local --strip-components=1
 ```
 
 Now the `radar-output-restructure` command should be available.

build.gradle.kts

@@ -16,7 +16,7 @@ plugins {
 }
 
 group = "org.radarbase"
-version = "2.3.0"
+version = "2.3.1"
 
 repositories {
     mavenCentral()
@@ -57,7 +57,10 @@ dependencies {
     val jacksonVersion: String by project
     api(platform("com.fasterxml.jackson:jackson-bom:$jacksonVersion"))
     implementation("com.fasterxml.jackson.core:jackson-databind")
-    implementation("com.fasterxml.jackson.dataformat:jackson-dataformat-yaml")
+    implementation("com.fasterxml.jackson.dataformat:jackson-dataformat-yaml") {
+        val snakeYamlVersion: String by project
+        runtimeOnly("org.yaml:snakeyaml:$snakeYamlVersion")
+    }
     implementation("com.fasterxml.jackson.dataformat:jackson-dataformat-csv")
     implementation("com.fasterxml.jackson.module:jackson-module-kotlin")
     implementation("com.fasterxml.jackson.datatype:jackson-datatype-jsr310")
@@ -86,14 +89,17 @@ dependencies {
         implementation(platform("io.netty:netty-bom:$nettyVersion"))
     }
     val opencsvVersion: String by project
-    implementation("com.opencsv:opencsv:$opencsvVersion")
+    implementation("com.opencsv:opencsv:$opencsvVersion") {
+        val apacheCommonsTextVersion: String by project
+        runtimeOnly("org.apache.commons:commons-text:$apacheCommonsTextVersion")
+    }
 
     val slf4jVersion: String by project
     implementation("org.slf4j:slf4j-api:$slf4jVersion")
 
     val log4jVersion: String by project
-    runtimeOnly("org.apache.logging.log4j:log4j-slf4j-impl:$log4jVersion")
-    runtimeOnly("org.apache.logging.log4j:log4j-api:$log4jVersion")
+    runtimeOnly("org.apache.logging.log4j:log4j-core:$log4jVersion")
+    runtimeOnly("org.apache.logging.log4j:log4j-slf4j2-impl:$log4jVersion")
     runtimeOnly("org.apache.logging.log4j:log4j-jul:$log4jVersion")
 
     val radarSchemasVersion: String by project

gradle.properties

@@ -1,28 +1,30 @@
 kotlin.code.style=official
 
-kotlinVersion=1.7.10
-dokkaVersion=1.7.10
+kotlinVersion=1.7.20
+dokkaVersion=1.7.20
 dockerComposeVersion=0.16.9
-dependencyUpdateVersion=0.42.0
+dependencyUpdateVersion=0.43.0
 nexusPublishVersion=1.1.0
 jsoupVersion=1.15.3
 
 coroutinesVersion=1.6.4
 avroVersion=1.11.1
 snappyVersion=1.1.8.4
-jacksonVersion=2.13.4
+jacksonVersion=2.13.4.20221013
 jCommanderVersion=1.82
 almworksVersion=1.1.2
-minioVersion=8.4.3
+minioVersion=8.4.5
 guavaVersion=31.1-jre
 opencsvVersion=5.7.0
 okhttpVersion=4.10.0
-jedisVersion=4.2.3
-slf4jVersion=1.7.36
-log4jVersion=2.18.0
-azureStorageVersion=12.19.0
-nettyVersion=4.1.80.Final
+jedisVersion=4.3.0
+slf4jVersion=2.0.3
+log4jVersion=2.19.0
+azureStorageVersion=12.20.0
+nettyVersion=4.1.84.Final
+snakeYamlVersion=1.32
+apacheCommonsTextVersion=1.10.0
 
-junitVersion=5.9.0
+junitVersion=5.9.1
 mockitoKotlinVersion=4.0.0
 radarSchemasVersion=0.7.9

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

@@ -34,13 +34,13 @@ interface Accountant : SuspendedCloseable {
         internal val offsets: OffsetRangeSet = OffsetRangeSet { DirectFunctionalValue(it) }
 
         fun add(transaction: Transaction) = time("accounting.add") {
-            offsets.add(transaction.topicPartition, transaction.offset, transaction.lastModified)
+            offsets.add(transaction)
         }
     }
 
-    class Transaction(
+    data class Transaction(
         val topicPartition: TopicPartition,
-        internal var offset: Long,
-        internal val lastModified: Instant,
+        val offset: Long,
+        val lastModified: Instant,
     )
 }

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

@@ -30,38 +30,29 @@ import java.util.concurrent.ConcurrentMap
  * when they overlap or are adjacent. When merging offsets, last processing time is taken as the
  * maximum of all the merged processing times.
  */
-class OffsetRangeSet {
-    private val factory: (OffsetIntervals) -> FunctionalValue<OffsetIntervals>
-    private val ranges: ConcurrentMap<TopicPartition, FunctionalValue<OffsetIntervals>>
+class OffsetRangeSet(
+    private val ranges: ConcurrentMap<TopicPartition, FunctionalValue<OffsetIntervals>>,
+    private val factory: (OffsetIntervals) -> FunctionalValue<OffsetIntervals>,
+) {
 
     /** Whether the stored offsets is empty.  */
     val isEmpty: Boolean
         get() = ranges.isEmpty()
 
-    @JvmOverloads
     constructor(
         factory: (OffsetIntervals) -> FunctionalValue<OffsetIntervals> = { LockedFunctionalValue(it) },
-    ) {
-        this.ranges = ConcurrentHashMap()
-        this.factory = factory
-    }
-
-    private constructor(
-        ranges: ConcurrentMap<TopicPartition, FunctionalValue<OffsetIntervals>>,
-        factory: (OffsetIntervals) -> FunctionalValue<OffsetIntervals>,
-    ) {
-        this.ranges = ranges
-        this.factory = factory
-    }
+    ) : this(ConcurrentHashMap(), factory)
 
     /** Add given offset range to seen offsets.  */
     fun add(range: TopicPartitionOffsetRange) {
         range.topicPartition.modifyIntervals { it.add(range.range) }
     }
 
     /** Add given single offset to seen offsets.  */
-    fun add(topicPartition: TopicPartition, offset: Long, lastModified: Instant) {
-        topicPartition.modifyIntervals { it.add(offset, lastModified) }
+    fun add(transaction: Accountant.Transaction) {
+        transaction.run {
+            topicPartition.modifyIntervals { it.add(offset, lastModified) }
+        }
     }
 
     /** Add all offset stream of given set to the current set.  */
@@ -81,23 +72,18 @@ class OffsetRangeSet {
     }
 
     /** Whether this range value completely contains the given range.  */
-    operator fun contains(range: TopicPartitionOffsetRange): Boolean {
-        return range.topicPartition.readIntervals { it.contains(range.range) }
-    }
+    operator fun contains(range: TopicPartitionOffsetRange): Boolean =
+        range.topicPartition.readIntervals { it.contains(range.range) }
 
     /** Whether this range value completely contains the given range.  */
-    fun contains(partition: TopicPartition, offset: Long, lastModified: Instant): Boolean {
-        return partition.readIntervals { it.contains(offset, lastModified) }
-    }
+    fun contains(partition: TopicPartition, offset: Long, lastModified: Instant): Boolean =
+        partition.readIntervals { it.contains(offset, lastModified) }
 
     /** Number of distinct offsets in given topic/partition.  */
-    fun size(topicPartition: TopicPartition): Int {
-        return topicPartition.readIntervals { it.size() }
-    }
+    fun size(topicPartition: TopicPartition): Int = topicPartition.readIntervals { it.size() }
 
-    fun remove(range: TopicPartitionOffsetRange) {
-        return range.topicPartition.modifyIntervals { it.remove(range.range) }
-    }
+    fun remove(range: TopicPartitionOffsetRange) =
+        range.topicPartition.modifyIntervals { it.remove(range.range) }
 
     fun withFactory(
         factory: (OffsetIntervals) -> FunctionalValue<OffsetIntervals>,
@@ -168,7 +154,11 @@ class OffsetRangeSet {
         factory,
     )
 
-    data class Range(val from: Long, val to: Long?, val lastProcessed: Instant = Instant.now()) {
+    data class Range(
+        val from: Long,
+        val to: Long?,
+        val lastProcessed: Instant = Instant.now(),
+    ) {
         @JsonIgnore
         val size: Long? = to?.let { it - from + 1 }
         fun ensureToOffset(): Range = if (to == null) copy(to = from) else this

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

@@ -4,6 +4,7 @@ import com.opencsv.CSVReader
 import com.opencsv.CSVWriter
 import org.apache.avro.generic.GenericRecord
 import org.radarbase.output.compression.Compression
+import org.radarbase.output.util.Md5Hasher
 import org.radarbase.output.util.ResourceContext.Companion.resourceContext
 import org.radarbase.output.util.SuspendedCloseable.Companion.useSuspended
 import org.radarbase.output.util.TimeUtil.parseDate
@@ -16,6 +17,7 @@ import java.io.InputStream
 import java.io.Reader
 import java.io.Writer
 import java.nio.file.Path
+import kotlin.collections.HashSet
 import kotlin.io.path.inputStream
 import kotlin.io.path.outputStream
 
@@ -38,10 +40,13 @@ class CsvAvroConverterFactory : RecordConverterFactory {
                 if (header == null) return false
                 val fields = fieldIndexes(header, distinctFields, ignoreFields)
                 var count = 0
+
+                val md5 = Md5Hasher()
+
                 val lineMap = buildMap {
                     lines.forEachIndexed { idx, line ->
                         count += 1
-                        put(ArrayWrapper(line.byIndex(fields)), idx)
+                        put(md5.hash(line.byIndex(fields)), idx)
                     }
                 }
 
@@ -208,21 +213,6 @@ class CsvAvroConverterFactory : RecordConverterFactory {
             return null
         }
 
-        private class ArrayWrapper<T>(val values: Array<T>) {
-            private val hashCode = values.contentHashCode()
-
-            override fun equals(other: Any?): Boolean {
-                if (this === other) return true
-                if (javaClass != other?.javaClass) return false
-
-                other as ArrayWrapper<*>
-
-                return values.contentEquals(other.values)
-            }
-
-            override fun hashCode(): Int = hashCode
-        }
-
         @Throws(IndexOutOfBoundsException::class)
         inline fun <reified T> Array<T>.byIndex(
             indexes: IntArray?,

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

@@ -0,0 +1,21 @@
+package org.radarbase.output.util
+
+import java.security.MessageDigest
+import java.util.*
+
+class Md5Hasher {
+    private val md5 = MessageDigest.getInstance("MD5")
+    private val base64 = Base64.getEncoder().withoutPadding()
+
+    fun hash(value: Array<String>): String {
+        value.forEachIndexed { i, field ->
+            if (i > 0) {
+                md5.update(','.code.toByte())
+            }
+            md5.update(field.toByteArray())
+        }
+        val hash = base64.encodeToString(md5.digest())
+        md5.reset()
+        return hash
+    }
+}

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

@@ -73,7 +73,7 @@ class FileCache(
     }
 
     suspend fun initialize(record: GenericRecord) {
-        val fileIsNew = targetStorage.status(path)?.takeIf { it.size > 0L } == null
+        var fileIsNew = targetStorage.status(path)?.takeIf { it.size > 0L } == null
 
         var outStream = compression.compress(fileName, tmpPath.outputStream().buffered())
 
@@ -82,13 +82,19 @@ class FileCache(
             inputStream = ByteArrayInputStream(ByteArray(0))
         } else {
             inputStream = time("write.copyOriginal") {
-                if (!copy(path, outStream, compression)) {
-                    // restart output buffer
-                    outStream.close()
-                    // clear output file
-                    outStream = compression.compress(fileName, tmpPath.outputStream().buffered())
+                try {
+                    if (!copy(path, outStream, compression)) {
+                        // restart output buffer
+                        outStream.close()
+                        // clear output file
+                        outStream =
+                            compression.compress(fileName, tmpPath.outputStream().buffered())
+                    }
+                    compression.decompress(targetStorage.newInputStream(path))
+                } catch (ex: FileNotFoundException) {
+                    fileIsNew = true
+                    ByteArrayInputStream(ByteArray(0))
                 }
-                compression.decompress(targetStorage.newInputStream(path))
             }
         }
 
@@ -191,6 +197,8 @@ class FileCache(
                     true
                 }
             }
+        } catch (ex: FileNotFoundException) {
+            throw ex
         } catch (ex: IOException) {
             var corruptPath: Path? = null
             var suffix = ""
@@ -205,17 +213,17 @@ class FileCache(
             }
             if (corruptPath != null) {
                 logger.error(
-                    "Original file {} could not be read: {}." + " Moved to {}.",
+                    "Original file {} could not be read: {}. Moved to {}.",
                     source,
-                    ex,
+                    ex.toString(),
                     corruptPath,
                 )
                 targetStorage.move(source, corruptPath)
             } else {
                 logger.error(
-                    "Original file {} could not be read: {}." + " Too many corrupt backups stored, removing file.",
+                    "Original file {} could not be read: {}. Too many corrupt backups stored, removing file.",
                     source,
-                    ex,
+                    ex.toString(),
                 )
             }
             false