uranusjr commented on code in PR #75538:
URL: https://github.com/apache/airflow/pull/75538#discussion_r4490912216
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Server.kt:
##########
@@ -210,13 +212,31 @@ class CoordinatorServer(
aSocket(selector).tcp().connect(logs).use { socket ->
logger.debug("Connected logs", mapOf("addr" to logs))
LogSender.configure(socket.openWriteChannel(autoFlush = true))
- deferral.await()
+ Runtime.getRuntime().addShutdownHook(Thread({
sendQueuedLogs(socket) }, "Airflow log flush"))
+ try {
+ deferral.await()
+ } finally {
+ // The socket closes when the enclosing use block ends, even
when serving fails, so write the
+ // queued lines while the socket is still open.
+ LogSender.detachAndFlush()
+ }
}
}
}
}
}
+ /**
+ * Writes the queued log lines, and waits until the socket has sent them and
closed. Closing the socket
+ * starts sending what is left in its buffer, so without the wait, the JVM
can exit before the last
+ * lines reach the supervisor.
+ */
+ private fun sendQueuedLogs(socket: Socket) {
+ LogSender.detachAndFlush()
+ socket.close()
+ runBlocking { socket.socketContext.join() }
+ }
Review Comment:
On SIGTERM to a non-stoppable task, the JVM can't exit until the backlog is
written, or the process is killed. Put a timeout on the hook.
--
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]
