NullPointerException information lost (#174)

The change brings extra logging to identify where the exception occurs. It changes the error log information.

Co-authored-by: stheppi <[email protected]>
Co-authored-by: Andrew Stevenson <[email protected]>

Author: 3 people

kafka-connect-cloud-common/src/main/scala/io/lenses/streamreactor/connect/cloud/common/sink/writer/Writer.scala

@@ -59,7 +59,15 @@ class Writer[SM <: FileMetadata](
     def innerMessageWrite(writingState: Writing): Either[NonFatalCloudSinkError, Unit] =
       writingState.formatWriter.write(messageDetail) match {
         case Left(err: Throwable) =>
-          logger.error(err.getMessage)
+          logger.error(
+            s"An error occurred while writing using ${writingState.formatWriter.getClass.getSimpleName}. " +
+              s"Details: Topic-Partition: ${messageDetail.topic.value}-${messageDetail.partition}, " +
+              s"Offset: ${messageDetail.offset.value}, " +
+              s"Key: ${messageDetail.key}, " +
+              s"Value: ${messageDetail.value}, " +
+              s"Headers: ${messageDetail.headers}.",
+            err,
+          )
           NonFatalCloudSinkError(err.getMessage, err.some).asLeft
         case Right(_) =>
           writeState =