("mavenJava") {
+ artifactId = "airflow-sdk-log4j2"
+ from(components["java"])
+ pom {
+ name = "Apache Airflow Java SDK Log4j 2 Appender"
+ description = "Routes Log4j 2 log calls from task code through the SDK to Airflow's task log store."
+ }
+ }
+ }
+}
diff --git a/java-sdk/log4j2/src/main/java/org/apache/airflow/sdk/log4j/AirflowLog4jAppender.java b/java-sdk/log4j2/src/main/java/org/apache/airflow/sdk/log4j/AirflowLog4jAppender.java
new file mode 100644
index 0000000000000..1c27cd1dd796c
--- /dev/null
+++ b/java-sdk/log4j2/src/main/java/org/apache/airflow/sdk/log4j/AirflowLog4jAppender.java
@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.airflow.sdk.log4j;
+
+import java.io.PrintWriter;
+import java.io.StringWriter;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.airflow.sdk.execution.Level;
+import org.apache.airflow.sdk.execution.Log;
+import org.apache.logging.log4j.core.Appender;
+import org.apache.logging.log4j.core.Core;
+import org.apache.logging.log4j.core.Filter;
+import org.apache.logging.log4j.core.LogEvent;
+import org.apache.logging.log4j.core.appender.AbstractAppender;
+import org.apache.logging.log4j.core.config.Property;
+import org.apache.logging.log4j.core.config.plugins.Plugin;
+import org.apache.logging.log4j.core.config.plugins.PluginAttribute;
+import org.apache.logging.log4j.core.config.plugins.PluginElement;
+import org.apache.logging.log4j.core.config.plugins.PluginFactory;
+import org.apache.logging.log4j.spi.StandardLevel;
+
+/**
+ * A Log4j {@link Appender} to route logs to Airflow.
+ *
+ * This class is not called explicitly. An annotation processor reads the
+ * class (since it's annotated with {@link Plugin}) and generates information
+ * needed by Log4J.
+ */
+@Plugin(
+ name = "AirflowAppender",
+ category = Core.CATEGORY_NAME,
+ elementType = Appender.ELEMENT_TYPE)
+public final class AirflowLog4jAppender extends AbstractAppender {
+
+ private AirflowLog4jAppender(String name, Filter filter) {
+ super(name, filter, null, true, Property.EMPTY_ARRAY);
+ }
+
+ @PluginFactory
+ public static AirflowLog4jAppender createAppender(
+ @PluginAttribute("name") String name, @PluginElement("Filter") Filter filter) {
+ return new AirflowLog4jAppender(name == null ? "AirflowAppender" : name, filter);
+ }
+
+ @Override
+ public void append(LogEvent event) {
+ var level = convert(event.getLevel());
+ var logger = event.getLoggerName();
+ if (!Log.INSTANCE.isEnabledForLevel(level, logger)) return;
+ // Log4J does not really provide a good way to access the underlying unformatted data
+ // since it allows vastly different logging mechanisms. We pre-format the message here
+ // and send the exception and marker separately as structured metadata.
+ String message = event.getMessage().getFormattedMessage();
+ Map args = new HashMap<>();
+ var thrown = event.getThrown();
+ if (thrown != null) {
+ args.put("exception", stackTrace(thrown));
+ }
+ var marker = event.getMarker();
+ if (marker != null) {
+ args.put("marker", marker.getName());
+ }
+ Log.INSTANCE.send(level, logger, message, args);
+ }
+
+ private static Level convert(org.apache.logging.log4j.Level level) {
+ var v = level.intLevel();
+ if (v < StandardLevel.ERROR.intLevel()) return Level.CRITICAL;
+ if (v < StandardLevel.WARN.intLevel()) return Level.ERROR;
+ if (v < StandardLevel.INFO.intLevel()) return Level.WARNING;
+ if (v < StandardLevel.DEBUG.intLevel()) return Level.INFO;
+ if (v < StandardLevel.TRACE.intLevel()) return Level.DEBUG;
+ return Level.NOTSET;
+ }
+
+ private static String stackTrace(Throwable t) {
+ var sw = new StringWriter();
+ t.printStackTrace(new PrintWriter(sw));
+ return sw.toString();
+ }
+}
diff --git a/java-sdk/log4j2/src/test/kotlin/org/apache/airflow/sdk/log4j/AirflowLog4jAppenderTest.kt b/java-sdk/log4j2/src/test/kotlin/org/apache/airflow/sdk/log4j/AirflowLog4jAppenderTest.kt
new file mode 100644
index 0000000000000..f8284f5c5b7e8
--- /dev/null
+++ b/java-sdk/log4j2/src/test/kotlin/org/apache/airflow/sdk/log4j/AirflowLog4jAppenderTest.kt
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.airflow.sdk.log4j
+
+import io.mockk.every
+import io.mockk.just
+import io.mockk.mockkObject
+import io.mockk.runs
+import io.mockk.slot
+import io.mockk.unmockkAll
+import io.mockk.verify
+import org.apache.airflow.sdk.execution.Level
+import org.apache.airflow.sdk.execution.Log
+import org.apache.logging.log4j.MarkerManager
+import org.apache.logging.log4j.core.impl.Log4jLogEvent
+import org.apache.logging.log4j.message.SimpleMessage
+import org.junit.jupiter.api.AfterEach
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.api.Test
+import org.apache.logging.log4j.Level as L4jLevel
+
+class AirflowLog4jAppenderTest {
+ private lateinit var appender: AirflowLog4jAppender
+
+ @BeforeEach
+ fun setUp() {
+ appender = AirflowLog4jAppender.createAppender("AirflowAppender", null)
+ mockkObject(Log)
+ every { Log.isEnabledForLevel(any(), any()) } returns true
+ every { Log.send(any(), any(), any(), any>()) } just runs
+ }
+
+ @AfterEach
+ fun tearDown() {
+ unmockkAll()
+ }
+
+ @Test
+ fun `factory honors the configured name and defaults when absent`() {
+ assertEquals("Airflow", AirflowLog4jAppender.createAppender("Airflow", null).name)
+ assertEquals("AirflowAppender", AirflowLog4jAppender.createAppender(null, null).name)
+ }
+
+ @Test
+ fun `level conversions`() {
+ val cases =
+ listOf(
+ L4jLevel.FATAL to Level.CRITICAL,
+ L4jLevel.ERROR to Level.ERROR,
+ L4jLevel.WARN to Level.WARNING,
+ L4jLevel.INFO to Level.INFO,
+ L4jLevel.DEBUG to Level.DEBUG,
+ L4jLevel.TRACE to Level.NOTSET,
+ L4jLevel.ALL to Level.NOTSET,
+ )
+ cases.forEach { (l4jLevel, expected) ->
+ val capturedLevel = slot()
+ every { Log.send(capture(capturedLevel), any(), any(), any>()) } just runs
+ appender.append(event("msg", l4jLevel))
+ assertEquals(expected, capturedLevel.captured, "Log4j $l4jLevel should map to SDK $expected")
+ }
+ }
+
+ @Test
+ fun `formatted message and logger name are forwarded`() {
+ val capturedArgs = slot>()
+ appender.append(event("hello world", L4jLevel.INFO, loggerName = "com.example.Task"))
+ verify { Log.send(Level.INFO, "com.example.Task", "hello world", capture(capturedArgs)) }
+ assertTrue(capturedArgs.captured.isEmpty())
+ }
+
+ @Test
+ fun `thrown is stored under the exception key`() {
+ val capturedArgs = slot>()
+ every { Log.send(any(), any(), any(), capture(capturedArgs)) } just runs
+ appender.append(event("failure", L4jLevel.ERROR, thrown = RuntimeException("kaboom")))
+ assertTrue(capturedArgs.captured["exception"].toString().contains("kaboom"))
+ }
+
+ @Test
+ fun `marker name is stored under the marker key`() {
+ val capturedArgs = slot>()
+ every { Log.send(any(), any(), any(), capture(capturedArgs)) } just runs
+ appender.append(event("hi", L4jLevel.INFO, marker = MarkerManager.getMarker("AUDIT")))
+ assertEquals("AUDIT", capturedArgs.captured["marker"])
+ }
+
+ private fun event(
+ message: String,
+ level: L4jLevel,
+ loggerName: String = "test.Logger",
+ thrown: Throwable? = null,
+ marker: org.apache.logging.log4j.Marker? = null,
+ ) = Log4jLogEvent
+ .newBuilder()
+ .setLoggerName(loggerName)
+ .setLevel(level)
+ .setMessage(SimpleMessage(message))
+ .setThrown(thrown)
+ .setMarker(marker)
+ .build()
+}
diff --git a/java-sdk/sdk/build.gradle.kts b/java-sdk/sdk/build.gradle.kts
index 216ef76513123..1fcd07292a4da 100644
--- a/java-sdk/sdk/build.gradle.kts
+++ b/java-sdk/sdk/build.gradle.kts
@@ -21,6 +21,7 @@ val airflowSupervisorSchemaVersion: String by project
plugins {
`java-library`
+ `java-test-fixtures`
id("airflow-jvm-conventions")
id("airflow-publish")
id("org.jetbrains.dokka") version "2.2.0"
@@ -251,6 +252,9 @@ publishing {
create("mavenJava") {
artifactId = "airflow-sdk"
from(components["java"])
+ // test-fixtures are not published to Maven Central.
+ suppressPomMetadataWarningsFor("testFixturesApiElements")
+ suppressPomMetadataWarningsFor("testFixturesRuntimeElements")
artifact(javadocJar)
pom {
name = "Apache Airflow Java SDK"
diff --git a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Server.kt b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Server.kt
index 2bab74bf23d2b..6f82ca3c4c3bd 100644
--- a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Server.kt
+++ b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Server.kt
@@ -25,6 +25,7 @@ import io.ktor.network.sockets.InetSocketAddress
import io.ktor.network.sockets.aSocket
import io.ktor.network.sockets.openReadChannel
import io.ktor.network.sockets.openWriteChannel
+import kotlinx.coroutines.CompletableDeferred
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.launch
@@ -139,20 +140,27 @@ class Server(
*/
suspend fun serveAsync(bundle: Bundle) =
coroutineScope {
+ val deferral = CompletableDeferred()
+
launch {
- aSocket(SelectorManager(Dispatchers.IO)).tcp().connect(comm).use { socket ->
- logger.debug("Connected comm", mapOf("addr" to comm))
- CoordinatorComm(
- bundle,
- socket.openReadChannel(),
- socket.openWriteChannel(autoFlush = true),
- ).startProcessing()
+ try {
+ aSocket(SelectorManager(Dispatchers.IO)).tcp().connect(comm).use { socket ->
+ logger.debug("Connected comm", mapOf("addr" to comm))
+ CoordinatorComm(
+ bundle,
+ socket.openReadChannel(),
+ socket.openWriteChannel(autoFlush = true),
+ ).startProcessing()
+ }
+ } finally {
+ deferral.complete(Unit)
}
}
launch {
aSocket(SelectorManager(Dispatchers.IO)).tcp().connect(logs).use { socket ->
logger.debug("Connected logs", mapOf("addr" to logs))
LogSender.configure(socket.openWriteChannel(autoFlush = true))
+ deferral.await()
}
}
}
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 27005f92082cb..b778e9aa79f9a 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
@@ -34,52 +34,188 @@ import java.util.concurrent.ConcurrentLinkedDeque
import kotlin.reflect.KClass
import kotlin.time.Clock
-enum class Level { ERROR, DEBUG, }
+// Adapted from Python logging.
+enum class Level(
+ val value: Short,
+) {
+ CRITICAL(50),
+ ERROR(40),
+ WARNING(30),
+ INFO(20),
+ DEBUG(10),
+ NOTSET(0),
+ ;
+
+ internal companion object {
+ val accepted = entries.associateBy { it.name }
+
+ fun parse(s: String?): Level? = s?.let { Level.accepted[it.trim().uppercase()] }
+ }
+}
+
+/**
+ * Parser for the namespace levels configuration.
+ *
+ * The value is a series of `=` pairs separated by whitespaces and/or commas.
+ * Each `` must name one of [Level]s (case-insensitive). When the same logger appears
+ * more than once, the last value wins.
+ */
+internal object NamespaceLevels {
+ const val ENV_VAR = "AIRFLOW__LOGGING__NAMESPACE_LEVELS"
+ private const val LOGGER_NAME = "org.apache.airflow.sdk.execution.NamespaceLevels"
+
+ /**
+ * Parse [raw] into per-logger [Level] overrides.
+ *
+ * Invalid entries are skipped — the affected logger then falls back to the global level — and
+ * reported once at [Level.ERROR], so a misconfiguration never takes down logging nor discards the
+ * valid overrides. A `null`, empty, or whitespace/comma-only value yields no overrides.
+ * Surrounding and repeated separators are ignored.
+ */
+ fun parse(raw: String?): Map {
+ if (raw.isNullOrBlank()) return emptyMap()
+
+ val levels = mutableMapOf()
+ val errors = mutableListOf()
+ raw
+ .split(Regex("""[\s,]+"""))
+ .filter { it.isNotEmpty() }
+ .forEach { entry -> parseEntry(entry, levels)?.let { errors += it } }
+
+ if (errors.isNotEmpty()) {
+ LogSender.send(
+ LogMessage(
+ event = "Ignoring invalid $ENV_VAR entries: ${errors.joinToString("; ")}",
+ arguments = emptyMap(),
+ logger = LOGGER_NAME,
+ level = Level.ERROR,
+ ),
+ )
+ }
+ return levels
+ }
+
+ /** Parse a single [entry] into [levels], or return a description of why it was skipped. */
+ private fun parseEntry(
+ entry: String,
+ levels: MutableMap,
+ ): String? {
+ val separator = entry.indexOf('=')
+ if (separator < 0) {
+ return "malformed entry \"$entry\", expected \"=\""
+ }
+
+ val logger = entry.substring(0, separator).trim()
+ if (logger.isEmpty()) {
+ return "malformed entry \"$entry\", logger name is empty"
+ }
+
+ val levelName = entry.substring(separator + 1).trim()
+ val level =
+ Level.parse(levelName)
+ ?: return "invalid level \"$levelName\" for logger \"$logger\", " +
+ "expected one of: ${Level.entries.joinToString(", ") { it.name }}"
+
+ levels[logger] = level
+ return null
+ }
+}
+
+/**
+ * Public entry point into Airflow's log pipeline.
+ *
+ * This is useful for Java-side logging providers such as [java.util.logging]
+ * and SLF4J to integrate logs they receive into Airflow.
+ *
+ * Not intended for use by task code.
+ */
+object Log {
+ private const val LOGGING_LEVEL_ENV = "AIRFLOW__LOGGING__LOGGING_LEVEL"
+
+ internal var globalThreshold = Level.parse(System.getenv(LOGGING_LEVEL_ENV)) ?: Level.INFO
+ internal var namedThresholds = NamespaceLevels.parse(System.getenv(NamespaceLevels.ENV_VAR))
+
+ /**
+ * Whether a [level] message from the logger called [name] should be emitted.
+ *
+ * Thresholds cascade down the dotted-name hierarchy, mirroring Java logging
+ * convention. A logger inherits the level of its nearest configured
+ * ancestor, so with `foo=WARNING` configured, loggers named e.g. `foo.bar`
+ * also resolves to `WARNING` without additional configuration. A logger with
+ * no configured ancestor falls back to [globalThreshold].
+ */
+ fun isEnabledForLevel(
+ level: Level,
+ name: String?,
+ ): Boolean {
+ var threshold = name
+ while (threshold != null) {
+ namedThresholds[threshold]?.let { return level.value >= it.value }
+ threshold = threshold.substringBeforeLast('.', missingDelimiterValue = "").ifEmpty { null }
+ }
+ return level.value >= globalThreshold.value
+ }
+
+ fun send(
+ level: Level,
+ logger: String,
+ event: String,
+ arguments: Map = emptyMap(),
+ ) {
+ if (!isEnabledForLevel(level, logger)) return
+ LogSender.send(LogMessage(event, arguments, logger, level))
+ }
+
+ fun send(
+ level: Level,
+ logger: String,
+ event: String,
+ buildArguments: MutableMap.() -> Unit,
+ ) = send(level, logger, event, buildMap(buildArguments))
+}
internal data class LogMessage(
val event: String,
- val arguments: Map,
- val logger: Logger,
+ val arguments: Map,
+ val logger: String,
val level: Level,
val timestamp: LocalDateTime = Clock.System.now().toLocalDateTime(TimeZone.currentSystemDefault()),
)
+/**
+ * Logger used by task scaffolding.
+ *
+ * This is a thin wrapper around [LogSender] that our own code can
+ * use instead of needing to go through a "real" logging provider.
+ */
internal class Logger(
- cls: KClass<*>,
+ val name: String,
) {
- val name: String? = cls.java.typeName
-
- // TODO: Actually implement level filtering.
- @Suppress("UNUSED_PARAMETER")
- fun isEnabledForLevel(level: Level): Boolean = true
+ constructor(cls: KClass<*>) : this(cls.java.typeName)
fun debug(
message: String,
arguments: Map = emptyMap(),
- ) {
- log(Level.DEBUG, message, arguments)
- }
+ ) = log(Level.DEBUG, message, arguments)
fun error(
message: String,
arguments: Map = emptyMap(),
- ) {
- log(Level.ERROR, message, arguments)
- }
+ ) = log(Level.ERROR, message, arguments)
private fun log(
level: Level,
event: String,
arguments: Map,
) {
- if (!isEnabledForLevel(level)) return
- LogSender.send(LogMessage(event, arguments, this, level))
+ if (!Log.isEnabledForLevel(level, name)) return
+ LogSender.send(LogMessage(event, arguments, name, level))
}
}
internal object LogSender {
private var writer: ByteWriteChannel? = null
- private val messages: ConcurrentLinkedDeque = ConcurrentLinkedDeque()
+ internal val messages: ConcurrentLinkedDeque = ConcurrentLinkedDeque()
fun configure(channel: ByteWriteChannel) {
writer = channel
@@ -106,7 +242,7 @@ internal object LogSender {
val map = message.arguments.toMutableMap()
map["event"] = message.event
map["level"] = message.level.name.lowercase()
- map["logger"] = message.logger.name ?: "(java)"
+ map["logger"] = message.logger
map["timestamp"] = message.timestamp
// TODO: Can this be done asynchronously instead?
runBlocking { writer.writeString("${map.toJsonElement()}\n") }
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
new file mode 100644
index 0000000000000..e1d7ebf0f63b3
--- /dev/null
+++ b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/execution/LogTest.kt
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.airflow.sdk.execution
+
+import org.junit.jupiter.api.AfterEach
+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
+
+class LogTest {
+ @AfterEach
+ fun tearDown() {
+ LogCapture.resetThresholds()
+ }
+
+ @Test
+ @DisplayName("Without overrides, a logger is gated by the global threshold")
+ fun shouldUseGlobalThreshold() {
+ Log.globalThreshold = Level.INFO
+ Log.namedThresholds = emptyMap()
+
+ assertTrue(Log.isEnabledForLevel(Level.WARNING, "any"))
+ assertTrue(Log.isEnabledForLevel(Level.INFO, "any"))
+ assertFalse(Log.isEnabledForLevel(Level.DEBUG, "any"))
+ }
+
+ @Test
+ @DisplayName("An exact namespace override applies to its logger; unrelated loggers use global")
+ fun shouldApplyNamespaceOverride() {
+ Log.globalThreshold = Level.INFO
+ Log.namedThresholds = mapOf("chatty" to Level.DEBUG)
+
+ assertTrue(Log.isEnabledForLevel(Level.DEBUG, "chatty"))
+ assertFalse(Log.isEnabledForLevel(Level.DEBUG, "other"))
+ assertTrue(Log.isEnabledForLevel(Level.INFO, "other"))
+ }
+
+ @Test
+ @DisplayName("A nested logger inherits the level of its nearest configured ancestor")
+ fun shouldInheritFromAncestor() {
+ Log.globalThreshold = Level.INFO
+ Log.namedThresholds = mapOf("foo" to Level.WARNING)
+
+ // foo.bar and foo.bar.rex have no exact entry, so they inherit foo=WARNING.
+ assertFalse(Log.isEnabledForLevel(Level.INFO, "foo.bar"))
+ assertFalse(Log.isEnabledForLevel(Level.INFO, "foo.bar.rex"))
+ assertTrue(Log.isEnabledForLevel(Level.WARNING, "foo.bar.rex"))
+ }
+
+ @Test
+ @DisplayName("The most specific configured ancestor wins")
+ fun shouldPreferMostSpecificAncestor() {
+ Log.globalThreshold = Level.INFO
+ Log.namedThresholds = mapOf("foo" to Level.WARNING, "foo.bar" to Level.DEBUG)
+
+ // foo.bar.rex's nearest ancestor is foo.bar=DEBUG, not foo=WARNING.
+ assertTrue(Log.isEnabledForLevel(Level.DEBUG, "foo.bar.rex"))
+ // A sibling under foo (but not under foo.bar) still resolves to foo=WARNING.
+ assertFalse(Log.isEnabledForLevel(Level.INFO, "foo.baz"))
+ }
+
+ @Test
+ @DisplayName("Inheritance respects dotted-segment boundaries, not raw string prefixes")
+ fun shouldNotMatchPartialSegmentPrefixes() {
+ Log.globalThreshold = Level.INFO
+ Log.namedThresholds = mapOf("foo" to Level.WARNING)
+
+ // "foobar" is not a child of "foo"; it must fall back to the global level.
+ assertTrue(Log.isEnabledForLevel(Level.INFO, "foobar"))
+ }
+}
diff --git a/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/execution/NamespaceLevelsTest.kt b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/execution/NamespaceLevelsTest.kt
new file mode 100644
index 0000000000000..e2a4cb306e568
--- /dev/null
+++ b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/execution/NamespaceLevelsTest.kt
@@ -0,0 +1,134 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.airflow.sdk.execution
+
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.api.DisplayName
+import org.junit.jupiter.api.Test
+
+class NamespaceLevelsTest {
+ @BeforeEach
+ fun setUp() {
+ LogCapture.drain() // discard anything buffered by earlier tests
+ }
+
+ @Test
+ @DisplayName("null, empty and separator-only input parse to no overrides and log nothing")
+ fun shouldParseBlankToEmptyMap() {
+ for (raw in listOf(null, "", " ", " , , ")) {
+ assertEquals(emptyMap(), NamespaceLevels.parse(raw), "levels for <$raw>")
+ }
+ assertTrue(LogCapture.drain().isEmpty(), "no log should be emitted for blank input")
+ }
+
+ @Test
+ @DisplayName("Should parse the documented example without logging")
+ fun shouldParseDocumentedExample() {
+ assertEquals(
+ mapOf("sqlalchemy" to Level.INFO, "sqlalchemy.engine" to Level.DEBUG),
+ NamespaceLevels.parse("sqlalchemy=INFO sqlalchemy.engine=DEBUG"),
+ )
+ assertTrue(LogCapture.drain().isEmpty(), "no log should be emitted for valid input")
+ }
+
+ @Test
+ @DisplayName("Should accept whitespace, commas and a mix as separators")
+ fun shouldAcceptVariousSeparators() {
+ val expected = mapOf("a" to Level.INFO, "b" to Level.DEBUG, "c" to Level.ERROR)
+ assertEquals(expected, NamespaceLevels.parse("a=INFO b=DEBUG c=ERROR"))
+ assertEquals(expected, NamespaceLevels.parse("a=INFO,b=DEBUG,c=ERROR"))
+ assertEquals(expected, NamespaceLevels.parse(" a=INFO ,, b=DEBUG\t c=ERROR "))
+ }
+
+ @Test
+ @DisplayName("Level names are case-insensitive")
+ fun shouldParseLevelCaseInsensitively() {
+ assertEquals(
+ mapOf("a" to Level.INFO, "b" to Level.CRITICAL),
+ NamespaceLevels.parse("a=info b=Critical"),
+ )
+ }
+
+ @Test
+ @DisplayName("When a logger is repeated the last value wins")
+ fun shouldLetLastValueWin() {
+ assertEquals(mapOf("a" to Level.ERROR), NamespaceLevels.parse("a=INFO a=ERROR"))
+ }
+
+ @Test
+ @DisplayName("An entry without '=' is skipped, valid entries kept, and an error is logged")
+ fun shouldSkipEntryWithoutEquals() {
+ // "sqlalchemy" keeps its override; only the malformed "botocore" falls back to the global level.
+ assertEquals(mapOf("sqlalchemy" to Level.INFO), NamespaceLevels.parse("sqlalchemy=INFO botocore"))
+
+ val error = LogCapture.drain().single()
+ assertEquals(Level.ERROR, error.level)
+ assertEquals("org.apache.airflow.sdk.execution.NamespaceLevels", error.logger)
+ assertTrue(error.event.contains("AIRFLOW__LOGGING__NAMESPACE_LEVELS"), error.event)
+ assertTrue(error.event.contains("botocore"), error.event)
+ assertTrue(error.event.contains("="), error.event)
+ }
+
+ @Test
+ @DisplayName("An entry with an empty logger name is skipped while valid entries are kept")
+ fun shouldSkipEmptyLoggerName() {
+ assertEquals(mapOf("a" to Level.DEBUG), NamespaceLevels.parse("=INFO a=DEBUG"))
+ assertTrue(
+ LogCapture
+ .drain()
+ .single()
+ .event
+ .contains("logger name is empty"),
+ )
+ }
+
+ @Test
+ @DisplayName("An unknown level is skipped and reported with the valid levels")
+ fun shouldSkipUnknownLevel() {
+ assertEquals(emptyMap(), NamespaceLevels.parse("sqlalchemy=VERBOSE"))
+
+ val event = LogCapture.drain().single().event
+ assertTrue(event.contains("VERBOSE"), event)
+ assertTrue(event.contains("sqlalchemy"), event)
+ assertTrue(event.contains("INFO"), event)
+ }
+
+ @Test
+ @DisplayName("An entry with an empty level is skipped")
+ fun shouldSkipEmptyLevel() {
+ assertEquals(emptyMap(), NamespaceLevels.parse("sqlalchemy="))
+ assertEquals(1, LogCapture.drain().size)
+ }
+
+ @Test
+ @DisplayName("Every invalid entry is reported in a single log, valid ones kept")
+ fun shouldReportEveryInvalidEntry() {
+ assertEquals(
+ mapOf("a" to Level.INFO, "c" to Level.DEBUG),
+ NamespaceLevels.parse("a=INFO botocore b=NOPE c=DEBUG"),
+ )
+
+ val event = LogCapture.drain().single().event
+ assertTrue(event.contains("botocore"), event)
+ assertTrue(event.contains("NOPE"), event)
+ }
+}
diff --git a/java-sdk/sdk/src/testFixtures/kotlin/org/apache/airflow/sdk/execution/LogCapture.kt b/java-sdk/sdk/src/testFixtures/kotlin/org/apache/airflow/sdk/execution/LogCapture.kt
new file mode 100644
index 0000000000000..5c72d36cbbc38
--- /dev/null
+++ b/java-sdk/sdk/src/testFixtures/kotlin/org/apache/airflow/sdk/execution/LogCapture.kt
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.airflow.sdk.execution
+
+data class CapturedLogMessage(
+ val level: Level,
+ val logger: String,
+ val event: String,
+ val arguments: Map,
+)
+
+object LogCapture {
+ fun drain(): List =
+ buildList {
+ var msg = LogSender.messages.poll()
+ while (msg != null) {
+ add(CapturedLogMessage(msg.level, msg.logger, msg.event, msg.arguments))
+ msg = LogSender.messages.poll()
+ }
+ }
+
+ fun resetThresholds() {
+ configureThresholds(Level.NOTSET, emptyMap())
+ }
+
+ fun configureThresholds(
+ global: Level,
+ named: Map = emptyMap(),
+ ) {
+ Log.globalThreshold = global
+ Log.namedThresholds = named
+ }
+}
diff --git a/java-sdk/settings.gradle.kts b/java-sdk/settings.gradle.kts
index fd4f4ac001c68..d8792763718da 100644
--- a/java-sdk/settings.gradle.kts
+++ b/java-sdk/settings.gradle.kts
@@ -22,4 +22,4 @@ plugins {
}
rootProject.name = "airflow-java-sdk"
-include("bom", "plugin", "processor", "sdk")
+include("bom", "jpl", "jul", "log4j2", "plugin", "processor", "sdk", "slf4j")
diff --git a/java-sdk/slf4j/build.gradle.kts b/java-sdk/slf4j/build.gradle.kts
new file mode 100644
index 0000000000000..5ed02fa64f7e7
--- /dev/null
+++ b/java-sdk/slf4j/build.gradle.kts
@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+plugins {
+ `java-library`
+ id("airflow-jvm-conventions")
+ id("airflow-publish")
+}
+
+val slf4jVersion: String by project
+
+abstract class GenerateConstants : DefaultTask() {
+ @get:Input
+ abstract val requestedSlf4jApiVersion: Property
+
+ @get:OutputDirectory
+ abstract val outputDir: DirectoryProperty
+
+ @TaskAction
+ fun generate() =
+ outputDir.get().file("org/apache/airflow/sdk/slf4j/Constants.kt").asFile.apply {
+ parentFile.mkdirs()
+ writeText(
+ """
+ // Generated by Gradle — do not edit.
+ package org.apache.airflow.sdk.slf4j
+
+ internal const val REQUESTED_SLF4J_API_VERSION = "${requestedSlf4jApiVersion.get()}"
+ """.trimIndent() + "\n",
+ )
+ }
+}
+
+val generateSlf4jApiVersion by tasks.registering(GenerateConstants::class) {
+ description = "Generate source for constants pulled from Gradle properties."
+ requestedSlf4jApiVersion = slf4jVersion
+ outputDir = layout.buildDirectory.dir("generated/sources/slf4jApiVersion/kotlin/main")
+}
+
+kotlin.sourceSets.main {
+ kotlin.srcDir(generateSlf4jApiVersion.map { it.outputs.files.singleFile })
+}
+
+dependencies {
+ api("org.slf4j:slf4j-api:$slf4jVersion")
+ implementation(project(":sdk"))
+ testImplementation(kotlin("test"))
+ testImplementation(testFixtures(project(":sdk")))
+}
+
+java {
+ withSourcesJar() // Required by Maven Central.
+}
+
+tasks.withType {
+ useJUnitPlatform()
+}
+
+publishing {
+ publications {
+ create("mavenJava") {
+ artifactId = "airflow-sdk-slf4j"
+ from(components["java"])
+ pom {
+ name = "Apache Airflow Java SDK SLF4J Provider"
+ description = "SLF4J logging provider for the Apache Airflow Java SDK. " +
+ "Routes SLF4J log calls from task code through the SDK to Airflow's task log store."
+ }
+ }
+ }
+}
diff --git a/java-sdk/slf4j/gradle.properties b/java-sdk/slf4j/gradle.properties
new file mode 100644
index 0000000000000..15392279b8306
--- /dev/null
+++ b/java-sdk/slf4j/gradle.properties
@@ -0,0 +1,18 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+slf4jVersion=2.0.17
diff --git a/java-sdk/slf4j/src/main/kotlin/org/apache/airflow/sdk/slf4j/AirflowSlf4jProvider.kt b/java-sdk/slf4j/src/main/kotlin/org/apache/airflow/sdk/slf4j/AirflowSlf4jProvider.kt
new file mode 100644
index 0000000000000..f353204a1eef6
--- /dev/null
+++ b/java-sdk/slf4j/src/main/kotlin/org/apache/airflow/sdk/slf4j/AirflowSlf4jProvider.kt
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.airflow.sdk.slf4j
+
+import org.apache.airflow.sdk.execution.Level
+import org.apache.airflow.sdk.execution.Log
+import org.slf4j.ILoggerFactory
+import org.slf4j.IMarkerFactory
+import org.slf4j.Logger
+import org.slf4j.Marker
+import org.slf4j.helpers.AbstractLogger
+import org.slf4j.helpers.BasicMarkerFactory
+import org.slf4j.helpers.MessageFormatter
+import org.slf4j.helpers.NOPMDCAdapter
+import org.slf4j.spi.MDCAdapter
+import org.slf4j.spi.SLF4JServiceProvider
+import java.util.concurrent.ConcurrentHashMap
+import org.slf4j.event.Level as SLevel
+
+private fun SLevel.convert(): Level =
+ when (this) {
+ SLevel.TRACE -> Level.NOTSET
+ SLevel.DEBUG -> Level.DEBUG
+ SLevel.INFO -> Level.INFO
+ SLevel.WARN -> Level.WARNING
+ SLevel.ERROR -> Level.ERROR
+ }
+
+internal class AirflowSlf4jLogger(
+ name: String,
+) : AbstractLogger() {
+ init {
+ this.name = name
+ }
+
+ override fun getFullyQualifiedCallerName(): String? = null
+
+ override fun handleNormalizedLoggingCall(
+ level: SLevel,
+ marker: Marker?,
+ messagePattern: String?,
+ arguments: Array?,
+ throwable: Throwable?,
+ ) {
+ val event = MessageFormatter.basicArrayFormat(messagePattern ?: "", arguments)
+ Log.send(level.convert(), name, event) {
+ throwable?.run { put("exception", stackTraceToString()) }
+ marker?.let { put("marker", it.name) }
+ }
+ }
+
+ override fun isTraceEnabled(marker: Marker?) = Log.isEnabledForLevel(Level.NOTSET, name)
+
+ override fun isTraceEnabled() = Log.isEnabledForLevel(Level.NOTSET, name)
+
+ override fun isDebugEnabled(marker: Marker?) = Log.isEnabledForLevel(Level.DEBUG, name)
+
+ override fun isDebugEnabled() = Log.isEnabledForLevel(Level.DEBUG, name)
+
+ override fun isInfoEnabled(marker: Marker?) = Log.isEnabledForLevel(Level.INFO, name)
+
+ override fun isInfoEnabled() = Log.isEnabledForLevel(Level.INFO, name)
+
+ override fun isWarnEnabled(marker: Marker?) = Log.isEnabledForLevel(Level.WARNING, name)
+
+ override fun isWarnEnabled() = Log.isEnabledForLevel(Level.WARNING, name)
+
+ override fun isErrorEnabled(marker: Marker?) = Log.isEnabledForLevel(Level.ERROR, name)
+
+ override fun isErrorEnabled() = Log.isEnabledForLevel(Level.ERROR, name)
+}
+
+internal class AirflowLoggerFactory : ILoggerFactory {
+ private val loggers = ConcurrentHashMap()
+
+ override fun getLogger(name: String): Logger = loggers.computeIfAbsent(name) { AirflowSlf4jLogger(it) }
+}
+
+class AirflowSlf4jProvider : SLF4JServiceProvider {
+ private lateinit var factory: AirflowLoggerFactory
+
+ override fun getLoggerFactory(): ILoggerFactory = factory
+
+ override fun getMarkerFactory(): IMarkerFactory = BasicMarkerFactory()
+
+ override fun getMDCAdapter(): MDCAdapter = NOPMDCAdapter()
+
+ override fun getRequestedApiVersion() = REQUESTED_SLF4J_API_VERSION
+
+ override fun initialize() {
+ factory = AirflowLoggerFactory()
+ }
+}
diff --git a/java-sdk/slf4j/src/main/resources/META-INF/services/org.slf4j.spi.SLF4JServiceProvider b/java-sdk/slf4j/src/main/resources/META-INF/services/org.slf4j.spi.SLF4JServiceProvider
new file mode 100644
index 0000000000000..1cec01dbd96c1
--- /dev/null
+++ b/java-sdk/slf4j/src/main/resources/META-INF/services/org.slf4j.spi.SLF4JServiceProvider
@@ -0,0 +1,18 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+org.apache.airflow.sdk.slf4j.AirflowSlf4jProvider
diff --git a/java-sdk/slf4j/src/test/kotlin/org/apache/airflow/sdk/slf4j/AirflowSlf4jLoggerTest.kt b/java-sdk/slf4j/src/test/kotlin/org/apache/airflow/sdk/slf4j/AirflowSlf4jLoggerTest.kt
new file mode 100644
index 0000000000000..b2f366cf86350
--- /dev/null
+++ b/java-sdk/slf4j/src/test/kotlin/org/apache/airflow/sdk/slf4j/AirflowSlf4jLoggerTest.kt
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.airflow.sdk.slf4j
+
+import org.apache.airflow.sdk.execution.Level
+import org.apache.airflow.sdk.execution.LogCapture
+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.BeforeEach
+import org.junit.jupiter.api.Test
+import org.slf4j.MarkerFactory
+import org.slf4j.event.Level as SLevel
+
+class AirflowSlf4jLoggerTest {
+ private lateinit var logger: AirflowSlf4jLogger
+
+ @BeforeEach
+ fun setUp() {
+ logger = AirflowSlf4jLogger("com.example.Task")
+ LogCapture.resetThresholds()
+ LogCapture.drain() // discard any messages buffered before this test
+ }
+
+ @Test
+ fun `level conversions`() {
+ val cases =
+ listOf(
+ SLevel.TRACE to Level.NOTSET,
+ SLevel.DEBUG to Level.DEBUG,
+ SLevel.INFO to Level.INFO,
+ SLevel.WARN to Level.WARNING,
+ SLevel.ERROR to Level.ERROR,
+ )
+ cases.forEach { (slf4jLevel, expected) ->
+ LogCapture.drain()
+ when (slf4jLevel) {
+ SLevel.TRACE -> logger.trace("m")
+ SLevel.DEBUG -> logger.debug("m")
+ SLevel.INFO -> logger.info("m")
+ SLevel.WARN -> logger.warn("m")
+ SLevel.ERROR -> logger.error("m")
+ }
+ val messages = LogCapture.drain().filter { it.logger == "com.example.Task" }
+ assertEquals(1, messages.size, "Expected exactly one message for SLF4J $slf4jLevel")
+ assertEquals(expected, messages.single().level, "SLF4J $slf4jLevel should map to SDK $expected")
+ }
+ }
+
+ @Test
+ fun `message and logger name are forwarded`() {
+ logger.info("hello")
+ val msg = LogCapture.drain().single { it.logger == "com.example.Task" }
+ assertEquals(Level.INFO, msg.level)
+ assertEquals("com.example.Task", msg.logger)
+ assertEquals("hello", msg.event)
+ }
+
+ @Test
+ fun `message parameters are rendered into the message`() {
+ logger.info("{} {}", "alpha", 42 as Any)
+ val msg = LogCapture.drain().single { it.logger == "com.example.Task" }
+ assertEquals("alpha 42", msg.event)
+ assertFalse(msg.arguments.containsKey("0"))
+ }
+
+ @Test
+ fun `throwable is stored under the exception key`() {
+ val ex = RuntimeException("boom")
+ logger.error("oops", ex)
+ val msg = LogCapture.drain().single { it.logger == "com.example.Task" }
+ assertTrue(msg.arguments["exception"].toString().contains("boom"))
+ }
+
+ @Test
+ fun `marker name is stored under the marker key`() {
+ logger.info(MarkerFactory.getMarker("AUDIT"), "hello")
+ val msg = LogCapture.drain().single { it.logger == "com.example.Task" }
+ assertEquals("AUDIT", msg.arguments["marker"])
+ }
+
+ @Test
+ fun `no marker key is added when none is supplied`() {
+ logger.info("hello")
+ val msg = LogCapture.drain().single { it.logger == "com.example.Task" }
+ assertFalse(msg.arguments.containsKey("marker"))
+ }
+
+ @Test
+ fun `named DEBUG override enables debug while the global level stays INFO`() {
+ LogCapture.configureThresholds(Level.INFO, mapOf("com.example.Task" to Level.DEBUG))
+
+ // AbstractLogger gates debug() on isDebugEnabled(); the per-logger override must let it through.
+ assertTrue(logger.isDebugEnabled)
+ logger.debug("hello")
+
+ val messages = LogCapture.drain().filter { it.logger == "com.example.Task" }
+ assertEquals(1, messages.size, "named DEBUG override should let debug through")
+ assertEquals(Level.DEBUG, messages.single().level)
+
+ // A logger without an override still follows the global INFO threshold.
+ assertFalse(AirflowSlf4jLogger("com.other.Task").isDebugEnabled)
+ }
+}