Merge pull request #82 from RADAR-base/release-1.1.3

Release 1.1.3

Author: blootsvoets

README.md

@@ -61,7 +61,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 hadoop -v "$PWD/output:/output" radarbase/radar-output-restructure:1.1.2-hdfs -n hdfs-namenode -o /output /myTopic
+docker run --rm -t --network hadoop -v "$PWD/output:/output" radarbase/radar-output-restructure:1.1.3-hdfs -n hdfs-namenode -o /output /myTopic
 ```
 if your docker cluster is running in the `hadoop` network and your output directory should be `./output`.
 
@@ -171,7 +171,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-1.1.2.tar.gz -C /usr/local --strip-components=1
+sudo tar -xzf build/distributions/radar-output-restructure-1.1.3.tar.gz -C /usr/local --strip-components=1
 ```
 
 Now the `radar-output-restructure` command should be available.

build.gradle

@@ -9,7 +9,7 @@ plugins {
 }
 
 group 'org.radarbase'
-version '1.1.2'
+version '1.1.3'
 mainClassName = 'org.radarbase.output.Application'
 
 sourceCompatibility = '1.8'
@@ -28,7 +28,7 @@ ext {
     jCommanderVersion = '1.78'
     almworksVersion = '1.1.1'
     junitVersion = '5.6.1'
-    minioVersion = '7.0.1'
+    minioVersion = '7.1.0'
     jedisVersion = '3.2.0'
     azureStorageVersion = '12.6.0'
 }
@@ -88,6 +88,11 @@ compileTestKotlin {
         jvmTarget = "1.8"
     }
 }
+compileIntegrationTestKotlin {
+    kotlinOptions {
+        jvmTarget = "1.8"
+    }
+}
 
 ext.sharedManifest = manifest {
     attributes(

restructure.yml

@@ -13,6 +13,9 @@ source:
     bucket: radar
     accessToken: minioadmin
     secretKey: minioadmin
+    # If true, try to read the metadata property "endOffset" to determine the
+    # final offset of an input object.
+    #endOffsetFromTags: false
   azure:
     endpoint: https://MyBlobStorageAccount.blob.core.windows.net
     # when using personal login
@@ -25,6 +28,9 @@ source:
     #sasToken: MyLongToken
     # if no credentials are supplied, this only works with a publicly writable blob storage
     container: MySourceContainer
+    # If true, try to read the metadata property "endOffset" to determine the
+    # final offset of an input object.
+    #endOffsetFromMetadata: false
   # only actually needed if source type is hdfs
   hdfs:
     nameNodes: [hdfs-namenode-1, hdfs-namenode-2]

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

@@ -1,11 +1,13 @@
 package org.radarbase.output
 
-import io.minio.PutObjectOptions
+import io.minio.*
 import io.minio.PutObjectOptions.MAX_PART_SIZE
 import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.Test
 import org.radarbase.output.config.*
 import org.radarbase.output.util.Timer
+import org.radarbase.output.util.bucketBuild
+import org.radarbase.output.util.objectBuild
 import java.nio.charset.StandardCharsets.UTF_8
 import java.nio.file.Paths
 
@@ -31,8 +33,8 @@ class RestructureS3IntegrationTest {
         )
         val application = Application(config)
         val sourceClient = sourceConfig.createS3Client()
-        if (!sourceClient.bucketExists(sourceConfig.bucket)) {
-            sourceClient.makeBucket(sourceConfig.bucket)
+        if (!sourceClient.bucketExists(BucketExistsArgs.Builder().bucketBuild(sourceConfig.bucket))) {
+            sourceClient.makeBucket(MakeBucketArgs.Builder().bucketBuild(sourceConfig.bucket))
         }
 
         val resourceFiles = listOf(
@@ -43,18 +45,24 @@ class RestructureS3IntegrationTest {
         val targetFiles = resourceFiles.map { Paths.get("in/$it") }
         resourceFiles.forEachIndexed { i, resourceFile ->
             javaClass.getResourceAsStream("/$resourceFile").use { statusFile ->
-                sourceClient.putObject(sourceConfig.bucket, targetFiles[i].toString(), statusFile, PutObjectOptions(-1, MAX_PART_SIZE))
+                sourceClient.putObject(PutObjectArgs.Builder().objectBuild(sourceConfig.bucket, targetFiles[i]) {
+                    stream(statusFile, -1, MAX_PART_SIZE)
+                })
             }
         }
 
         application.start()
 
         val targetClient = targetConfig.createS3Client()
-        val files = targetClient.listObjects(targetConfig.bucket, "output")
+        val files = targetClient.listObjects(ListObjectsArgs.Builder().bucketBuild(targetConfig.bucket) {
+                    prefix("output")
+                    recursive(true)
+                    useUrlEncodingType(false)
+                })
                 .map { it.get().objectName() }
                 .toList()
 
-        application.redisPool.resource.use { redis ->
+        application.redisHolder.execute { redis ->
             assertEquals(1L, redis.del("offsets/application_server_status.json"))
             assertEquals(1L, redis.del("offsets/android_phone_acceleration.json"))
         }
@@ -70,26 +78,36 @@ class RestructureS3IntegrationTest {
                         "$secondParticipantOutput/schema-android_phone_acceleration.json"),
                 files)
 
-        println(targetClient.getObject(targetConfig.bucket, "$firstParticipantOutput/20200128_1300.csv").readBytes().toString(UTF_8))
+        println(targetClient.getObject(GetObjectArgs.builder()
+                .bucketBuild(targetConfig.bucket) {
+                    `object`("$firstParticipantOutput/20200128_1300.csv")
+                }
+        ).readBytes().toString(UTF_8))
 
         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(targetConfig.bucket, "$firstParticipantOutput/20200128_1300.csv")
+        assertEquals(csvContents, targetClient.getObject(GetObjectArgs.Builder()
+                .bucketBuild(targetConfig.bucket) {
+                    `object`("$firstParticipantOutput/20200128_1300.csv")
+                })
                 .readBytes()
                 .toString(UTF_8))
 
         targetFiles.forEach {
-            sourceClient.removeObject(sourceConfig.bucket, it.toString())
+            sourceClient.removeObject(RemoveObjectArgs.Builder()
+                    .objectBuild(sourceConfig.bucket, it))
         }
-        sourceClient.removeBucket(sourceConfig.bucket)
+        sourceClient.removeBucket(RemoveBucketArgs.Builder().bucketBuild(sourceConfig.bucket))
         files.forEach {
-            targetClient.removeObject(targetConfig.bucket, it)
+            targetClient.removeObject(RemoveObjectArgs.Builder().bucketBuild(targetConfig.bucket) {
+                `object`(it)
+            })
         }
-        targetClient.removeBucket(targetConfig.bucket)
+        targetClient.removeBucket(RemoveBucketArgs.Builder().bucketBuild(targetConfig.bucket))
 
         println(Timer)
     }

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

@@ -13,21 +13,21 @@ import java.time.Instant
 
 class OffsetRangeRedisTest {
     private lateinit var testFile: Path
-    private lateinit var redisPool: JedisPool
+    private lateinit var redisHolder: RedisHolder
     private lateinit var offsetPersistence: OffsetPersistenceFactory
     private val lastModified = Instant.now()
 
     @BeforeEach
     @Throws(IOException::class)
     fun setUp() {
         testFile = Paths.get("test/topic")
-        redisPool = JedisPool()
-        offsetPersistence = OffsetRedisPersistence(redisPool)
+        redisHolder = RedisHolder(JedisPool())
+        offsetPersistence = OffsetRedisPersistence(redisHolder)
     }
 
     @AfterEach
     fun tearDown() {
-        redisPool.resource.use { it.del(testFile.toString()) }
+        redisHolder.execute { it.del(testFile.toString()) }
     }
 
     @Test
@@ -40,7 +40,7 @@ class OffsetRangeRedisTest {
 
         assertEquals(true, offsetPersistence.read(testFile)?.isEmpty)
 
-        redisPool.resource.use { it.del(testFile.toString()) }
+        redisHolder.execute { it.del(testFile.toString()) }
 
         assertNull(offsetPersistence.read(testFile))
     }
@@ -74,7 +74,7 @@ class OffsetRangeRedisTest {
             rangeFile.add(TopicPartitionOffsetRange.parseFilename("a+0+4+4", lastModified))
         }
 
-        redisPool.resource.use { redis ->
+        redisHolder.execute { redis ->
             val range = redisOffsetReader.readValue<OffsetRedisPersistence.Companion.RedisOffsetRangeSet>(redis.get(testFile.toString()))
             assertEquals(OffsetRedisPersistence.Companion.RedisOffsetRangeSet(listOf(
                     OffsetRedisPersistence.Companion.RedisOffsetIntervals("a", 0, listOf(

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

@@ -9,20 +9,20 @@ import org.junit.jupiter.api.Test
 import redis.clients.jedis.JedisPool
 
 internal class RedisRemoteLockManagerTest {
-    private lateinit var redisPool: JedisPool
+    private lateinit var redisHolder: RedisHolder
     private lateinit var lockManager1: RemoteLockManager
     private lateinit var lockManager2: RemoteLockManager
 
     @BeforeEach
     fun setUp() {
-        redisPool = JedisPool()
-        lockManager1 = RedisRemoteLockManager(redisPool, "locks")
-        lockManager2 = RedisRemoteLockManager(redisPool, "locks")
+        redisHolder = RedisHolder(JedisPool())
+        lockManager1 = RedisRemoteLockManager(redisHolder, "locks")
+        lockManager2 = RedisRemoteLockManager(redisHolder, "locks")
     }
 
     @AfterEach
     fun tearDown() {
-        redisPool.close()
+        redisHolder.close()
     }
 
     @Test