After review

snowplow · Jul 26, 2023 · df6372d · df6372d
1 parent c58edc2
commit df6372d
Show file tree

Hide file tree

Showing 5 changed files with 13 additions and 11 deletions.
diff --git a/modules/aws/src/main/scala/com/snowplowanalytics/snowplow/rdbloader/aws/S3.scala b/modules/aws/src/main/scala/com/snowplowanalytics/snowplow/rdbloader/aws/S3.scala
@@ -23,7 +23,7 @@ import software.amazon.awssdk.services.s3.S3AsyncClient
 import com.snowplowanalytics.snowplow.rdbloader.common.cloud.BlobStorage
 import com.snowplowanalytics.snowplow.rdbloader.common.cloud.BlobStorage.Key
 
-import java.nio.charset.StandardCharsets.UTF_8
+import java.nio.charset.StandardCharsets
 
 object S3 {
 
@@ -58,7 +58,7 @@ object S3 {
                          def get(path: Key): F[Either[Throwable, String]] =
                            getBytes(path).compile
                              .to(Array)
-                             .map(array => new String(array, UTF_8))
+                             .map(array => new String(array, StandardCharsets.UTF_8))
                              .attempt
 
                          def list(folder: BlobStorage.Folder, recursive: Boolean): Stream[F, BlobStorage.BlobObject] = {

diff --git a/...zure/src/main/scala/com/snowplowanalytics/snowplow/rdbloader/azure/AzureBlobStorage.scala b/...zure/src/main/scala/com/snowplowanalytics/snowplow/rdbloader/azure/AzureBlobStorage.scala
@@ -20,6 +20,7 @@ import com.snowplowanalytics.snowplow.rdbloader.common.cloud.BlobStorage
 import com.snowplowanalytics.snowplow.rdbloader.common.cloud.BlobStorage.{Folder, Key}
 import fs2.{Pipe, Stream}
 import java.net.URI
+import java.nio.charset.StandardCharsets
 
 class AzureBlobStorage[F[_]: Async] private (store: AzureStore[F], configuredPath: AzureBlobStorage.PathParts) extends BlobStorage[F] {
 
@@ -44,7 +45,7 @@ class AzureBlobStorage[F[_]: Async] private (store: AzureStore[F], configuredPat
   override def get(key: Key): F[Either[Throwable, String]] =
     getBytes(key).compile
       .to(Array)
-      .map(array => new String(array))
+      .map(array => new String(array, StandardCharsets.UTF_8))
       .attempt
 
   override def getBytes(key: Key): Stream[F, Byte] =

diff --git a/modules/gcp/src/main/scala/com/snowplowanalytics/snowplow/rdbloader/gcp/GCS.scala b/modules/gcp/src/main/scala/com/snowplowanalytics/snowplow/rdbloader/gcp/GCS.scala
@@ -22,6 +22,8 @@ import fs2.{Pipe, Stream}
 import com.snowplowanalytics.snowplow.rdbloader.common.cloud.BlobStorage
 import com.snowplowanalytics.snowplow.rdbloader.common.cloud.BlobStorage.{Folder, Key}
 
+import java.nio.charset.StandardCharsets
+
 object GCS {
 
   def blobStorage[F[_]: Async]: Resource[F, BlobStorage[F]] =
@@ -73,7 +75,7 @@ object GCS {
                          override def get(key: Key): F[Either[Throwable, String]] =
                            getBytes(key).compile
                              .to(Array)
-                             .map(array => new String(array))
+                             .map(array => new String(array, StandardCharsets.UTF_8))
                              .attempt
 
                          override def keyExists(key: Key): F[Boolean] = {

diff --git a/...wplow/rdbloader/transformer/stream/kafka/experimental/AzureTransformerSpecification.scala b/...wplow/rdbloader/transformer/stream/kafka/experimental/AzureTransformerSpecification.scala
@@ -29,7 +29,7 @@ import java.util.UUID
  */
 // format: on
 trait AzureTransformerSpecification extends TransformerSpecification with AppDependencies.Provider {
-  skipAllIf(anyEnvironementVariableMissing())
+  skipAllIf(anyEnvironmentVariableMissing())
 
   override def createDependencies(): Resource[IO, AppDependencies] =
     for {
@@ -89,7 +89,7 @@ object AzureTransformerSpecification {
     outputHubKeyEnv
   )
 
-  def anyEnvironementVariableMissing(): Boolean =
+  def anyEnvironmentVariableMissing(): Boolean =
     requiredEnvironmentVariables.exists(varName => System.getenv(varName) == null)
 
 }
diff --git a/...s/snowplow/rdbloader/transformer/stream/kafka/experimental/TransformerSpecification.scala b/...s/snowplow/rdbloader/transformer/stream/kafka/experimental/TransformerSpecification.scala
@@ -229,11 +229,10 @@ abstract class TransformerSpecification extends Specification with AppDependenci
       .map(value => parseShreddingCompleteMessage(value.right.get))
 
   private def parseShreddingCompleteMessage(message: String): LoaderMessage.ShreddingComplete =
-    LoaderMessage
-      .fromString(message)
-      .right
-      .get
-      .asInstanceOf[LoaderMessage.ShreddingComplete]
+    LoaderMessage.fromString(message) match {
+      case Right(parsedMessage: LoaderMessage.ShreddingComplete) => parsedMessage
+      case other => throw new IllegalStateException(s"Provided message is not a valid shredding complete message - $other")
+    }
 
 }