Header fix?

kafka-connect-mqtt/src/main/scala/com/datamountaineer/streamreactor/connect/mqtt/source/MqttSourceTask.scala

@@ -30,6 +30,8 @@ import org.apache.kafka.common.config.ConfigException
 import org.apache.kafka.connect.source.SourceRecord
 import org.apache.kafka.connect.source.SourceTask
 
+import java.nio.charset.CodingErrorAction
+import scala.io.{Codec, Source}
 import scala.jdk.CollectionConverters.ListHasAsScala
 import scala.jdk.CollectionConverters.MapHasAsScala
 import scala.util.Failure
@@ -43,11 +45,7 @@ class MqttSourceTask extends SourceTask with StrictLogging {
   private val manifest = JarManifest(getClass.getProtectionDomain.getCodeSource.getLocation)
 
   override def start(props: util.Map[String, String]): Unit = {
-
-    logger.info(scala.io.Source.fromInputStream(
-      this.getClass.getResourceAsStream("/mqtt-source-ascii.txt"),
-    ).mkString + s" $version")
-    logger.info(manifest.printManifest())
+    printAsciiHeader()
 
     val conf = if (context.configs().isEmpty) props else context.configs()
 
@@ -90,6 +88,17 @@ class MqttSourceTask extends SourceTask with StrictLogging {
     enableProgress = settings.enableProgress
   }
 
+  private def printAsciiHeader(): Unit = {
+    implicit val codec: Codec = Codec("UTF-8")
+    codec.onMalformedInput(CodingErrorAction.REPLACE)
+    codec.onUnmappableCharacter(CodingErrorAction.REPLACE)
+    logger.info(
+      Source.fromInputStream(
+        getClass.getResourceAsStream("/mqtt-source-ascii.txt"),
+      ).mkString + s" $version")
+    logger.info(manifest.printManifest())
+  }
+
   /**
     * Get all the messages accumulated so far.
     */