Merge pull request #524 from RADAR-base/coroutines

Coroutines

Author: blootsvoets

.github/workflows/scheduled_snyk.yaml

@@ -8,12 +8,14 @@ jobs:
     env:
       REPORT_FILE: test.json
     steps:
-      - uses: actions/checkout@master
+      - uses: actions/checkout@v3
       - uses: snyk/actions/setup@master
+        with:
+          snyk-version: v1.931.0
 
-      - uses: actions/setup-java@v2
+      - uses: actions/setup-java@v3
         with:
-          distribution: zulu
+          distribution: temurin
           java-version: 17
 
       - name: Setup Gradle
@@ -27,3 +29,11 @@ jobs:
           --configuration-matching='^runtimeClasspath$'
           --json-file-output=${{ env.REPORT_FILE }}
           --org=radar-base
+
+      - name: Report new vulnerabilities
+        uses: thehyve/report-vulnerability@master
+        with:
+          report-file: ${{ env.REPORT_FILE }}
+        env:
+          TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        if: ${{ failure() }}

.github/workflows/snyk.yaml

@@ -7,12 +7,14 @@ jobs:
   security:
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@master
+      - uses: actions/checkout@v3
       - uses: snyk/actions/setup@master
+        with:
+          snyk-version: v1.931.0
 
-      - uses: actions/setup-java@v2
+      - uses: actions/setup-java@v3
         with:
-          distribution: zulu
+          distribution: temurin
           java-version: 17
 
       - name: Setup Gradle
@@ -23,7 +25,7 @@ jobs:
           SNYK_TOKEN: ${{ secrets.SNYK_TOKEN }}
         run: >
           snyk test
-          --severity-threshold=high
           --configuration-matching='^runtimeClasspath$'
           --fail-on=upgradable
           --org=radar-base
+          --severity-threshold=high

Dockerfile

@@ -32,7 +32,7 @@ MAINTAINER Joris Borgdorff <[email protected]>, Yatharth Ranjan<yatharth.ranjan@k
 LABEL description="RADAR-base output data restructuring"
 
 ENV RADAR_OUTPUT_RESTRUCTURE_OPTS="" \
-  JAVA_OPTS="-Djava.security.egd=file:/dev/./urandom -XX:+UseG1GC -XX:MaxHeapFreeRatio=10 -XX:MinHeapFreeRatio=10"
+  JAVA_OPTS="-XX:+UseG1GC -XX:MaxHeapFreeRatio=10 -XX:MinHeapFreeRatio=10"
 
 COPY --from=builder /code/build/third-party/* /usr/lib/
 COPY --from=builder /code/build/scripts/* /usr/bin/

build.gradle.kts

@@ -50,6 +50,8 @@ dependencies {
     runtimeOnly("org.xerial.snappy:snappy-java:$snappyVersion")
 
     implementation(kotlin("reflect"))
+    val coroutinesVersion: String by project
+    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:$coroutinesVersion")
 
     val jacksonVersion: String by project
     api(platform("com.fasterxml.jackson:jackson-bom:$jacksonVersion"))
@@ -95,6 +97,7 @@ dependencies {
 
     val radarSchemasVersion: String by project
     testImplementation("org.radarbase:radar-schemas-commons:$radarSchemasVersion")
+    testImplementation("org.jetbrains.kotlinx:kotlinx-coroutines-test:$coroutinesVersion")
 
     val junitVersion: String by project
     testImplementation("org.junit.jupiter:junit-jupiter-api:$junitVersion")

gradle.properties

@@ -7,6 +7,7 @@ dependencyUpdateVersion=0.42.0
 nexusPublishVersion=1.1.0
 jsoupVersion=1.14.3
 
+coroutinesVersion=1.6.1
 avroVersion=1.11.0
 snappyVersion=1.1.8.4
 jacksonVersion=2.13.2.20220328

restructure.yml

@@ -1,6 +1,6 @@
 service:
   # Whether to run the application as a polling service.
-  enable: false
+  enable: true
   # Polling interval in seconds.
   interval: 30
 
@@ -118,7 +118,7 @@ cleaner:
   # Enable cleaning up old source files
   enable: true
   # Interval in seconds to clean data
-  interval: 1260  # 21 minutes
+  interval: 50  # 21 minutes
   # Number of days after which a source file is considered old
   age: 7
 

src/integrationTest/java/org/radarbase/output/RestructureS3IntegrationTest.kt

@@ -2,9 +2,12 @@ package org.radarbase.output
 
 import io.minio.*
 import io.minio.ObjectWriteArgs.MAX_PART_SIZE
+import kotlinx.coroutines.*
+import kotlinx.coroutines.test.runTest
 import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.Test
 import org.radarbase.output.config.*
+import org.radarbase.output.util.SuspendedCloseable.Companion.useSuspended
 import org.radarbase.output.util.Timer
 import org.radarbase.output.util.bucketBuild
 import org.radarbase.output.util.objectBuild
@@ -13,7 +16,7 @@ import java.nio.file.Paths
 
 class RestructureS3IntegrationTest {
     @Test
-    fun integration() {
+    fun integration() = runTest {
         Timer.isEnabled = true
         val sourceConfig = S3Config(
                 endpoint ="http://localhost:9000",
@@ -38,77 +41,109 @@ class RestructureS3IntegrationTest {
         }
 
         val resourceFiles = listOf(
-                "application_server_status/partition=1/application_server_status+1+0000000018+0000000020.avro",
-                "application_server_status/partition=1/application_server_status+1+0000000021.avro",
-                "android_phone_acceleration/partition=0/android_phone_acceleration+0+0003018784.avro"
+            "application_server_status/partition=1/application_server_status+1+0000000018+0000000020.avro",
+            "application_server_status/partition=1/application_server_status+1+0000000021.avro",
+            "android_phone_acceleration/partition=0/android_phone_acceleration+0+0003018784.avro",
         )
         val targetFiles = resourceFiles.map { Paths.get("in/$it") }
-        resourceFiles.forEachIndexed { i, resourceFile ->
-            javaClass.getResourceAsStream("/$resourceFile").use { statusFile ->
-                sourceClient.putObject(PutObjectArgs.Builder().objectBuild(sourceConfig.bucket, targetFiles[i]) {
-                    stream(statusFile, -1, MAX_PART_SIZE)
-                })
+        resourceFiles.mapIndexed { i, resourceFile ->
+            launch(Dispatchers.IO) {
+                this@RestructureS3IntegrationTest.javaClass.getResourceAsStream("/$resourceFile")
+                    .useSuspended { statusFile ->
+                        sourceClient.putObject(PutObjectArgs.Builder()
+                            .objectBuild(sourceConfig.bucket, targetFiles[i]) {
+                                stream(statusFile, -1, MAX_PART_SIZE)
+                            })
+                    }
             }
-        }
+        }.joinAll()
 
         application.start()
 
         val targetClient = targetConfig.createS3Client()
-        val files = targetClient.listObjects(ListObjectsArgs.Builder().bucketBuild(targetConfig.bucket) {
-                    prefix("output")
-                    recursive(true)
-                    useUrlEncodingType(false)
-                })
-                .map { it.get().objectName() }
-                .toList()
 
         application.redisHolder.execute { redis ->
-            assertEquals(1L, redis.del("offsets/application_server_status.json"))
-            assertEquals(1L, redis.del("offsets/android_phone_acceleration.json"))
+            launch { assertEquals(1L, redis.del("offsets/application_server_status.json")) }
+            launch { assertEquals(1L, redis.del("offsets/android_phone_acceleration.json")) }
         }
 
         val firstParticipantOutput = "output/STAGING_PROJECT/1543bc93-3c17-4381-89a5-c5d6272b827c/application_server_status"
         val secondParticipantOutput = "output/radar-test-root/4ab9b985-6eec-4e51-9a29-f4c571c89f99/android_phone_acceleration"
-        assertEquals(
-                listOf(
-                        "$firstParticipantOutput/20200128_1300.csv",
-                        "$firstParticipantOutput/20200128_1400.csv",
-                        "$firstParticipantOutput/schema-application_server_status.json",
-                        "$secondParticipantOutput/20200528_1000.csv",
-                        "$secondParticipantOutput/schema-android_phone_acceleration.json"),
-                files)
-
-        println(targetClient.getObject(GetObjectArgs.builder()
-                .bucketBuild(targetConfig.bucket) {
-                    `object`("$firstParticipantOutput/20200128_1300.csv")
-                }
-        ).readBytes().toString(UTF_8))
 
-        val csvContents = """
+        val files = coroutineScope {
+            launch(Dispatchers.IO) {
+                val csvContents = """
                 key.projectId,key.userId,key.sourceId,value.time,value.serverStatus,value.ipAddress
                 STAGING_PROJECT,1543bc93-3c17-4381-89a5-c5d6272b827c,99caf236-bbe6-4eed-9c63-fba77349821d,1.58021982003E9,CONNECTED,
                 STAGING_PROJECT,1543bc93-3c17-4381-89a5-c5d6272b827c,99caf236-bbe6-4eed-9c63-fba77349821d,1.58021982003E9,CONNECTED,
 
                 """.trimIndent()
-        assertEquals(csvContents, targetClient.getObject(GetObjectArgs.Builder()
-                .bucketBuild(targetConfig.bucket) {
-                    `object`("$firstParticipantOutput/20200128_1300.csv")
-                })
-                .readBytes()
-                .toString(UTF_8))
-
-        targetFiles.forEach {
-            sourceClient.removeObject(RemoveObjectArgs.Builder()
-                    .objectBuild(sourceConfig.bucket, it))
-        }
-        sourceClient.removeBucket(RemoveBucketArgs.Builder().bucketBuild(sourceConfig.bucket))
-        files.forEach {
-            targetClient.removeObject(RemoveObjectArgs.Builder().bucketBuild(targetConfig.bucket) {
-                `object`(it)
-            })
+
+                val targetContent = targetClient.getObject(GetObjectArgs.Builder()
+                    .bucketBuild(targetConfig.bucket) {
+                        `object`("$firstParticipantOutput/20200128_1300.csv")
+                    }).use { response ->
+                    response.readBytes()
+                }
+
+                assertEquals(csvContents, targetContent.toString(UTF_8))
+            }
+
+            withContext(Dispatchers.IO) {
+                targetClient.listObjects(ListObjectsArgs.Builder()
+                    .bucketBuild(targetConfig.bucket) {
+                        prefix("output")
+                        recursive(true)
+                        useUrlEncodingType(false)
+                    })
+                    .map { it.get().objectName() }
+                    .toHashSet()
+            }
         }
-        targetClient.removeBucket(RemoveBucketArgs.Builder().bucketBuild(targetConfig.bucket))
 
+        assertEquals(
+            hashSetOf(
+                "$firstParticipantOutput/20200128_1300.csv",
+                "$firstParticipantOutput/20200128_1400.csv",
+                "$firstParticipantOutput/schema-application_server_status.json",
+                "$secondParticipantOutput/20200528_1000.csv",
+                "$secondParticipantOutput/schema-android_phone_acceleration.json",
+            ),
+            files,
+        )
+
+        coroutineScope {
+            // delete source files
+            launch {
+                targetFiles.map {
+                    launch(Dispatchers.IO) {
+                        sourceClient.removeObject(RemoveObjectArgs.Builder()
+                            .objectBuild(sourceConfig.bucket, it))
+                    }
+                }.joinAll()
+
+                launch(Dispatchers.IO) {
+                    sourceClient.removeBucket(RemoveBucketArgs.Builder()
+                        .bucketBuild(sourceConfig.bucket))
+                }
+            }
+
+            // delete target files
+            launch {
+                files.map {
+                    launch(Dispatchers.IO) {
+                        targetClient.removeObject(RemoveObjectArgs.Builder()
+                            .bucketBuild(targetConfig.bucket) {
+                                `object`(it)
+                            })
+                    }
+                }.joinAll()
+                launch(Dispatchers.IO) {
+                    targetClient.removeBucket(RemoveBucketArgs.Builder()
+                        .bucketBuild(targetConfig.bucket))
+                }
+            }
+        }
         println(Timer)
     }
 }

src/integrationTest/java/org/radarbase/output/accounting/OffsetRangeRedisTest.kt

@@ -1,10 +1,12 @@
 package org.radarbase.output.accounting
 
+import kotlinx.coroutines.test.runTest
 import org.junit.jupiter.api.AfterEach
 import org.junit.jupiter.api.Assertions.*
 import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.Test
 import org.radarbase.output.accounting.OffsetRedisPersistence.Companion.redisOffsetReader
+import org.radarbase.output.util.SuspendedCloseable.Companion.useSuspended
 import redis.clients.jedis.JedisPool
 import java.io.IOException
 import java.nio.file.Path
@@ -27,16 +29,18 @@ class OffsetRangeRedisTest {
 
     @AfterEach
     fun tearDown() {
-        redisHolder.execute { it.del(testFile.toString()) }
+        runTest {
+            redisHolder.execute { it.del(testFile.toString()) }
+        }
     }
 
     @Test
     @Throws(IOException::class)
-    fun readEmpty() {
+    fun readEmpty() = runTest {
         assertNull(offsetPersistence.read(testFile))
 
         // will create on write
-        offsetPersistence.writer(testFile).close()
+        offsetPersistence.writer(this@runTest, testFile).closeAndJoin()
 
         assertEquals(true, offsetPersistence.read(testFile)?.isEmpty)
 
@@ -47,8 +51,8 @@ class OffsetRangeRedisTest {
 
     @Test
     @Throws(IOException::class)
-    fun write() {
-        offsetPersistence.writer(testFile).use { rangeFile ->
+    fun write() = runTest {
+        offsetPersistence.writer(this@runTest, testFile).useSuspended { rangeFile ->
             rangeFile.add(TopicPartitionOffsetRange.parseFilename("a+0+0+1", lastModified))
             rangeFile.add(TopicPartitionOffsetRange.parseFilename("a+0+1+2", lastModified))
         }
@@ -67,8 +71,8 @@ class OffsetRangeRedisTest {
 
     @Test
     @Throws(IOException::class)
-    fun cleanUp() {
-        offsetPersistence.writer(testFile).use { rangeFile ->
+    fun cleanUp() = runTest {
+        offsetPersistence.writer(this@runTest, testFile).useSuspended { rangeFile ->
             rangeFile.add(TopicPartitionOffsetRange.parseFilename("a+0+0+1", lastModified))
             rangeFile.add(TopicPartitionOffsetRange.parseFilename("a+0+1+2", lastModified))
             rangeFile.add(TopicPartitionOffsetRange.parseFilename("a+0+4+4", lastModified))