diff --git a/contributing-docs/30_new_language_sdk.rst b/contributing-docs/30_new_language_sdk.rst index 86c65efc716ea..0be264e97c73e 100644 --- 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 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..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 @@ -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, 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()) } 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..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 @@ -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() + } + } } 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"],