Java SDK: Send log timestamps in UTC (#74783)

* Java SDK: Send log timestamps in UTC

Signed-off-by: PoAn Yang <payang@apache.org>

* Fix Java SDK log time zone and a supervisor crash on bad timestamps

Signed-off-by: PoAn Yang <payang@apache.org>

---------

Signed-off-by: PoAn Yang <payang@apache.org>

java-sdk/gradle/verification-metadata.xml

@@ -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"/>

java-sdk/sdk/build.gradle.kts

@@ -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")

java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Logger.kt

@@ -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())
   }

java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/execution/LogTest.kt

@@ -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()
+    }
+  }
 }