jason810496 · Oct 20, 2026 · Oct 15, 2026 · Oct 19, 2026
diff --git a/contributing-docs/30_new_language_sdk.rst b/contributing-docs/30_new_language_sdk.rst
@@ -293,13 +293,21 @@ is a structlog-style event:
 
 .. code-block:: text
 
-    {"event": "Starting extraction", "level": "info", "logger": "com.example.SalesPipeline", "timestamp": "2026-06-22T12:00:00", "rows": 42}
+    {"event": "Starting extraction", "level": "info", "logger": "com.example.SalesPipeline", "timestamp": "2026-06-22T12:00:00Z", "rows": 42}
 
 * ``event`` is the log message.
 * ``level`` is the level **name** in lower case. The following levels are
   supported: ``critical``, ``error``, ``warning``, ``info``, ``debug``, and
   ``notset``. A line carrying any other level is dropped.
-* ``timestamp`` is an ISO-8601 timestamp.
+* ``timestamp`` is an RFC 3339 timestamp with a UTC offset, such as
+  ``2026-06-22T12:00:00Z``. Airflow assumes that a timestamp without an offset
+  is in the time zone set by ``[core] default_timezone``, which is UTC by
+  default. If the worker's time zone differs from that setting, records without
+  an offset show the wrong time and appear out of order in the task log.
+  Include the seconds even when they are zero. The supervisor cannot parse a
+  timestamp without seconds, such as ``2026-06-22T12:00``. It moves that value
+  to an ``invalid_timestamp`` field and uses the time it received the record
+  instead.
 * Any remaining keys are forwarded as structured fields on the log record.
 
 See Python documentation on logging for details on log levels. Since log level

@@ -2478,19 +2478,6 @@
             <sha256 value="ff6a22da40040938751db0ae21177e76517dbf126a76796f5426727bf76c1228" origin="Generated by Gradle"/>
          </artifact>
       </component>
-      <component group="org.jetbrains.kotlinx" name="kotlinx-datetime" version="0.8.0">
-         <artifact name="kotlinx-datetime-0.8.0.module">
-            <sha256 value="34793c0991afd4d6cd0df8737d5c380fb84da097147c70ecec1a1d68b06ccc63" origin="Generated by Gradle"/>
-         </artifact>
-      </component>
-      <component group="org.jetbrains.kotlinx" name="kotlinx-datetime-jvm" version="0.8.0">
-         <artifact name="kotlinx-datetime-jvm-0.8.0.jar">
-            <sha256 value="ee44b51f8977c2f51d575299a4fc6782ab3700eba4cfc7fad548cced1543936a" origin="Generated by Gradle"/>
-         </artifact>
-         <artifact name="kotlinx-datetime-jvm-0.8.0.module">
-            <sha256 value="5fe52da3877cebe3eae765f9dea8758009a894d2441cdd6e8ac6fba538f171e7" origin="Generated by Gradle"/>
-         </artifact>
-      </component>
       <component group="org.jetbrains.kotlinx" name="kotlinx-html-jvm" version="0.9.1">
          <artifact name="kotlinx-html-jvm-0.9.1.jar">
             <sha256 value="b68f39cd9ad7b46de0a0e77e7f5908d4e7661f3d0c85d2d9171543fcd5b156fb" origin="Generated by Gradle"/>

@@ -59,7 +59,6 @@ dependencies {
     implementation("com.xenomachina:kotlin-argparser:2.0.7")
     implementation("io.ktor:ktor-network:3.6.0")
     implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.11.0")
-    implementation("org.jetbrains.kotlinx:kotlinx-datetime:0.8.0")
     implementation("org.jetbrains.kotlinx:kotlinx-serialization-json:1.11.0")
     implementation("org.msgpack:msgpack-core:0.9.12")
     implementation("org.msgpack:jackson-dataformat-msgpack:0.9.12")

@@ -22,17 +22,14 @@ package org.apache.airflow.sdk.execution
 import io.ktor.utils.io.ByteWriteChannel
 import io.ktor.utils.io.writeString
 import kotlinx.coroutines.runBlocking
-import kotlinx.datetime.LocalDateTime
-import kotlinx.datetime.TimeZone
-import kotlinx.datetime.toLocalDateTime
 import kotlinx.serialization.json.JsonElement
 import kotlinx.serialization.json.JsonNull
 import kotlinx.serialization.json.JsonPrimitive
 import kotlinx.serialization.json.buildJsonArray
 import kotlinx.serialization.json.buildJsonObject
+import java.time.Instant
 import java.util.concurrent.ConcurrentLinkedDeque
 import kotlin.reflect.KClass
-import kotlin.time.Clock
 
 // Adapted from Python logging.
 enum class Level(
@@ -179,7 +176,7 @@ internal data class LogMessage(
   val arguments: Map<String, Any?>,
   val logger: String,
   val level: Level,
-  val timestamp: LocalDateTime = Clock.System.now().toLocalDateTime(TimeZone.currentSystemDefault()),
+  val timestamp: Instant = Instant.now(),
 )
 
 /**
@@ -245,6 +242,11 @@ internal object LogSender {
     }
   }
 
+  internal fun reset() {
+    writer = null
+    messages.clear()
+  }
+
   private fun sendTo(
     writer: ByteWriteChannel,
     message: LogMessage,
@@ -253,7 +255,7 @@ internal object LogSender {
     map["event"] = message.event
     map["level"] = message.level.name.lowercase()
     map["logger"] = message.logger
-    map["timestamp"] = message.timestamp
+    map["timestamp"] = message.timestamp.toString()
     // TODO: Can this be done asynchronously instead?
     runBlocking { writer.writeString("${map.toJsonElement()}\n") }
   }
@@ -270,5 +272,5 @@ private fun Any?.toJsonElement(): JsonElement =
     is Number -> JsonPrimitive(this)
     is String -> JsonPrimitive(this)
     null -> JsonNull
-    else -> JsonPrimitive(toString()) // Also correctly handles Kotlinx DateTime.
+    else -> JsonPrimitive(toString())
   }
@@ -19,11 +19,23 @@
 
 package org.apache.airflow.sdk.execution
 
+import io.ktor.utils.io.ByteChannel
+import io.ktor.utils.io.readLine
+import kotlinx.coroutines.runBlocking
+import kotlinx.serialization.json.Json
+import kotlinx.serialization.json.JsonObject
+import kotlinx.serialization.json.jsonObject
+import kotlinx.serialization.json.jsonPrimitive
 import org.junit.jupiter.api.AfterEach
+import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.Assertions.assertFalse
 import org.junit.jupiter.api.Assertions.assertTrue
 import org.junit.jupiter.api.DisplayName
 import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.Timeout
+import java.time.Instant
+import java.util.TimeZone
+import java.util.concurrent.TimeUnit
 
 class LogTest {
   @AfterEach
@@ -86,4 +98,54 @@ class LogTest {
     // "foobar" is not a child of "foo"; it must fall back to the global level.
     assertTrue(Log.isEnabledForLevel(Level.INFO, "foobar"))
   }
+
+  @Test
+  @DisplayName("Should send the timestamp in UTC when the JVM runs in a non-UTC time zone")
+  @Timeout(value = 30, unit = TimeUnit.SECONDS)
+  fun shouldSendTimestampInUtc() {
+    val defaultZone = TimeZone.getDefault()
+    try {
+      TimeZone.setDefault(TimeZone.getTimeZone("Asia/Taipei"))
+      val before = Instant.now()
+      val message = LogMessage("hello", emptyMap(), "test", Level.INFO)
+      val after = Instant.now()
+
+      val timestamp = sendAndRead(message)["timestamp"]!!.jsonPrimitive.content
+      assertTrue(timestamp.endsWith("Z"), "timestamp should be in UTC with a Z suffix: $timestamp")
+      // The extra second on each side allows for the wall clock being adjusted during the test. A local
+      // time sent as UTC would still be hours off.
+      assertTrue(
+        Instant.parse(timestamp) in before.minusSeconds(1)..after.plusSeconds(1),
+        "timestamp should be when the message was created: $timestamp",
+      )
+    } finally {
+      TimeZone.setDefault(defaultZone)
+    }
+  }
+
+  @Test
+  @DisplayName("Should keep the seconds of a timestamp that falls on a whole minute")
+  @Timeout(value = 30, unit = TimeUnit.SECONDS)
+  fun shouldKeepSecondsOnWholeMinute() {
+    val message = LogMessage("hello", emptyMap(), "test", Level.INFO, Instant.parse("2026-10-13T09:41:00Z"))
+
+    // The supervisor cannot parse a timestamp without seconds, such as 2026-10-13T09:41.
+    assertEquals("2026-10-13T09:41:00Z", sendAndRead(message)["timestamp"]!!.jsonPrimitive.content)
+  }
+
+  /** Sends [message] through [LogSender] and returns the JSON object that [LogSender] writes for the message. */
+  private fun sendAndRead(message: LogMessage): JsonObject {
+    val channel = ByteChannel(autoFlush = true)
+    try {
+      LogSender.reset()
+      LogSender.configure(channel)
+      LogSender.send(message)
+      val line = runBlocking { Json.parseToJsonElement(channel.readLine()!!) }.jsonObject
+      assertEquals(message.event, line["event"]!!.jsonPrimitive.content, "the first line should be the message just sent")
+      return line
+    } finally {
+      // Other test classes read LogSender.messages, which LogSender fills only while it has no channel.
+      LogSender.reset()
+    }
+  }
 }
@@ -2907,7 +2907,15 @@ def process_log_messages_from_subprocess(
         if ts := event.get("timestamp"):
             # We use msgspec to decode the timestamp as it does it orders of magnitude quicker than
             # datetime.strptime cn
-            event["timestamp"] = msgspec.json.decode(f'"{ts}"', type=datetime)
+            try:
+                event["timestamp"] = msgspec.json.decode(f'"{ts}"', type=datetime)
+            except msgspec.DecodeError:
+                # The subprocess sends the timestamp field, and msgspec cannot always parse its value. For
+                # example, user code can call self.log.info("...", timestamp=1697190085).
+                # An uncaught DecodeError would stop the task supervisor before it reports the
+                # task state, or stop the whole Dag processor during Dag parsing. Moving the value to
+                # invalid_timestamp lets MaybeTimeStamper add the time the supervisor received the line.
+                event["invalid_timestamp"] = event.pop("timestamp")
 
         if exc := event.pop("exception", None):
             # TODO: convert the dict back to a pretty stack trace

@@ -5548,6 +5548,30 @@ def test_process_log_messages_from_subprocess(monkeypatch, caplog):
     ]
 
 
+@pytest.mark.parametrize(
+    "bad_timestamp",
+    [
+        pytest.param(1697190085, id="epoch-number"),
+        pytest.param("2026-10-13T17:41", id="no-seconds"),
+        # msgspec raises a plain DecodeError (not a ValidationError) for the newline, so this case
+        # fails if the except clause only catches ValidationError.
+        pytest.param("2026-10-13\n17:41:25", id="newline"),
+    ],
+)
+def test_process_log_messages_keeps_line_with_invalid_timestamp(bad_timestamp, caplog, time_machine):
+    time_machine.move_to(timezone.datetime(2026, 10, 15, 8, 0), tick=False)
+
+    gen = process_log_messages_from_subprocess(loggers=(structlog.get_logger(),))
+    next(gen)
+
+    gen.send(
+        msgspec.json.encode({"level": "info", "event": "Batch done", "timestamp": bad_timestamp}) + b"\n"
+    )
+
+    assert caplog.messages == ["Batch done"]
+    assert {"timestamp": "2026-10-15T08:00:00Z", "invalid_timestamp": bad_timestamp} in caplog
+
+
 @pytest.mark.parametrize(
     "error_message",
     ["write to closed file", "I/O operation on closed file"],