Improve error message to show type and toString representation of obj…

…ect of unknown type (#111)

kafka-connect-common/src/main/scala/io/lenses/streamreactor/connect/cloud/common/sink/extractors/KafkaConnectExtractor.scala

@@ -27,6 +27,7 @@ import java.util
 import java.lang
 import java.nio.ByteBuffer
 import scala.jdk.CollectionConverters.MapHasAsJava
+import scala.util.Try
 
 object KafkaConnectExtractor extends LazyLogging {
 
@@ -36,8 +37,7 @@ object KafkaConnectExtractor extends LazyLogging {
   def extractFromValue(sinkRecord: SinkRecord, path: Option[String]): Either[Throwable, AnyRef] =
     extract(sinkRecord.value(), Option(sinkRecord.valueSchema()), path)
 
-  // TODO: test with all different types
-  private def extract(
+  private[extractors] def extract(
     extractFrom:   AnyRef,
     extractSchema: Option[Schema],
     maybePath:     Option[String],
@@ -58,12 +58,16 @@ object KafkaConnectExtractor extends LazyLogging {
       case (arrayVal: Array[_], _) => arrayVal.asRight
       case (decimal: BigDecimal, _) => decimal.asRight
       case (decimal: java.math.BigDecimal, _) => decimal.asRight
-      case null => null
+      case (null, _) => Right(null)
       case (structVal: Struct, Some(pnp)) => StructExtractor.extractPathFromStruct(structVal, pnp)
       case (mapVal: Map[_, _], Some(pnp)) => MapExtractor.extractPathFromMap(mapVal.asJava, pnp, extractSchema.orNull)
       case (mapVal: util.Map[_, _], Some(pnp)) => MapExtractor.extractPathFromMap(mapVal, pnp, extractSchema.orNull)
       case (listVal: util.List[_], Some(pnp)) => ArrayExtractor.extractPathFromArray(listVal, pnp, extractSchema.orNull)
-      case otherVal => new ConnectException("Unknown value type: " + otherVal.getClass.getName).asLeft
+      case (leftVal, rightVal) => new ConnectException(
+          s"Unknown value type: `${Try(leftVal.getClass.getName).getOrElse("undefined")}`, string representation: '${Try(
+            leftVal.toString,
+          ).getOrElse("undefined")}', path: `${rightVal.getOrElse("Empty")}`",
+        ).asLeft
     }
   }
 

kafka-connect-common/src/test/scala/io/lenses/streamreactor/connect/cloud/common/sink/extractors/KafkaConnectExtractorTest.scala

@@ -0,0 +1,103 @@
+/*
+ * Copyright 2017-2024 Lenses.io Ltd
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.lenses.streamreactor.connect.cloud.common.sink.extractors
+
+import cats.implicits.catsSyntaxOptionId
+import org.apache.kafka.connect.data.Schema
+import org.apache.kafka.connect.data.SchemaBuilder
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+import org.apache.kafka.connect.data.Struct
+import org.apache.kafka.connect.sink.SinkRecord
+import org.mockito.MockitoSugar
+import org.scalatest.EitherValues
+
+import scala.jdk.CollectionConverters.MapHasAsJava
+import scala.jdk.CollectionConverters.SeqHasAsJava
+
+class KafkaConnectExtractorTest extends AnyFlatSpec with Matchers with MockitoSugar with EitherValues {
+
+  "extractFromKey" should "return the correct value for a given key" in {
+    val sinkRecord = mock[SinkRecord]
+    when(sinkRecord.key()).thenReturn("testKey")
+    when(sinkRecord.keySchema()).thenReturn(null)
+
+    val result = KafkaConnectExtractor.extractFromKey(sinkRecord, None)
+    result shouldBe Right("testKey")
+  }
+
+  "extractFromValue" should "return the correct value for a given value" in {
+    val sinkRecord = mock[SinkRecord]
+    when(sinkRecord.value()).thenReturn("testValue")
+    when(sinkRecord.valueSchema()).thenReturn(null)
+
+    val result = KafkaConnectExtractor.extractFromValue(sinkRecord, None)
+    result shouldBe Right("testValue")
+  }
+
+  "extract" should "handle different types correctly" in {
+    KafkaConnectExtractor.extract(123: java.lang.Integer, None, None) shouldBe Right(123)
+    KafkaConnectExtractor.extract(123L: java.lang.Long, None, None) shouldBe Right(123L)
+    KafkaConnectExtractor.extract(123.45: java.lang.Double, None, None) shouldBe Right(123.45)
+    KafkaConnectExtractor.extract(true: java.lang.Boolean, None, None) shouldBe Right(true)
+    KafkaConnectExtractor.extract("testString", None, None) shouldBe Right("testString")
+  }
+
+  it should "handle Struct type correctly" in {
+
+    val fieldName  = "field1"
+    val fieldValue = "fieldValue"
+
+    val structSchema = SchemaBuilder.struct().field(fieldName, Schema.STRING_SCHEMA).build()
+
+    val struct = new Struct(structSchema).put(fieldName, fieldValue)
+
+    val result = KafkaConnectExtractor.extract(struct, None, Some(fieldName))
+    result shouldBe Right(fieldValue)
+  }
+
+  it should "handle Map type correctly" in {
+    val map = Map("key1" -> "value1").asJava
+
+    val partitionNamePath = "key1"
+    val result            = KafkaConnectExtractor.extract(map, None, Some(partitionNamePath))
+    result shouldBe Right("value1")
+  }
+
+  it should "handle List type correctly" in {
+    val list              = List("value1", "value2").asJava
+    val partitionNamePath = "0"
+
+    val result = KafkaConnectExtractor.extract(list,
+                                               SchemaBuilder.array(Schema.STRING_SCHEMA).build().some,
+                                               Some(partitionNamePath),
+    )
+    result shouldBe Right("value1")
+  }
+
+  it should "return an error for unknown types" in {
+    val unknownType = new Object()
+    val result      = KafkaConnectExtractor.extract(unknownType, None, None)
+    val message     = result.left.value.getMessage
+    message should startWith("Unknown value type: `java.lang.Object`")
+    message should endWith("path: `Empty`")
+  }
+
+  it should "return an error for null types" in {
+    val result = KafkaConnectExtractor.extract(null, None, None)
+    result.value should be(null)
+  }
+}