Merge pull request #83 from RADAR-base/release-1.1.4

Release 1.1.4

Author: blootsvoets

build.gradle

@@ -9,7 +9,7 @@ plugins {
 }
 
 group 'org.radarbase'
-version '1.1.3'
+version '1.1.4'
 mainClassName = 'org.radarbase.output.Application'
 
 sourceCompatibility = '1.8'
@@ -30,6 +30,7 @@ ext {
     junitVersion = '5.6.1'
     minioVersion = '7.1.0'
     jedisVersion = '3.2.0'
+    slf4jVersion = '1.7.30'
     azureStorageVersion = '12.6.0'
 }
 
@@ -69,7 +70,9 @@ dependencies {
     implementation group: 'org.apache.hadoop', name: 'hadoop-common', version: hadoopVersion
 
     implementation "org.jetbrains.kotlin:kotlin-stdlib-jdk8"
+    implementation group: 'org.slf4j', name: 'slf4j-api', version: slf4jVersion
 
+    runtimeOnly group: 'org.slf4j', name: 'slf4j-log4j12', version: slf4jVersion
     runtimeOnly group: 'org.apache.hadoop', name: 'hadoop-hdfs-client', version: hadoopVersion
     testImplementation group: 'org.junit.jupiter', name: 'junit-jupiter-api', version: junitVersion
     testImplementation group: 'org.junit.jupiter', name: 'junit-jupiter-params', version: junitVersion

src/main/java/org/radarbase/output/cleaner/TimestampExtractionCheck.kt

@@ -28,10 +28,12 @@ class TimestampExtractionCheck(
                 return false
             }
             RestructureWorker.extractRecords(input) { records ->
-                records.all { record ->
-                    cachedRecords += 1
-                    containsRecord(file.topic, record)
-                }
+                records
+                        .mapIndexed { i, record ->
+                            cachedRecords += 1L
+                            containsRecord(file, file.range.range.from + i.toLong(), record)
+                        }
+                        .all { it }
             }
         }
         if (cachedRecords > batchSize) {
@@ -46,19 +48,28 @@ class TimestampExtractionCheck(
     }
 
 
-    private fun containsRecord(topic: String, record: GenericRecord): Boolean {
+    private fun containsRecord(topicFile: TopicFile, offset: Long, record: GenericRecord): Boolean {
         var suffix = 0
 
         do {
             val (path) = pathFactory.getRecordOrganization(
-                    topic, record, suffix)
+                    topicFile.topic, record, suffix)
 
             try {
                 when (cacheStore.contains(path, record)) {
-                    TimestampFileCacheStore.FindResult.FILE_NOT_FOUND -> return false
-                    TimestampFileCacheStore.FindResult.NOT_FOUND -> return false
+                    TimestampFileCacheStore.FindResult.FILE_NOT_FOUND -> {
+                        logger.warn("Target {} for record of {} (offset {}) has not been created yet.", path, topicFile.path, offset)
+                        return false
+                    }
+                    TimestampFileCacheStore.FindResult.NOT_FOUND -> {
+                        logger.warn("Target {} does not contain record of {} (offset {})", path, topicFile.path, offset)
+                        return false
+                    }
                     TimestampFileCacheStore.FindResult.FOUND -> return true
-                    TimestampFileCacheStore.FindResult.BAD_SCHEMA -> suffix += 1  // continue next suffix
+                    TimestampFileCacheStore.FindResult.BAD_SCHEMA -> {
+                        logger.debug("Schema of {} does not match schema of {} (offset {})", path, topicFile.path, offset)
+                        suffix += 1 // continue next suffix
+                    }
                 }
             } catch (ex: IOException) {
                 logger.error("Failed to read target file {} for checking data integrity", path, ex)

src/main/java/org/radarbase/output/cleaner/TimestampFileCacheStore.kt

@@ -61,19 +61,13 @@ class TimestampFileCacheStore(private val factory: FileStoreFactory) {
                     }
 
             time("cleaner.contains") {
-                if (fileCache.contains(record)) FindResult.FOUND else {
-                    logger.warn("Target {} does not contain record {}", path, record)
-                    FindResult.NOT_FOUND
-                }
+                if (fileCache.contains(record)) FindResult.FOUND else FindResult.NOT_FOUND
             }
         } catch (ex: FileNotFoundException) {
-            logger.warn("Target {} for {} has not been created yet.", path, record)
             FindResult.FILE_NOT_FOUND
         } catch (ex: IllegalArgumentException) {
-            logger.warn("Schema of {} does not match schema of record {}: {}", path, record, ex.message)
             FindResult.BAD_SCHEMA
         } catch (ex: IndexOutOfBoundsException) {
-            logger.warn("Schema of {} does not match schema of record {} (wrong number of columns)", path, record)
             FindResult.BAD_SCHEMA
         }
     }