Improves the error in case the input on BytesFormatWriter is not bina…

…ry. (#925) Matches on ByteBuffer as well, Connect framework through the deserialisers can return that Co-authored-by: stheppi <[email protected]>
lensesio · Mar 24, 2023 · ba916f5 · ba916f5
1 parent 9fd0b34
commit ba916f5
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 2 deletions.
diff --git a/...-s3/src/main/scala/io/lenses/streamreactor/connect/aws/s3/formats/BytesFormatWriter.scala b/...-s3/src/main/scala/io/lenses/streamreactor/connect/aws/s3/formats/BytesFormatWriter.scala
@@ -72,9 +72,9 @@ class BytesFormatWriter(outputStreamFn: () => S3OutputStream, bytesWriteMode: By
   def convertToBytes(sinkData: SinkData): Either[Throwable, Array[Byte]] =
     sinkData match {
       case ByteArraySinkData(array, _) => array.asRight
-      case _ =>
+      case v =>
         new IllegalStateException(
-          "Non-binary content received.  Please check your configuration.  It may be advisable to ensure you are using org.apache.kafka.connect.converters.ByteArrayConverter\", exception)\n      case Success(value) => value",
+          s"Non-binary content received: ${v.getClass.getName} .  Please check your configuration.  It may be advisable to ensure you are using org.apache.kafka.connect.converters.ByteArrayConverter.",
         ).asLeft
     }
 

diff --git a/...ala/io/lenses/streamreactor/connect/aws/s3/sink/conversion/ValueToSinkDataConverter.scala b/...ala/io/lenses/streamreactor/connect/aws/s3/sink/conversion/ValueToSinkDataConverter.scala
@@ -21,6 +21,7 @@ import org.apache.kafka.connect.data.Schema
 import org.apache.kafka.connect.data.Struct
 import org.apache.kafka.connect.errors.ConnectException
 
+import java.nio.ByteBuffer
 import java.util
 import scala.jdk.CollectionConverters.MapHasAsScala
 
@@ -39,6 +40,7 @@ object ValueToSinkDataConverter {
     case mapVal:    Map[_, _]      => MapSinkDataConverter(mapVal, schema)
     case mapVal:    util.Map[_, _] => MapSinkDataConverter(mapVal.asScala.toMap, schema)
     case bytesVal:  Array[Byte]    => ByteArraySinkData(bytesVal, schema)
+    case bytesVal:  ByteBuffer     => ByteArraySinkData(bytesVal.array(), schema)
     case arrayVal:  Array[_]       => ArraySinkDataConverter(arrayVal, schema)
     case listVal:   util.List[_]   => ArraySinkDataConverter(listVal.toArray, schema)
     case null     => NullSinkData(schema)