Simplify topicFile creation

Author: blootsvoets

src/main/java/org/radarbase/output/source/S3SourceStorage.kt

@@ -31,16 +31,17 @@ class S3SourceStorage(
             }
 
     override fun createTopicFile(topic: String, status: SimpleFileStatus): TopicFile {
-        val topicFile = super.createTopicFile(topic, status)
-        return if (topicFile.range.range.to == null) {
+        var topicFile = super.createTopicFile(topic, status)
+
+        if (topicFile.range.range.to == null) {
             val tags = s3Client.getObjectTags(GetObjectTagsArgs.Builder().objectBuild(bucket, status.path))
             val endOffset = tags.get()["endOffset"]?.toLongOrNull()
             if (endOffset != null) {
-                topicFile.copy(range = topicFile.range.mapRange { it.copy(to = endOffset) })
-            } else {
-                topicFile
+                topicFile = topicFile.copy(range = topicFile.range.mapRange { it.copy(to = endOffset) })
             }
-        } else topicFile
+        }
+
+        return topicFile
     }
 
     override fun delete(path: Path) {

src/main/java/org/radarbase/output/source/SourceStorage.kt

@@ -1,6 +1,7 @@
 package org.radarbase.output.source
 
 import org.apache.avro.file.SeekableInput
+import org.radarbase.output.accounting.TopicPartitionOffsetRange
 import java.io.Closeable
 import java.nio.file.Path
 import java.time.Instant
@@ -15,7 +16,9 @@ interface SourceStorage {
     /** Delete given file. Will not delete any directories. */
     fun delete(path: Path)
     fun createTopicFile(topic: String, status: SimpleFileStatus): TopicFile {
-        return TopicFile(topic, status.path, status.lastModified ?: Instant.now())
+        val lastModified = status.lastModified ?: Instant.now()
+        val range = TopicPartitionOffsetRange.parseFilename(status.path.fileName.toString(), lastModified)
+        return TopicFile(topic, status.path, lastModified, range)
     }
 
     /** Find records and topics. */