Merge pull request #80 from RADAR-base/release-1.1.2

Release 1.1.2

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.1-hdfs -n hdfs-namenode -o /output /myTopic
+docker run --rm -t --network hadoop -v "$PWD/output:/output" radarbase/radar-output-restructure:1.1.2-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.1.tar.gz -C /usr/local --strip-components=1
+sudo tar -xzf build/distributions/radar-output-restructure-1.1.2.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.1'
+version '1.1.2'
 mainClassName = 'org.radarbase.output.Application'
 
 sourceCompatibility = '1.8'

restructure.yml

@@ -7,25 +7,49 @@ service:
 # Source data resource
 # @since: 0.7.0
 source:
-  type: s3  # hdfs or s3
+  type: s3  # hdfs, azure or s3
   s3:
     endpoint: http://localhost:9000  # using AWS S3 endpoint is also possible.
     bucket: radar
     accessToken: minioadmin
     secretKey: minioadmin
+  azure:
+    endpoint: https://MyBlobStorageAccount.blob.core.windows.net
+    # when using personal login
+    #username: User
+    #password: Password
+    # when using shared access tokens
+    #accountName: MyBlobStorageAccount
+    #accountKey: MyLongToken
+    # when using a specially made SAS token
+    #sasToken: MyLongToken
+    # if no credentials are supplied, this only works with a publicly writable blob storage
+    container: MySourceContainer
   # only actually needed if source type is hdfs
   hdfs:
     nameNodes: [hdfs-namenode-1, hdfs-namenode-2]
 
 # Target data resource
 # @since: 0.7.0
 target:
-  type: s3  # s3 or local
+  type: s3  # s3, azure or local
   s3:
     endpoint: http://localhost:9000
     bucket: out
     accessToken: minioadmin
     secretKey: minioadmin
+  azure:
+    endpoint: https://MyBlobStorageAccount.blob.core.windows.net
+    # when using personal login
+    #username: User
+    #password: Password
+    # when using shared access tokens
+    #accountName: MyBlobStorageAccount
+    #accountKey: MyLongToken
+    # when using a specially made SAS token
+    #sasToken: MyLongToken
+    # if no credentials are supplied, this only works with a publicly writable blob storage
+    container: MyTargetContainer
   # only actually needed if target type is local
   local:
     userId: 1000  # write as regular user, use -1 to use current user (default).
@@ -102,7 +126,7 @@ paths:
   # Output directory in target
   output: /output
   # Output path construction factory
-  factory: org.radarbase.output.path.MonthlyObservationKeyPathFactory
+  #factory: org.radarbase.output.path.MonthlyObservationKeyPathFactory
   # Additional properties
   # properties: {}
 

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

@@ -1,7 +1,8 @@
 package org.radarbase.output.accounting
 
 import org.hamcrest.MatcherAssert.assertThat
-import org.hamcrest.Matchers.*
+import org.hamcrest.Matchers.not
+import org.hamcrest.Matchers.nullValue
 import org.junit.jupiter.api.AfterEach
 import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.Test

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

@@ -19,31 +19,29 @@ package org.radarbase.output
 import com.beust.jcommander.JCommander
 import com.beust.jcommander.ParameterException
 import org.radarbase.output.accounting.*
+import org.radarbase.output.cleaner.SourceDataCleaner
 import org.radarbase.output.compression.Compression
 import org.radarbase.output.config.CommandLineArgs
 import org.radarbase.output.config.RestructureConfig
 import org.radarbase.output.format.RecordConverterFactory
+import org.radarbase.output.path.RecordPathFactory
 import org.radarbase.output.source.SourceStorage
 import org.radarbase.output.source.SourceStorageFactory
-import org.radarbase.output.path.RecordPathFactory
 import org.radarbase.output.target.TargetStorage
 import org.radarbase.output.target.TargetStorageFactory
 import org.radarbase.output.util.ProgressBar.Companion.format
+import org.radarbase.output.util.TimeUtil.durationSince
 import org.radarbase.output.util.Timer
 import org.radarbase.output.worker.FileCacheStore
-import org.radarbase.output.cleaner.SourceDataCleaner
-import org.radarbase.output.util.TimeUtil.durationSince
 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.Duration
 import java.time.Instant
 import java.time.LocalDateTime
 import java.time.format.DateTimeFormatter
-import java.time.temporal.Temporal
 import java.util.concurrent.Executors
 import java.util.concurrent.TimeUnit
 import kotlin.system.exitProcess

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

@@ -16,18 +16,18 @@
 
 package org.radarbase.output
 
-import java.io.IOException
 import org.radarbase.output.accounting.Accountant
 import org.radarbase.output.accounting.OffsetPersistenceFactory
 import org.radarbase.output.accounting.RemoteLockManager
-import org.radarbase.output.config.RestructureConfig
 import org.radarbase.output.compression.Compression
+import org.radarbase.output.config.RestructureConfig
 import org.radarbase.output.format.RecordConverterFactory
-import org.radarbase.output.source.SourceStorage
 import org.radarbase.output.path.RecordPathFactory
+import org.radarbase.output.source.SourceStorage
 import org.radarbase.output.target.TargetStorage
 import org.radarbase.output.worker.FileCacheStore
 import redis.clients.jedis.JedisPool
+import java.io.IOException
 
 /** Factory for all factory classes and settings.  */
 interface FileStoreFactory {

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

@@ -4,7 +4,6 @@ import org.apache.hadoop.fs.FSDataOutputStream
 import org.apache.hadoop.fs.FileSystem
 import org.apache.hadoop.fs.Path
 import org.slf4j.LoggerFactory
-import java.lang.Exception
 
 class HdfsRemoteLockManager(
         private val fileSystem: FileSystem,

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

@@ -16,6 +16,7 @@
 
 package org.radarbase.output.accounting
 
+import com.fasterxml.jackson.annotation.JsonIgnore
 import org.radarbase.output.util.FunctionalValue
 import org.radarbase.output.util.LockedFunctionalValue
 import org.radarbase.output.util.ReadOnlyFunctionalValue
@@ -164,6 +165,7 @@ class OffsetRangeSet {
 
 
     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
         override fun toString() = "($from - $to, $lastProcessed)"

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

@@ -18,6 +18,7 @@ package org.radarbase.output.accounting
 
 import com.fasterxml.jackson.databind.ObjectReader
 import com.fasterxml.jackson.databind.ObjectWriter
+import com.fasterxml.jackson.databind.SerializationFeature
 import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule
 import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
 import org.radarbase.output.util.PostponedWriter
@@ -95,8 +96,10 @@ class OffsetRedisPersistence(
                 val ranges: List<OffsetRangeSet.Range>)
 
         private val logger = LoggerFactory.getLogger(OffsetRedisPersistence::class.java)
-        private val mapper = jacksonObjectMapper()
-                .registerModule(JavaTimeModule())
+        private val mapper = jacksonObjectMapper().apply {
+            registerModule(JavaTimeModule())
+            configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false)
+        }
         val redisOffsetWriter: ObjectWriter = mapper.writerFor(RedisOffsetRangeSet::class.java)
         val redisOffsetReader: ObjectReader = mapper.readerFor(RedisOffsetRangeSet::class.java)
     }

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

@@ -16,7 +16,7 @@
 
 package org.radarbase.output.accounting
 
-import java.util.Objects
+import java.util.*
 
 @Suppress("EqualsOrHashCode")
 data class TopicPartition(val topic: String, val partition: Int) : Comparable<TopicPartition> {