jason810496 commented on code in PR #74783:
URL: https://github.com/apache/airflow/pull/74783#discussion_r4313114166
##########
java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/execution/LogTest.kt:
##########
@@ -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)
Review Comment:
nit: this leaves the `LogSender` writer on the cancelled channel for every
later test in the JVM. It works because `send()` queues when the channel is
closed, but a small internal `LogSender.reset()` called here would make the
cleanup explicit instead of relying on that.
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Logger.kt:
##########
@@ -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
Review Comment:
nit: the rest of the SDK uses `java.time.Instant` (`Serde.kt`,
`TriggerRunner.kt`, `Fields.kt`). `java.time.Instant.now().toString()` gives
the same UTC string and always includes the seconds, so we could use it here
and keep one `Instant` type in the SDK.
##########
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``.
Review Comment:
The root cause is on the supervisor side: the `msgspec.json.decode(...,
type=datetime)` call at
[supervisor.py#L2880](https://github.com/apache/airflow/blob/524a8aa2c939640b0bcaa51eae09aa5f5b968b49/task-sdk/src/airflow/sdk/execution_time/supervisor.py#L2880)
isn't guarded, so any SDK that sends a timestamp without seconds still stops
the supervisor before it reports the final state.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
