Feat/5.0 changes

kafka-connect-aws-s3/src/it/scala/io/lenses/streamreactor/connect/aws/s3/sink/S3AvroWriterManagerTest.scala

@@ -134,6 +134,69 @@ class S3AvroWriterManagerTest extends AnyFlatSpec with Matchers with S3ProxyCont
 
   }
 
+  "avro sink" should "write BigDecimal" in {
+    val sink = S3WriterManager.from(avroConfig)
+    val usersWithDecimal1 =
+      new Struct(UsersSchemaDecimal)
+        .put("name", "sam")
+        .put("title", "mr")
+        .put(
+          "salary",
+          BigDecimal("100.43").setScale(18).bigDecimal,
+        )
+    val writeRes1 = sink.write(
+      TopicPartitionOffset(Topic(TopicName), 1, Offset(1L)),
+      MessageDetail(
+        NullSinkData(None),
+        StructSinkData(usersWithDecimal1),
+        Map.empty,
+        None,
+        Topic(TopicName),
+        1,
+        Offset(1L),
+      ),
+    )
+
+    writeRes1.isRight should be(true)
+
+    val usersWithDecimal2 =
+      new Struct(UsersSchemaDecimal)
+        .put("name", "maria")
+        .put("title", "ms")
+        .put(
+          "salary",
+          BigDecimal("100.43").setScale(18).bigDecimal,
+        )
+
+    val writeRes2 = sink.write(
+      TopicPartitionOffset(Topic(TopicName), 1, Offset(2L)),
+      MessageDetail(
+        NullSinkData(None),
+        StructSinkData(usersWithDecimal2),
+        Map.empty,
+        None,
+        Topic(TopicName),
+        1,
+        Offset(2L),
+      ),
+    )
+    writeRes2.isRight should be(true)
+    sink.close()
+
+    listBucketPath(BucketName, "streamReactorBackups/myTopic/1/").size should be(1)
+
+    val byteArray = remoteFileAsBytes(BucketName, "streamReactorBackups/myTopic/1/2.avro")
+    val genericRecords: List[GenericRecord] = avroFormatReader.read(byteArray)
+    genericRecords.size should be(2)
+
+    genericRecords(0).get("name").toString should be("sam")
+    val bb1 = genericRecords(0).get("salary").asInstanceOf[ByteBuffer]
+    Decimal.toLogical(Decimal.builder(18).optional().build(), bb1.array()) should be(
+      BigDecimal(100.43).setScale(18).bigDecimal,
+    )
+
+  }
+
   "avro sink" should "write start a new file in case of schema change" in {
 
     val secondSchema: Schema = SchemaBuilder.struct()

kafka-connect-aws-s3/src/it/scala/io/lenses/streamreactor/connect/aws/s3/sink/S3JsonWriterManagerTest.scala

@@ -176,4 +176,74 @@ class S3JsonWriterManagerTest extends AnyFlatSpec with Matchers with S3ProxyCont
     )
   }
 
+  "json sink" should "write bigdecimal to json" in {
+
+    val bucketAndPrefix = S3Location(BucketName, PathPrefix.some)
+    val config = S3SinkConfig(
+      S3Config(
+        None,
+        Some(Identity),
+        Some(Credential),
+        AuthMode.Credentials,
+      ),
+      bucketOptions = Seq(
+        SinkBucketOptions(
+          TopicName.some,
+          bucketAndPrefix,
+          commitPolicy    = CommitPolicy(Count(1)),
+          formatSelection = JsonFormatSelection,
+          keyNamer = new S3KeyNamer(
+            AvroFormatSelection,
+            defaultPartitionSelection(Values),
+            new OffsetS3FileNamer(
+              identity[String],
+              JsonFormatSelection.extension,
+            ),
+            new PaddingService(Map[String, PaddingStrategy](
+              "partition" -> NoOpPaddingStrategy,
+              "offset"    -> LeftPadPaddingStrategy(12, 0),
+            )),
+          ),
+          localStagingArea   = LocalStagingArea(localRoot),
+          partitionSelection = defaultPartitionSelection(Values),
+          dataStorage        = DataStorageSettings.disabled,
+        ),
+      ),
+      offsetSeekerOptions = OffsetSeekerOptions(5),
+      compressionCodec,
+      batchDelete = true,
+    )
+
+    val sink = S3WriterManager.from(config)
+
+    val topic  = Topic(TopicName)
+    val offset = Offset(1L)
+    val usersWithDecimal =
+      new Struct(UsersSchemaDecimal)
+        .put("name", "sam")
+        .put("title", "mr")
+        .put(
+          "salary",
+          BigDecimal("100.43").setScale(18).bigDecimal,
+        )
+    sink.write(
+      TopicPartitionOffset(topic, 1, offset),
+      MessageDetail(NullSinkData(None),
+                    StructSinkData(usersWithDecimal),
+                    Map.empty[String, SinkData],
+                    None,
+                    topic,
+                    0,
+                    offset,
+      ),
+    )
+
+    sink.close()
+
+    listBucketPath(BucketName, "streamReactorBackups/myTopic/1/").size should be(1)
+
+    remoteFileAsString(BucketName, "streamReactorBackups/myTopic/1/1.json") should be(
+      s"""{"name":"sam","title":"mr","salary":100.430000000000000000}""",
+    )
+  }
 }