[SPARK-56243][SS] Throw detailed error on malformed Kafka record timestamps

### What changes were proposed in this pull request?

Throw a detailed `KAFKA_MALFORMED_RECORD_TIMESTAMP` message when a record has a malformed or incorrect precision timestamp.

### Why are the changes needed?

Some users may use custom producers which set the timestamp to an incorrect precision level than what is expected by Kafka. Upon hitting this issue, they would get a generic arithmetic overflow error.

### Does this PR introduce _any_ user-facing change?

Yes, it introduces a new error message, however there is no other behavioral change.

### How was this patch tested?

N/A

### Was this patch authored or co-authored using generative AI tooling?

No.

Closes #55038 from kavpreetgrewal/SPARK-56243-timestamp-parsing.

Authored-by: Kavpreet Grewal <kavpreet.grewal@databricks.com>
Signed-off-by: Jungtaek Lim <kabhwan.opensource@gmail.com>

Author: kavpreetgrewal

connector/kafka-0-10-sql/src/main/resources/error/kafka-error-conditions.json

@@ -120,5 +120,13 @@
     "message" : [
       "The specified start <offsetType> <startOffset> is greater than the end <offsetType> <endOffset> for topic <topic> partition <partition>."
     ]
+  },
+  "KAFKA_MALFORMED_RECORD_TIMESTAMP" : {
+    "message": [
+      "Record at topic <topic> partition <partition> offset <offset> has timestamp value <timestamp> that overflows when converting to Spark's microsecond-precision TimestampType.",
+      "Kafka record timestamps must be in milliseconds since epoch per the Kafka protocol. The provided value suggests the producer may be writing timestamps in microseconds or nanoseconds instead.",
+      "Ensure the Kafka producer sets the ProducerRecord timestamp in milliseconds."
+    ],
+    "sqlState": "22008"
   }
 }

connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaExceptions.scala

@@ -178,6 +178,25 @@ object KafkaExceptions {
         "assignedPartitions" -> assignedPartitions.toString))
   }
 
+  def kafkaMalformedRecordTimestamp(
+      topic: String,
+      partition: Int,
+      offset: Long,
+      timestamp: Long,
+      cause: Throwable
+  ): KafkaIllegalStateException = {
+    new KafkaIllegalStateException(
+      errorClass = "KAFKA_MALFORMED_RECORD_TIMESTAMP",
+      messageParameters = Map(
+        "topic" -> topic,
+        "partition" -> partition.toString,
+        "offset" -> offset.toString,
+        "timestamp" -> timestamp.toString
+      ),
+      cause = cause
+    )
+  }
+
   def nullTopicInData(): KafkaIllegalStateException = {
     new KafkaIllegalStateException(
       errorClass = "KAFKA_NULL_TOPIC_IN_DATA",

connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaRecordToRowConverter.scala

@@ -39,13 +39,13 @@ private[kafka010] class KafkaRecordToRowConverter {
   val toInternalRowWithoutHeaders: Record => InternalRow =
     (cr: Record) => InternalRow(
       cr.key, cr.value, UTF8String.fromString(cr.topic), cr.partition, cr.offset,
-      DateTimeUtils.fromJavaTimestamp(new Timestamp(cr.timestamp)), cr.timestampType.id
+      parseTimestamp(cr), cr.timestampType.id
     )
 
   val toInternalRowWithHeaders: Record => InternalRow =
     (cr: Record) => InternalRow(
       cr.key, cr.value, UTF8String.fromString(cr.topic), cr.partition, cr.offset,
-      DateTimeUtils.fromJavaTimestamp(new Timestamp(cr.timestamp)), cr.timestampType.id,
+      parseTimestamp(cr), cr.timestampType.id,
       if (cr.headers.iterator().hasNext) {
         new GenericArrayData(cr.headers.iterator().asScala
           .map(header =>
@@ -90,4 +90,19 @@ private[kafka010] object KafkaRecordToRowConverter {
   def kafkaSchema(includeHeaders: Boolean): StructType = {
     if (includeHeaders) schemaWithHeaders else schemaWithoutHeaders
   }
+
+  private def parseTimestamp(cr: Record): Long = {
+    try {
+      DateTimeUtils.fromJavaTimestamp(new Timestamp(cr.timestamp))
+    } catch {
+      case e: ArithmeticException =>
+        throw KafkaExceptions.kafkaMalformedRecordTimestamp(
+          cr.topic(),
+          cr.partition(),
+          cr.offset(),
+          cr.timestamp(),
+          e
+        )
+    }
+  }
 }

connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchSourceSuite.scala

@@ -1589,6 +1589,36 @@ abstract class KafkaMicroBatchSourceSuiteBase extends KafkaSourceSuiteBase with
       q.stop()
     }
   }
+
+  test("SPARK-56243: malformed record timestamp throws detailed error") {
+    val topic = newTopic()
+    testUtils.createTopic(topic, partitions = 1)
+
+    // Nanosecond-precision timestamp that overflows millis-to-micros conversion
+    val nanoTimestamp = 1712345678123456789L
+    val record = new RecordBuilder(topic, "value").partition(0).timestamp(nanoTimestamp).build()
+    testUtils.sendMessages(Seq(record))
+
+    val kafka = spark
+      .readStream
+      .format("kafka")
+      .option("kafka.bootstrap.servers", testUtils.brokerAddress)
+      .option("subscribe", topic)
+      .option("startingOffsets", "earliest")
+      .load()
+
+    testStream(kafka)(
+      StartStream(),
+      ExpectFailure[KafkaIllegalStateException](e => {
+        val ex = e.asInstanceOf[KafkaIllegalStateException]
+        assert(ex.getCondition === "KAFKA_MALFORMED_RECORD_TIMESTAMP")
+        assert(ex.getSqlState === "22008")
+        assert(ex.getMessage.contains(topic))
+        assert(ex.getMessage.contains(nanoTimestamp.toString))
+        assert(ex.getCause.isInstanceOf[ArithmeticException])
+      })
+    )
+  }
 }
 
 abstract class KafkaMicroBatchV1SourceSuite extends KafkaMicroBatchSourceSuiteBase {