This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch v3-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-4-test by this push:
new 36e9a46b483 [v3-4-test] Java SDK: Send log timestamps in UTC (#74783)
(#75186)
36e9a46b483 is described below
commit 36e9a46b48383dde01bb0f525b8ffc9491b0ba9e
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Tue Oct 20 14:08:39 2026 +0530
[v3-4-test] Java SDK: Send log timestamps in UTC (#74783) (#75186)
* Java SDK: Send log timestamps in UTC
* Fix Java SDK log time zone and a supervisor crash on bad timestamps
---------
(cherry picked from commit 9091cd0c7b0cb2e957bfa8faf88f087b73d69ae2)
Signed-off-by: PoAn Yang <[email protected]>
Co-authored-by: PoAn Yang <[email protected]>
---
contributing-docs/30_new_language_sdk.rst | 12 ++++-
java-sdk/gradle/verification-metadata.xml | 13 -----
java-sdk/sdk/build.gradle.kts | 1 -
.../org/apache/airflow/sdk/execution/Logger.kt | 16 +++---
.../org/apache/airflow/sdk/execution/LogTest.kt | 62 ++++++++++++++++++++++
.../src/airflow/sdk/execution_time/supervisor.py | 10 +++-
.../task_sdk/execution_time/test_supervisor.py | 24 +++++++++
7 files changed, 114 insertions(+), 24 deletions(-)
diff --git a/contributing-docs/30_new_language_sdk.rst
b/contributing-docs/30_new_language_sdk.rst
index 86c65efc716..0be264e97c7 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 c474c7afdf4..170dd07993b 100644
--- a/java-sdk/gradle/verification-metadata.xml
+++ b/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"/>
diff --git a/java-sdk/sdk/build.gradle.kts b/java-sdk/sdk/build.gradle.kts
index 88226f5c57b..2ba31187168 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 04757386aa4..975cc5ac768 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<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())
}
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 e1d7ebf0f63..5bef1223f2d 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 8b23068ee50..388075858bd 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 ae5c2af97d7..12a76d25a1b 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):
]
[email protected](
+ "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"],
