Remove unused functionality

blootsvoets · blootsvoets · commit f09712c8a5f8 · 2022-09-01T12:29:47.000+02:00
diff --git a/src/main/java/org/radarbase/output/cleaner/TimestampExtractionCheck.kt b/src/main/java/org/radarbase/output/cleaner/TimestampExtractionCheck.kt
@@ -61,7 +61,7 @@ class TimestampExtractionCheck(
         var suffix = 0
 
         do {
-            val (path) = pathFactory.getRecordOrganization(
+            val path = pathFactory.getRecordPath(
                 topicFile.topic,
                 record,
                 suffix
diff --git a/src/main/java/org/radarbase/output/path/FixedPathFormatterPlugin.kt b/src/main/java/org/radarbase/output/path/FixedPathFormatterPlugin.kt
@@ -1,18 +1,50 @@
 package org.radarbase.output.path
 
 import org.radarbase.output.path.RecordPathFactory.Companion.sanitizeId
+import org.slf4j.LoggerFactory
+import java.time.ZoneOffset
+import java.time.format.DateTimeFormatter
 
 class FixedPathFormatterPlugin : PathFormatterPlugin() {
+    private lateinit var timeBinFormat: DateTimeFormatter
     override val name: String = "fixed"
 
     override val allowedFormats: String = allowedParamNames.joinToString(separator = ", ")
 
+    override fun init(properties: Map<String, String>) {
+        super.init(properties)
+        timeBinFormat = createTimeBinFormatter(properties["timeBinFormat"])
+    }
+
+    private fun createTimeBinFormatter(pattern: String?): DateTimeFormatter {
+        pattern ?: return HOURLY_TIME_BIN_FORMAT
+
+        return try {
+            DateTimeFormatter
+                .ofPattern(pattern)
+                .withZone(ZoneOffset.UTC)
+        } catch (ex: IllegalArgumentException) {
+            logger.error(
+                "Cannot use time bin format {}, using {} instead",
+                pattern,
+                HOURLY_TIME_BIN_FORMAT,
+                ex,
+            )
+            HOURLY_TIME_BIN_FORMAT
+        }
+    }
+
     override fun lookup(parameterContents: String): PathFormatParameters.() -> String = when (parameterContents) {
         "projectId" -> ({ sanitizeId(key.get("projectId"), "unknown-project") })
         "userId" -> ({ sanitizeId(key.get("userId"), "unknown-user") })
         "sourceId" -> ({ sanitizeId(key.get("sourceId"), "unknown-source") })
         "topic" -> ({ topic })
-        "filename" -> ({ timeBin + attempt.toAttemptSuffix() + extension })
+        "filename" -> (
+            {
+                val timeBin = sanitizeId(time?.let { timeBinFormat.format(it) }, "unknown-time")
+                timeBin + attempt.toAttemptSuffix() + extension
+            }
+        )
         "attempt" -> ({ attempt.toAttemptSuffix() })
         "extension" -> ({ extension })
         else -> throw IllegalArgumentException("Unknown path parameter $parameterContents")
@@ -31,6 +63,12 @@ class FixedPathFormatterPlugin : PathFormatterPlugin() {
             "extension",
         )
 
+        val HOURLY_TIME_BIN_FORMAT: DateTimeFormatter = DateTimeFormatter
+            .ofPattern("yyyyMMdd_HH'00'")
+            .withZone(ZoneOffset.UTC)
+
         private fun Int.toAttemptSuffix() = if (this == 0) "" else "_$this"
+
+        private val logger = LoggerFactory.getLogger(FixedPathFormatterPlugin::class.java)
     }
 }
diff --git a/src/main/java/org/radarbase/output/path/FormattedPathFactory.kt b/src/main/java/org/radarbase/output/path/FormattedPathFactory.kt
@@ -25,63 +25,46 @@ import kotlin.reflect.jvm.jvmName
 
 open class FormattedPathFactory : RecordPathFactory() {
     private lateinit var formatter: PathFormatter
-    private lateinit var properties: Map<String, String>
+    private lateinit var config: PathFormatterConfig
     private var topicFormatters: Map<String, PathFormatter> = emptyMap()
 
     override fun init(properties: Map<String, String>) {
         super.init(properties)
 
-        this.properties = DEFAULTS + properties
-        formatter = createFormatter(this.properties)
+        this.config = DEFAULTS.withValues(properties)
+        formatter = config.toPathFormatter()
         logger.info("Formatting path with {}", formatter)
     }
 
-    private fun instantiatePlugins(
-        pluginClassNames: String,
-        properties: Map<String, String>,
-    ): List<PathFormatterPlugin> = pluginClassNames
-        .trim()
-        .split("\\s+".toRegex())
-        .mapNotNull { it.toPathFormatterPlugin() }
-        .onEach { it.init(properties) }
-
     override fun addTopicConfiguration(topicConfig: Map<String, TopicConfig>) {
         topicFormatters = topicConfig
             .filter { (_, config) -> config.pathProperties.isNotEmpty() }
             .mapValues { (_, config) ->
-                createFormatter(properties + config.pathProperties)
+                this.config.withValues(config.pathProperties)
+                    .toPathFormatter()
             }
             .onEach { (topic, formatter) ->
                 logger.info("Formatting path of topic {} with {}", topic, formatter)
             }
     }
 
-    private fun createFormatter(properties: Map<String, String>): PathFormatter {
-        val format = checkNotNull(properties["format"])
-        val pluginClassNames = checkNotNull(properties["plugins"])
-        val plugins = instantiatePlugins(pluginClassNames, properties)
-
-        return PathFormatter(format, plugins)
-    }
-
     override fun getRelativePath(
         topic: String,
         key: GenericRecord,
         value: GenericRecord,
         time: Instant?,
         attempt: Int,
     ): Path = (topicFormatters[topic] ?: formatter)
-        .format(PathFormatParameters(topic, key, value, time, attempt, extension, this::getTimeBin))
-
-    override fun getCategory(
-        key: GenericRecord,
-        value: GenericRecord,
-    ): String = sanitizeId(key.get("sourceId"), "unknown-source")
+        .format(PathFormatParameters(topic, key, value, time, attempt, extension))
 
     companion object {
-        internal val DEFAULTS = mapOf(
-            "format" to "\${projectId}/\${userId}/\${topic}/\${filename}",
-            "plugins" to "fixed time key value",
+        private fun PathFormatterConfig.toPathFormatter(): PathFormatter {
+            return PathFormatter(format, createPlugins())
+        }
+
+        internal val DEFAULTS = PathFormatterConfig(
+            format = "\${projectId}/\${userId}/\${topic}/\${filename}",
+            pluginNames = "fixed time key value",
         )
         private val logger = LoggerFactory.getLogger(FormattedPathFactory::class.java)
 
@@ -107,5 +90,27 @@ open class FormattedPathFactory : RecordPathFactory() {
                 }
             }
         }
+
+        data class PathFormatterConfig(
+            val format: String,
+            val pluginNames: String,
+            val properties: Map<String, String> = mapOf(),
+        ) {
+            fun createPlugins(): List<PathFormatterPlugin> = pluginNames
+                .trim()
+                .split("\\s+".toRegex())
+                .mapNotNull { it.toPathFormatterPlugin() }
+                .onEach { it.init(properties) }
+
+            fun withValues(values: Map<String, String>): PathFormatterConfig {
+                val newProperties = HashMap(properties).apply {
+                    putAll(values)
+                }
+                val format = newProperties.remove("format") ?: this.format
+                val pluginNames = newProperties.remove("plugins") ?: this.pluginNames
+
+                return PathFormatterConfig(format, pluginNames, newProperties)
+            }
+        }
     }
 }
diff --git a/src/main/java/org/radarbase/output/path/ObservationKeyPathFactory.kt b/src/main/java/org/radarbase/output/path/ObservationKeyPathFactory.kt
@@ -16,32 +16,4 @@
 
 package org.radarbase.output.path
 
-import org.apache.avro.generic.GenericRecord
-import java.nio.file.Path
-import java.nio.file.Paths
-import java.time.Instant
-
-open class ObservationKeyPathFactory : RecordPathFactory() {
-    override fun getRelativePath(
-        topic: String,
-        key: GenericRecord,
-        value: GenericRecord,
-        time: Instant?,
-        attempt: Int,
-    ): Path {
-        val projectId = sanitizeId(key.get("projectId"), "unknown-project")
-        val userId = sanitizeId(key.get("userId"), "unknown-user")
-
-        val attemptSuffix = if (attempt == 0) "" else "_$attempt"
-        val outputFileName = getTimeBin(time) + attemptSuffix + extension
-
-        return Paths.get(projectId, userId, topic, outputFileName)
-    }
-
-    override fun getCategory(
-        key: GenericRecord,
-        value: GenericRecord,
-    ): String {
-        return sanitizeId(key.get("sourceId"), "unknown-source")
-    }
-}
+open class ObservationKeyPathFactory : FormattedPathFactory()
diff --git a/src/main/java/org/radarbase/output/path/PathFormatParameters.kt b/src/main/java/org/radarbase/output/path/PathFormatParameters.kt
@@ -10,8 +10,4 @@ data class PathFormatParameters(
     val time: Instant?,
     val attempt: Int,
     val extension: String,
-    val computeTimeBin: (time: Instant?) -> String,
-) {
-    val timeBin: String
-        get() = computeTimeBin(time)
-}
+)
diff --git a/src/main/java/org/radarbase/output/path/RecordPathFactory.kt b/src/main/java/org/radarbase/output/path/RecordPathFactory.kt
@@ -26,35 +26,13 @@ import org.radarbase.output.util.TimeUtil
 import org.slf4j.LoggerFactory
 import java.nio.file.Path
 import java.time.Instant
-import java.time.ZoneOffset.UTC
-import java.time.format.DateTimeFormatter
 import java.util.regex.Pattern
 
 abstract class RecordPathFactory : Plugin {
     lateinit var root: Path
     lateinit var extension: String
     lateinit var fileStoreFactory: FileStoreFactory
 
-    protected open var timeBinFormat: DateTimeFormatter = HOURLY_TIME_BIN_FORMAT
-
-    override fun init(properties: Map<String, String>) {
-        super.init(properties)
-        properties["timeBinFormat"]?.let {
-            try {
-                timeBinFormat = DateTimeFormatter
-                    .ofPattern(it)
-                    .withZone(UTC)
-            } catch (ex: IllegalArgumentException) {
-                logger.error(
-                    "Cannot use time bin format {}, using {} instead",
-                    it,
-                    timeBinFormat,
-                    ex,
-                )
-            }
-        }
-    }
-
     /**
      * Get the organization of given record in given topic.
      * @param topic Kafka topic name
@@ -63,11 +41,11 @@ abstract class RecordPathFactory : Plugin {
      * paths already existed and are incompatible.
      * @return organization of given record
      */
-    open fun getRecordOrganization(
+    open fun getRecordPath(
         topic: String,
         record: GenericRecord,
         attempt: Int,
-    ): RecordOrganization {
+    ): Path {
         val keyField = record.get("key")
         val valueField = record.get("value") as? GenericRecord
 
@@ -79,19 +57,17 @@ abstract class RecordPathFactory : Plugin {
         val keyRecord: GenericRecord = if (keyField is GenericRecord) {
             keyField
         } else {
-            GenericRecordBuilder(observationKeySchema)
-                .set("projectId", valueField.getOrNull("projectId"))
-                .set("userId", keyField.toString())
-                .set("sourceId", valueField.getOrNull("sourceId") ?: "unknown")
-                .build()
+            GenericRecordBuilder(observationKeySchema).apply {
+                set("projectId", valueField.getOrNull("projectId"))
+                set("userId", keyField.toString())
+                set("sourceId", valueField.getOrNull("sourceId") ?: "unknown")
+            }.build()
         }
 
         val time = TimeUtil.getDate(keyRecord, valueField)
 
         val relativePath = getRelativePath(topic, keyRecord, valueField, time, attempt)
-        val outputPath = root.resolve(relativePath)
-        val category = getCategory(keyRecord, valueField)
-        return RecordOrganization(outputPath, category, time)
+        return root.resolve(relativePath)
     }
 
     /**
@@ -112,38 +88,10 @@ abstract class RecordPathFactory : Plugin {
         attempt: Int,
     ): Path
 
-    /**
-     * Get the category of a record, representing a partitioning for a given topic and user.
-     * @param key record key
-     * @param value record value
-     * @return category name.
-     */
-    abstract fun getCategory(key: GenericRecord, value: GenericRecord): String
-
-    open fun getTimeBin(time: Instant?): String = time
-        ?.let { timeBinFormat.format(time) }
-        ?: "unknown_date"
-
-    /**
-     * Organization of a record.
-     */
-    data class RecordOrganization(
-        /** Path that the record should be stored in. */
-        val path: Path,
-        /** Category or partition that the record belongs to. */
-        val category: String,
-        /** Time contained in the record, if any. */
-        val time: Instant?,
-    )
-
     companion object {
         private val logger = LoggerFactory.getLogger(RecordPathFactory::class.java)
         private val ILLEGAL_CHARACTER_PATTERN = Pattern.compile("[^a-zA-Z0-9_-]+")
 
-        val HOURLY_TIME_BIN_FORMAT: DateTimeFormatter = DateTimeFormatter
-            .ofPattern("yyyyMMdd_HH'00'")
-            .withZone(UTC)
-
         fun sanitizeId(id: Any?, defaultValue: String): String = id
             ?.let { ILLEGAL_CHARACTER_PATTERN.matcher(it.toString()).replaceAll("") }
             ?.takeIf { it.isNotEmpty() }
diff --git a/src/main/java/org/radarbase/output/worker/RestructureWorker.kt b/src/main/java/org/radarbase/output/worker/RestructureWorker.kt
@@ -167,7 +167,7 @@ internal class RestructureWorker(
     ) {
         var currentSuffix = 0
         do {
-            val (path) = pathFactory.getRecordOrganization(
+            val path = pathFactory.getRecordPath(
                 transaction.topicPartition.topic,
                 record,
                 attempt = currentSuffix
diff --git a/src/test/java/org/radarbase/output/RadarHdfsRestructureTest.kt b/src/test/java/org/radarbase/output/RadarHdfsRestructureTest.kt
diff --git a/src/test/java/org/radarbase/output/path/PathFormatterTest.kt b/src/test/java/org/radarbase/output/path/PathFormatterTest.kt
