FrankYang0529 commented on code in PR #74783:
URL: https://github.com/apache/airflow/pull/74783#discussion_r4315453403
##########
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:
Added `LogSender.reset()`. It sets the channel to null and clears the queue.
`sendAndRead` calls `reset()` before `configure` and in `finally`.
--
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]
