From e0964fa7735fd9861307cf27bfcdd057e8a9d1ab Mon Sep 17 00:00:00 2001 From: PoAn Yang Date: Thu, 15 Oct 2026 15:39:10 +0800 Subject: [PATCH 1/2] Java SDK: Send log timestamps in UTC Signed-off-by: PoAn Yang --- contributing-docs/30_new_language_sdk.rst | 10 ++- java-sdk/gradle/verification-metadata.xml | 13 ---- java-sdk/sdk/build.gradle.kts | 1 - .../apache/airflow/sdk/execution/Logger.kt | 10 ++- .../apache/airflow/sdk/execution/LogTest.kt | 64 +++++++++++++++++++ 5 files changed, 76 insertions(+), 22 deletions(-) diff --git a/contributing-docs/30_new_language_sdk.rst b/contributing-docs/30_new_language_sdk.rst index 86c65efc716ea..cb680654fa586 100644 --- a/contributing-docs/30_new_language_sdk.rst +++ b/contributing-docs/30_new_language_sdk.rst @@ -293,13 +293,19 @@ 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 reads a timestamp without an offset as a + time in ``[core] default_timezone``, which is UTC by default. If the worker + runs in another time zone, records without an offset then show the wrong time + and appear out of order in the task log. Include the seconds even when they + are zero. The supervisor fails on a timestamp without seconds, such as + ``2026-06-22T12:00``. * 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 diff --git a/java-sdk/gradle/verification-metadata.xml b/java-sdk/gradle/verification-metadata.xml index c474c7afdf489..170dd07993b41 100644 --- a/java-sdk/gradle/verification-metadata.xml +++ b/java-sdk/gradle/verification-metadata.xml @@ -2478,19 +2478,6 @@ - - - - - - - - - - - - - diff --git a/java-sdk/sdk/build.gradle.kts b/java-sdk/sdk/build.gradle.kts index 88226f5c57b75..2ba31187168e6 100644 --- a/java-sdk/sdk/build.gradle.kts +++ b/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") diff --git a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Logger.kt b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Logger.kt index 04757386aa46b..3694fb5ae6026 100644 --- a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Logger.kt +++ b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Logger.kt @@ -22,9 +22,6 @@ 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 @@ -33,6 +30,7 @@ import kotlinx.serialization.json.buildJsonObject import java.util.concurrent.ConcurrentLinkedDeque import kotlin.reflect.KClass import kotlin.time.Clock +import kotlin.time.Instant // Adapted from Python logging. enum class Level( @@ -179,7 +177,7 @@ internal data class LogMessage( val arguments: Map, val logger: String, val level: Level, - val timestamp: LocalDateTime = Clock.System.now().toLocalDateTime(TimeZone.currentSystemDefault()), + val timestamp: Instant = Clock.System.now(), ) /** @@ -253,7 +251,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 +268,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()) } diff --git a/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/execution/LogTest.kt b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/execution/LogTest.kt index e1d7ebf0f63b3..aebc11334f2c8 100644 --- a/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/execution/LogTest.kt +++ b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/execution/LogTest.kt @@ -19,11 +19,25 @@ 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.util.TimeZone +import java.util.concurrent.TimeUnit +import kotlin.time.Clock +import kotlin.time.Duration.Companion.seconds +import kotlin.time.Instant class LogTest { @AfterEach @@ -86,4 +100,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 = Clock.System.now() + val message = LogMessage("hello", emptyMap(), "test", Level.INFO) + val after = Clock.System.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 - 1.seconds)..(after + 1.seconds), + "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.messages.clear() + 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 { + // Once this channel is closed, LogSender queues messages again. Later tests read the queued messages. + channel.cancel(null) + } + } } From 8c0e67c66563b9966f93bca6cf3cdda0e5482008 Mon Sep 17 00:00:00 2001 From: PoAn Yang Date: Mon, 19 Oct 2026 23:05:51 +0800 Subject: [PATCH 2/2] Fix Java SDK log time zone and a supervisor crash on bad timestamps Signed-off-by: PoAn Yang --- contributing-docs/30_new_language_sdk.rst | 14 ++++++----- .../apache/airflow/sdk/execution/Logger.kt | 10 +++++--- .../apache/airflow/sdk/execution/LogTest.kt | 16 ++++++------- .../airflow/sdk/execution_time/supervisor.py | 10 +++++++- .../execution_time/test_supervisor.py | 24 +++++++++++++++++++ 5 files changed, 55 insertions(+), 19 deletions(-) diff --git a/contributing-docs/30_new_language_sdk.rst b/contributing-docs/30_new_language_sdk.rst index cb680654fa586..0be264e97c73e 100644 --- a/contributing-docs/30_new_language_sdk.rst +++ b/contributing-docs/30_new_language_sdk.rst @@ -300,12 +300,14 @@ is a structlog-style event: supported: ``critical``, ``error``, ``warning``, ``info``, ``debug``, and ``notset``. A line carrying any other level is dropped. * ``timestamp`` is an RFC 3339 timestamp with a UTC offset, such as - ``2026-06-22T12:00:00Z``. Airflow reads a timestamp without an offset as a - time in ``[core] default_timezone``, which is UTC by default. If the worker - runs in another time zone, records without an offset then show the wrong time - and appear out of order in the task log. Include the seconds even when they - are zero. The supervisor fails on a timestamp without seconds, such as - ``2026-06-22T12:00``. + ``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 diff --git a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Logger.kt b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Logger.kt index 3694fb5ae6026..975cc5ac76821 100644 --- a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Logger.kt +++ b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Logger.kt @@ -27,10 +27,9 @@ 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 -import kotlin.time.Instant // Adapted from Python logging. enum class Level( @@ -177,7 +176,7 @@ internal data class LogMessage( val arguments: Map, val logger: String, val level: Level, - val timestamp: Instant = Clock.System.now(), + val timestamp: Instant = Instant.now(), ) /** @@ -243,6 +242,11 @@ internal object LogSender { } } + internal fun reset() { + writer = null + messages.clear() + } + private fun sendTo( writer: ByteWriteChannel, message: LogMessage, diff --git a/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/execution/LogTest.kt b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/execution/LogTest.kt index aebc11334f2c8..5bef1223f2d07 100644 --- a/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/execution/LogTest.kt +++ b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/execution/LogTest.kt @@ -33,11 +33,9 @@ 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 -import kotlin.time.Clock -import kotlin.time.Duration.Companion.seconds -import kotlin.time.Instant class LogTest { @AfterEach @@ -108,16 +106,16 @@ class LogTest { val defaultZone = TimeZone.getDefault() try { TimeZone.setDefault(TimeZone.getTimeZone("Asia/Taipei")) - val before = Clock.System.now() + val before = Instant.now() val message = LogMessage("hello", emptyMap(), "test", Level.INFO) - val after = Clock.System.now() + 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 - 1.seconds)..(after + 1.seconds), + Instant.parse(timestamp) in before.minusSeconds(1)..after.plusSeconds(1), "timestamp should be when the message was created: $timestamp", ) } finally { @@ -139,15 +137,15 @@ class LogTest { private fun sendAndRead(message: LogMessage): JsonObject { val channel = ByteChannel(autoFlush = true) try { - LogSender.messages.clear() + 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 { - // Once this channel is closed, LogSender queues messages again. Later tests read the queued messages. - channel.cancel(null) + // Other test classes read LogSender.messages, which LogSender fills only while it has no channel. + LogSender.reset() } } } diff --git a/task-sdk/src/airflow/sdk/execution_time/supervisor.py b/task-sdk/src/airflow/sdk/execution_time/supervisor.py index 8b23068ee505b..388075858bd23 100644 --- a/task-sdk/src/airflow/sdk/execution_time/supervisor.py +++ b/task-sdk/src/airflow/sdk/execution_time/supervisor.py @@ -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 diff --git a/task-sdk/tests/task_sdk/execution_time/test_supervisor.py b/task-sdk/tests/task_sdk/execution_time/test_supervisor.py index ae5c2af97d73b..12a76d25a1be4 100644 --- a/task-sdk/tests/task_sdk/execution_time/test_supervisor.py +++ b/task-sdk/tests/task_sdk/execution_time/test_supervisor.py @@ -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"],