Merge pull request #87 from RADAR-base/release-1.1.5

blootsvoets · web-flow · commit 178ebf0030c7 · 2020-10-28T11:02:39.000+01:00
Release 1.1.5
diff --git a/Dockerfile b/Dockerfile
@@ -10,26 +10,22 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-FROM openjdk:14 AS builder
+FROM gradle:6.6.1-jdk11 AS builder
 
 RUN mkdir /code
 WORKDIR /code
 
-ENV GRADLE_OPTS -Dorg.gradle.daemon=false
-
-COPY ./gradle /code/gradle
-COPY ./gradlew /code/
-RUN ./gradlew --version
+ENV GRADLE_USER_HOME=/code/.gradlecache
 
 COPY ./build.gradle ./gradle.properties ./settings.gradle /code/
 
-RUN ./gradlew downloadDependencies copyDependencies startScripts
+RUN gradle downloadDependencies copyDependencies startScripts
 
 COPY ./src /code/src
 
-RUN ./gradlew jar
+RUN gradle jar
 
-FROM openjdk:14
+FROM openjdk:11-jre-slim
 
 MAINTAINER Joris Borgdorff <joris@thehyve.nl>, Yatharth Ranjan<yatharth.ranjan@kcl.ac.uk>
 
diff --git a/README.md b/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.3-hdfs -n hdfs-namenode -o /output /myTopic
+docker run --rm -t --network hadoop -v "$PWD/output:/output" radarbase/radar-output-restructure:1.1.5-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.3.tar.gz -C /usr/local --strip-components=1
+sudo tar -xzf build/distributions/radar-output-restructure-1.1.5.tar.gz -C /usr/local --strip-components=1
 ```
 
 Now the `radar-output-restructure` command should be available.
diff --git a/build.gradle b/build.gradle
@@ -4,12 +4,12 @@ plugins {
     id 'application'
     id 'com.jfrog.bintray' version '1.8.5'
     id 'maven-publish'
-    id 'org.jetbrains.kotlin.jvm' version '1.3.72'
-    id 'com.avast.gradle.docker-compose' version "0.10.10"
+    id 'org.jetbrains.kotlin.jvm'
+    id 'com.avast.gradle.docker-compose' version "0.13.4"
 }
 
 group 'org.radarbase'
-version '1.1.4.1'
+version '1.1.5'
 mainClassName = 'org.radarbase.output.Application'
 
 sourceCompatibility = '1.8'
@@ -22,16 +22,16 @@ ext {
     githubUrl = "https://github.com/${githubRepoName}"
     issueUrl = "${githubUrl}/issues"
 
-    avroVersion = '1.9.2'
-    jacksonVersion = '2.10.3'
-    hadoopVersion = '3.1.2'
+    avroVersion = '1.10.0'
+    jacksonVersion = '2.11.3'
+    hadoopVersion = '3.3.0'
     jCommanderVersion = '1.78'
-    almworksVersion = '1.1.1'
-    junitVersion = '5.6.1'
-    minioVersion = '7.1.0'
-    jedisVersion = '3.2.0'
+    almworksVersion = '1.1.2'
+    junitVersion = '5.6.2'
+    minioVersion = '7.1.4'
+    jedisVersion = '3.3.0'
     slf4jVersion = '1.7.30'
-    azureStorageVersion = '12.6.0'
+    azureStorageVersion = '12.8.0'
 }
 
 repositories {
@@ -69,7 +69,8 @@ dependencies {
     implementation group: 'org.apache.avro', name: 'avro-mapred', version: avroVersion
     implementation group: 'org.apache.hadoop', name: 'hadoop-common', version: hadoopVersion
 
-    implementation "org.jetbrains.kotlin:kotlin-stdlib-jdk8"
+    implementation "org.jetbrains.kotlin:kotlin-stdlib-jdk8:$kotlinVersion"
+    implementation "org.jetbrains.kotlin:kotlin-reflect:$kotlinVersion"
     implementation group: 'org.slf4j', name: 'slf4j-api', version: slf4jVersion
 
     runtimeOnly group: 'org.slf4j', name: 'slf4j-log4j12', version: slf4jVersion
@@ -163,13 +164,18 @@ tasks.withType(Tar){
 }
 
 task downloadDependencies {
-    description "Pre-downloads dependencies"
-    configurations.compileClasspath.files
-    configurations.runtimeClasspath.files
+    doLast {
+        description "Pre-downloads dependencies"
+        configurations.compileClasspath.files
+        configurations.runtimeClasspath.files
+    }
+    outputs.upToDateWhen { false }
 }
 
 task copyDependencies(type: Copy) {
-    from configurations.runtimeClasspath.files
+    from {
+        configurations.runtimeClasspath.files
+    }
     into "$buildDir/third-party/"
 }
 
@@ -255,5 +261,5 @@ bintray {
 }
 
 wrapper {
-    gradleVersion '6.3'
+    gradleVersion '6.6.1'
 }
diff --git a/gradle.properties b/gradle.properties
@@ -0,0 +1 @@
+kotlinVersion=1.4.0
diff --git a/gradle/wrapper/gradle-wrapper.jar b/gradle/wrapper/gradle-wrapper.jar
diff --git a/gradle/wrapper/gradle-wrapper.properties b/gradle/wrapper/gradle-wrapper.properties
@@ -1,5 +1,5 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-6.3-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-6.6.1-bin.zip
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists
diff --git a/gradlew b/gradlew
@@ -82,6 +82,7 @@ esac
 
 CLASSPATH=$APP_HOME/gradle/wrapper/gradle-wrapper.jar
 
+
 # Determine the Java command to use to start the JVM.
 if [ -n "$JAVA_HOME" ] ; then
     if [ -x "$JAVA_HOME/jre/sh/java" ] ; then
@@ -129,6 +130,7 @@ fi
 if [ "$cygwin" = "true" -o "$msys" = "true" ] ; then
     APP_HOME=`cygpath --path --mixed "$APP_HOME"`
     CLASSPATH=`cygpath --path --mixed "$CLASSPATH"`
+
     JAVACMD=`cygpath --unix "$JAVACMD"`
 
     # We build the pattern for arguments to be converted via cygpath
diff --git a/gradlew.bat b/gradlew.bat
@@ -40,7 +40,7 @@ if defined JAVA_HOME goto findJavaFromJavaHome
 
 set JAVA_EXE=java.exe
 %JAVA_EXE% -version >NUL 2>&1
-if "%ERRORLEVEL%" == "0" goto init
+if "%ERRORLEVEL%" == "0" goto execute
 
 echo.
 echo ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH.
@@ -54,7 +54,7 @@ goto fail
 set JAVA_HOME=%JAVA_HOME:"=%
 set JAVA_EXE=%JAVA_HOME%/bin/java.exe
 
-if exist "%JAVA_EXE%" goto init
+if exist "%JAVA_EXE%" goto execute
 
 echo.
 echo ERROR: JAVA_HOME is set to an invalid directory: %JAVA_HOME%
@@ -64,28 +64,14 @@ echo location of your Java installation.
 
 goto fail
 
-:init
-@rem Get command-line arguments, handling Windows variants
-
-if not "%OS%" == "Windows_NT" goto win9xME_args
-
-:win9xME_args
-@rem Slurp the command line arguments.
-set CMD_LINE_ARGS=
-set _SKIP=2
-
-:win9xME_args_slurp
-if "x%~1" == "x" goto execute
-
-set CMD_LINE_ARGS=%*
-
 :execute
 @rem Setup the command line
 
 set CLASSPATH=%APP_HOME%\gradle\wrapper\gradle-wrapper.jar
 
+
 @rem Execute Gradle
-"%JAVA_EXE%" %DEFAULT_JVM_OPTS% %JAVA_OPTS% %GRADLE_OPTS% "-Dorg.gradle.appname=%APP_BASE_NAME%" -classpath "%CLASSPATH%" org.gradle.wrapper.GradleWrapperMain %CMD_LINE_ARGS%
+"%JAVA_EXE%" %DEFAULT_JVM_OPTS% %JAVA_OPTS% %GRADLE_OPTS% "-Dorg.gradle.appname=%APP_BASE_NAME%" -classpath "%CLASSPATH%" org.gradle.wrapper.GradleWrapperMain %*
 
 :end
 @rem End local scope for the variables with windows NT shell
diff --git a/settings.gradle b/settings.gradle
@@ -1 +1,11 @@
+pluginManagement {
+    resolutionStrategy {
+        eachPlugin {
+            if ( requested.id.id == 'org.jetbrains.kotlin.jvm' ) {
+                useModule( "org.jetbrains.kotlin:kotlin-gradle-plugin:$kotlinVersion" )
+            }
+        }
+    }
+}
+
 rootProject.name = 'radar-output-restructure'
diff --git a/src/main/java/org/radarbase/output/format/CsvAvroConverterFactory.kt b/src/main/java/org/radarbase/output/format/CsvAvroConverterFactory.kt
@@ -19,20 +19,20 @@ class CsvAvroConverterFactory: RecordConverterFactory {
     override val formats: Collection<String> = setOf("csv")
 
     @Throws(IOException::class)
-    override fun deduplicate(fileName: String, source: Path, target: Path, compression: Compression, distinctFields: Set<String>, ignoreFields: Set<String>) {
+    override fun deduplicate(fileName: String, source: Path, target: Path, compression: Compression, distinctFields: Set<String>, ignoreFields: Set<String>): Boolean {
         val (header, lineIndexes) = Files.newInputStream(source).use { input ->
             processLines(input, compression) { header, lines ->
-                if (header == null) return
+                if (header == null) return false
                 val fields = fieldIndexes(header, distinctFields, ignoreFields)
                 var count = 0
                 val lineMap = lines
                         .onEach { count += 1 }
-                        .mapIndexed { i, line -> Pair(ArrayWrapper(line.byIndex(fields)), i) }
+                        .mapIndexed { idx, line -> Pair(ArrayWrapper(line.byIndex(fields)), idx) }
                         .toMap(HashMap())
 
                 if (lineMap.size == count) {
                     logger.debug("File {} is already deduplicated. Skipping.", fileName)
-                    return
+                    return false
                 }
 
                 Pair(header, lineMap.values
@@ -52,6 +52,7 @@ class CsvAvroConverterFactory: RecordConverterFactory {
                 })
             }
         }
+        return true
     }
 
     private fun writeLines(target: Path, fileName: String, compression: Compression, lines: Sequence<Array<String>>) {
diff --git a/src/main/java/org/radarbase/output/format/RecordConverterFactory.kt b/src/main/java/org/radarbase/output/format/RecordConverterFactory.kt
@@ -54,9 +54,14 @@ interface RecordConverterFactory : Format {
     val hasHeader: Boolean
         get() = false
 
+    /**
+     * Deduplicate [source] with [fileName] and store the result in [target]. Returns `false` if no
+     * deduplication was performed because it was not necessary. In that case, [target] is not
+     * created.
+     */
     @Throws(IOException::class)
     fun deduplicate(fileName: String, source: Path, target: Path,
-                    compression: Compression, distinctFields: Set<String> = emptySet(), ignoreFields: Set<String> = emptySet()) {
+                    compression: Compression, distinctFields: Set<String> = emptySet(), ignoreFields: Set<String> = emptySet()): Boolean {
         val withHeader = hasHeader
 
         val (header, lines) = Files.newInputStream(source).use {
@@ -74,6 +79,8 @@ interface RecordConverterFactory : Format {
             writer ->
             writeFile(writer, header, lines)
         } } } }
+
+        return true
     }
 
     fun readTimeSeconds(source: InputStream, compression: Compression): Pair<Array<String>?, List<Double>>?
diff --git a/src/main/java/org/radarbase/output/worker/FileCache.kt b/src/main/java/org/radarbase/output/worker/FileCache.kt
@@ -133,11 +133,18 @@ class FileCache(
             if (deduplicate.enable == true) {
                 time("close.deduplicate") {
                     val dedupTmp = tmpPath.resolveSibling("${tmpPath.fileName}.dedup")
-                    converterFactory.deduplicate(fileName, tmpPath, dedupTmp, compression, deduplicate.distinctFields ?: emptySet(), deduplicate.ignoreFields ?: emptySet())
-                    try {
-                        Files.move(dedupTmp, tmpPath, StandardCopyOption.REPLACE_EXISTING, StandardCopyOption.ATOMIC_MOVE)
-                    } catch (ex: AtomicMoveNotSupportedException) {
-                        Files.move(dedupTmp, tmpPath, StandardCopyOption.REPLACE_EXISTING)
+                    if (converterFactory.deduplicate(
+                                    fileName,
+                                    source = tmpPath,
+                                    target = dedupTmp,
+                                    compression = compression,
+                                    distinctFields = deduplicate.distinctFields ?: emptySet(),
+                                    ignoreFields = deduplicate.ignoreFields ?: emptySet())) {
+                        try {
+                            Files.move(dedupTmp, tmpPath, StandardCopyOption.REPLACE_EXISTING, StandardCopyOption.ATOMIC_MOVE)
+                        } catch (ex: AtomicMoveNotSupportedException) {
+                            Files.move(dedupTmp, tmpPath, StandardCopyOption.REPLACE_EXISTING)
+                        }
                     }
                 }
             }
diff --git a/src/main/java/org/radarbase/output/worker/FileCacheStore.kt b/src/main/java/org/radarbase/output/worker/FileCacheStore.kt
@@ -21,14 +21,12 @@ import org.apache.avro.generic.GenericRecord
 import org.radarbase.output.FileStoreFactory
 import org.radarbase.output.accounting.Accountant
 import org.radarbase.output.util.TemporaryDirectory
-import org.radarbase.output.util.ThrowingConsumer.tryCatch
 import org.radarbase.output.util.Timer.time
 import org.radarbase.output.worker.FileCacheStore.WriteResponse.*
 import org.slf4j.LoggerFactory
 import java.io.Closeable
 import java.io.Flushable
 import java.io.IOException
-import java.io.UncheckedIOException
 import java.nio.file.Files
 import java.nio.file.Path
 import java.util.*
@@ -86,13 +84,9 @@ constructor(private val factory: FileStoreFactory, private val accountant: Accou
         }
 
         return try {
-            if (fileCache.writeRecord(record, transaction)) {
-                if (existingCache != null) CACHE_AND_WRITE else NO_CACHE_AND_WRITE
-            } else {
-                // The file path was not in cache but the file exists and this write is
-                // unsuccessful because of different number of columns
-                if (existingCache != null) CACHE_AND_NO_WRITE else NO_CACHE_AND_NO_WRITE
-            }
+            WriteResponse.valueOf(
+                    isCacheHit = existingCache != null,
+                    isSuccessful = fileCache.writeRecord(record, transaction))
         } catch (ex: IOException) {
             logger.error("Failed to write record. Closing cache {}.", fileCache.path, ex)
             fileCache.markError()
@@ -141,7 +135,8 @@ constructor(private val factory: FileStoreFactory, private val accountant: Accou
     @Throws(IOException::class)
     override fun flush() {
         try {
-            allCaches { it.close() }
+            caches.values.parallelStream()
+                    .forEach(FileCache::close)
             accountant.flush()
         } finally {
             caches.clear()
@@ -154,16 +149,6 @@ constructor(private val factory: FileStoreFactory, private val accountant: Accou
         tmpDir.close()
     }
 
-    @Throws(IOException::class)
-    private fun allCaches(cacheHandler: (FileCache) -> Unit) {
-        try {
-            caches.values.parallelStream()
-                    .forEach(tryCatch(cacheHandler, "Failed to update caches."))
-        } catch (ex: UncheckedIOException) {
-            throw ex.cause ?: ex
-        }
-    }
-
     /**
      * Response codes for each write record case.
      */
@@ -180,7 +165,18 @@ constructor(private val factory: FileStoreFactory, private val accountant: Accou
         /** Cache miss and write was successful.  */
         NO_CACHE_AND_WRITE(false, true),
         /** Cache miss and write was unsuccessful because of a mismatch in number of columns.  */
-        NO_CACHE_AND_NO_WRITE(false, false)
+        NO_CACHE_AND_NO_WRITE(false, false);
+
+        companion object {
+            fun valueOf(isCacheHit: Boolean, isSuccessful: Boolean) = when {
+                isSuccessful && isCacheHit -> CACHE_AND_WRITE
+                isSuccessful -> NO_CACHE_AND_WRITE
+                // The file path was not in cache but the file exists and this write is
+                // unsuccessful because of different number of columns
+                isCacheHit -> CACHE_AND_NO_WRITE
+                else -> NO_CACHE_AND_NO_WRITE
+            }
+        }
     }
 
     companion object {
