Make reading end offset configurable

Author: blootsvoets

restructure.yml

@@ -13,6 +13,9 @@ source:
     bucket: radar
     accessToken: minioadmin
     secretKey: minioadmin
+    # If true, try to read the metadata property "endOffset" to determine the
+    # final offset of an input object.
+    #endOffsetFromTags: false
   azure:
     endpoint: https://MyBlobStorageAccount.blob.core.windows.net
     # when using personal login
@@ -25,6 +28,9 @@ source:
     #sasToken: MyLongToken
     # if no credentials are supplied, this only works with a publicly writable blob storage
     container: MySourceContainer
+    # If true, try to read the metadata property "endOffset" to determine the
+    # final offset of an input object.
+    #endOffsetFromMetadata: false
   # only actually needed if source type is hdfs
   hdfs:
     nameNodes: [hdfs-namenode-1, hdfs-namenode-2]

src/main/java/org/radarbase/output/config/RestructureConfig.kt

@@ -318,20 +318,32 @@ data class S3Config(
         /** Secret key belonging to access token. */
         val secretKey: String,
         /** Bucket name. */
-        val bucket: String) {
-    fun createS3Client() = MinioClient.Builder()
+        val bucket: String,
+        /** If no endOffset is in the filename, read it from object tags. */
+        val endOffsetFromTags: Boolean = false
+) {
+    fun createS3Client(): MinioClient = MinioClient.Builder()
             .endpoint(endpoint)
             .credentials(accessToken, secretKey)
             .build()
 }
 
 data class AzureConfig(
+        /** URL to reach object store at. */
         val endpoint: String,
+        /** Name of the Azure Blob Storage container. */
         val container: String,
+        /** If no endOffset is in the filename, read it from object metadata. */
+        val endOffsetFromMetadata: Boolean = false,
+        /** Azure username. */
         val username: String?,
+        /** Azure password. */
         val password: String?,
+        /** Shared Azure Blob Storage account name. */
         val accountName: String?,
+        /** Shared Azure Blob Storage account key. */
         val accountKey: String?,
+        /** Azure SAS token for a configured service. */
         val sasToken: String?
 ) {
     fun createAzureClient(): BlobServiceClient = BlobServiceClientBuilder().apply {

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

@@ -3,6 +3,7 @@ package org.radarbase.output.source
 import com.azure.storage.blob.BlobServiceClient
 import org.apache.avro.file.SeekableFileInput
 import org.apache.avro.file.SeekableInput
+import org.radarbase.output.config.AzureConfig
 import org.radarbase.output.util.TemporaryDirectory
 import org.radarbase.output.util.toKey
 import java.nio.file.Files
@@ -11,10 +12,12 @@ import java.nio.file.Paths
 
 class AzureSourceStorage(
         client: BlobServiceClient,
-        container: String,
+        config: AzureConfig,
         private val tempPath: Path
 ): SourceStorage {
-    private val blobContainerClient = client.getBlobContainerClient(container)
+    private val blobContainerClient = client.getBlobContainerClient(config.container)
+    private val readOffsetFromMetadata = config.endOffsetFromMetadata
+
     private fun blobClient(path: Path) = blobContainerClient.getBlobClient(path.toKey())
 
     override fun list(path: Path): Sequence<SimpleFileStatus> = blobContainerClient.listBlobsByHierarchy("$path/")
@@ -25,7 +28,7 @@ class AzureSourceStorage(
     override fun createTopicFile(topic: String, status: SimpleFileStatus): TopicFile {
         var topicFile = super.createTopicFile(topic, status)
 
-        if (topicFile.range.range.to == null) {
+        if (readOffsetFromMetadata && topicFile.range.range.to == null) {
             try {
                 val endOffset = blobClient(topicFile.path).properties.metadata["endOffset"]?.toLongOrNull()
 
@@ -36,7 +39,7 @@ class AzureSourceStorage(
                             })
                 }
             } catch (ex: Exception) {
-                // never mind
+                // skip reading end offset
             }
         }
 

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

@@ -3,6 +3,7 @@ package org.radarbase.output.source
 import io.minio.*
 import org.apache.avro.file.SeekableFileInput
 import org.apache.avro.file.SeekableInput
+import org.radarbase.output.config.S3Config
 import org.radarbase.output.util.TemporaryDirectory
 import org.radarbase.output.util.bucketBuild
 import org.radarbase.output.util.objectBuild
@@ -14,10 +15,12 @@ import java.nio.file.Paths
 
 class S3SourceStorage(
         private val s3Client: MinioClient,
-        private val bucket: String,
+        config: S3Config,
         private val tempPath: Path
 ): SourceStorage {
     override val walker: SourceStorageWalker = GeneralSourceStorageWalker(this)
+    private val bucket = config.bucket
+    private val readEndOffset = config.endOffsetFromTags
 
     override fun list(path: Path): Sequence<SimpleFileStatus> = s3Client.listObjects(
             ListObjectsArgs.Builder().bucketBuild(bucket) {
@@ -34,14 +37,18 @@ class S3SourceStorage(
     override fun createTopicFile(topic: String, status: SimpleFileStatus): TopicFile {
         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 = topicFile.copy(
-                        range = topicFile.range.mapRange {
-                            it.copy(to = endOffset)
-                        })
+        if (readEndOffset && topicFile.range.range.to == null) {
+            try {
+                val tags = s3Client.getObjectTags(GetObjectTagsArgs.Builder().objectBuild(bucket, status.path))
+                val endOffset = tags.get()["endOffset"]?.toLongOrNull()
+                if (endOffset != null) {
+                    topicFile = topicFile.copy(
+                            range = topicFile.range.mapRange {
+                                it.copy(to = endOffset)
+                            })
+                }
+            } catch (ex: Exception) {
+                // skip reading end offset
             }
         }
 

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

@@ -20,7 +20,7 @@ class SourceStorageFactory(private val resourceConfig: ResourceConfig, private v
         ResourceType.S3 -> {
             val s3Config = requireNotNull(resourceConfig.s3)
             val minioClient = requireNotNull(s3SourceClient)
-            S3SourceStorage(minioClient, s3Config.bucket, tempPath)
+            S3SourceStorage(minioClient, s3Config, tempPath)
         }
         ResourceType.HDFS -> {
             val hdfsConfig = requireNotNull(resourceConfig.hdfs)
@@ -29,8 +29,8 @@ class SourceStorageFactory(private val resourceConfig: ResourceConfig, private v
         }
         ResourceType.AZURE -> {
             val azureClient = requireNotNull(azureSourceClient)
-            val config = requireNotNull(resourceConfig.azure)
-            AzureSourceStorage(azureClient, config.container, tempPath)
+            val azureConfig = requireNotNull(resourceConfig.azure)
+            AzureSourceStorage(azureClient, azureConfig, tempPath)
         }
         else -> throw IllegalStateException("Cannot create kafka storage for type ${resourceConfig.sourceType}")
     }