jason810496 commented on code in PR #68725: URL: https://github.com/apache/airflow/pull/68725#discussion_r3449593950
########## java-sdk/log4j2/src/main/java/org/apache/airflow/sdk/log4j/AirflowLog4jAppender.java: ########## @@ -0,0 +1,92 @@ +/* + * 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.Collections; +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.PluginFactory; +import org.apache.logging.log4j.spi.StandardLevel; + +/** + * A Log4j {@link Appender} to route logs to Airflow. + * + * <p>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() { + return new AirflowLog4jAppender("AirflowAppender", null); + } Review Comment: `createAppender()` takes no arguments and hardcodes the name as `"AirflowAppender"`, with no `@PluginAttribute("name")`. Log4j2's `AppendersPlugin` keys the appender map by `appender.getName()`, and `AppenderRef` resolves against that key. The documented `log4j2.xml` uses `<AirflowAppender name="Airflow"/>` with `<AppenderRef ref="Airflow"/>`, but this appender always registers as `"AirflowAppender"`, so `ref="Airflow"` won't resolve — Log4j logs "Unable to locate appender" and drops it. As written, the example routes nothing to Airflow. Either accept the configured name (and import `PluginAttribute` / `PluginElement`): ```java @PluginFactory public static AirflowLog4jAppender createAppender( @PluginAttribute("name") String name, @PluginElement("Filter") Filter filter) { return new AirflowLog4jAppender(name == null ? "AirflowAppender" : name, filter); } ``` or change the docs to `name="AirflowAppender"` / `ref="AirflowAppender"`. The unit test calls `createAppender()` directly and never builds the appender from XML, so this path is currently untested. ########## airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst: ########## @@ -230,6 +230,159 @@ See the Java SDK's published JavaDoc for more details. .. TODO: (AIP-108) Put a link here once we publish the JavaDoc. +.. _java-sdk/logging: + +Logging +------- + +Task code can emit log records through any common Java logging framework. The SDK ships optional +integration libraries that forward those records to Airflow's task log store, where they appear +alongside the standard task output in the Airflow UI. + +Declare a logger as a static field on the task class, using the class's own type as the name. This +is the conventional pattern regardless of which logging framework you choose: + +.. code-block:: java + + private static final System.Logger log = + System.getLogger(SalesPipeline.class.getName()); + + @Builder.Task(id = "extract") + public long extract(Client client) { + log.log(System.Logger.Level.INFO, "Starting extraction"); + return recordCount; + } + +The Gradle snippets below show the dependency declarations; all Airflow artifact versions are managed +by ``airflow-sdk-bom``. Maven users apply the same artifact IDs following the pattern in +:ref:`java-sdk/build/maven`. + +.. _java-sdk/logging/jpl: + +``System.Logger`` (Java Platform Logging) +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +Java 9's new logging facade ``java.lang.System.Logger`` (JEP 264), commonly abbreviated *JPL*, can be +used by libraries without pulling in any third-party API. The ``airflow-sdk-jpl`` artifact registers an +``AirflowSystemLoggerFinder`` via ``ServiceLoader``, which routes all ``System.Logger`` calls directly +to Airflow's task log store. + +.. code-block:: groovy + + implementation("org.apache.airflow:airflow-sdk-jpl:${version}") + +No configuration file or startup call is required. The ``ServiceLoader`` mechanism discovers the +provider automatically as long as the JAR is on the classpath. + +.. note:: + + Do not add a second ``System.LoggerFinder`` implementation alongside + ``airflow-sdk-jpl``. The JVM selects one finder via ``ServiceLoader``; having + multiple providers on the classpath leads to unpredictable behaviour. + +.. _java-sdk/logging/slf4j: + +SLF4J 2.x +~~~~~~~~~ + +The SLF4J binding is discovered automatically via ``ServiceLoader``; no configuration file or +startup call is required. + +.. code-block:: groovy + + implementation("org.apache.airflow:airflow-sdk-slf4j:${version}") + +The above automatically pulls in the SLF4J API, so you don't need to add ``slf4j-api`` yourself. + +.. note:: + + Do not add a second SLF4J binding (such as ``logback-classic`` or ``slf4j-simple``) alongside + ``airflow-sdk-slf4j``. SLF4J 2.x warns about multiple bindings and selects one unpredictably. + +.. _java-sdk/logging/log4j2: + +Log4j 2 +~~~~~~~ + +``airflow-sdk-log4j2`` declares ``log4j-api`` as a transitive dependency, so you do not need to add the latter +separately. You must also place ``log4j-core``on the runtime classpath to host the plugin loader that Review Comment: Missing space: the `log4j-core` literal is glued to the following word ("...core``on..."). The suggestion inserts the space. ```suggestion separately. You must also place ``log4j-core`` on the runtime classpath to host the plugin loader that ``` ########## java-sdk/jpl/src/main/kotlin/org/apache/airflow/sdk/jpl/AirflowSystemLoggerFinder.kt: ########## @@ -0,0 +1,91 @@ +/* + * 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.jpl + +import org.apache.airflow.sdk.execution.Level +import org.apache.airflow.sdk.execution.Log +import java.util.MissingResourceException +import java.util.ResourceBundle +import kotlin.collections.set + +/** + * [System.LoggerFinder] that routes [System.Logger] calls to the Airflow task log store. + * + * Registered via [META-INF/services/java.lang.System$LoggerFinder][System.LoggerFinder]. + * All loggers share a single implementation regardless of the requesting module. + */ +class AirflowSystemLoggerFinder : System.LoggerFinder() { + override fun getLogger( + name: String, + module: Module, + ): System.Logger = AirflowSystemLogger(name) +} + +internal class AirflowSystemLogger( + private val name: String, +) : System.Logger { + override fun getName(): String = name + + override fun isLoggable(level: System.Logger.Level): Boolean = + level != System.Logger.Level.OFF && Log.isEnabledForLevel(level.convert(), name) + + override fun log( + level: System.Logger.Level, + bundle: ResourceBundle?, + msg: String?, + thrown: Throwable, + ) { + if (!isLoggable(level)) return + Log.send(level.convert(), name, bundle.resolve(msg)) { + put("exception", thrown.toString()) Review Comment: Two issues in this overload: 1. The exception is stored as `thrown.toString()` (class + message only), dropping the stack trace. The other three providers capture the full trace — JUL/SLF4J use `stackTraceToString()`, Log4j2 uses `printStackTrace`. 2. The JDK contract for `System.Logger.log(Level, ResourceBundle, String, Throwable)` allows `thrown == null`; the non-null Kotlin param inserts a null-check intrinsic that would NPE on an explicit null. Low likelihood (normal calls route to the params overload), but cheap to harden. Both addressed together: ```suggestion thrown: Throwable?, ) { if (!isLoggable(level)) return Log.send(level.convert(), name, bundle.resolve(msg)) { put("exception", thrown?.stackTraceToString() ?: "") ``` ########## java-sdk/slf4j/src/main/kotlin/org/apache/airflow/sdk/slf4j/AirflowSlf4jProvider.kt: ########## @@ -0,0 +1,111 @@ +/* + * 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.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<out Any?>?, + throwable: Throwable?, + ) { + // Since the Python side is using a structlog pattern, let's just send the message pattern as-is + // with unrendered placeholders and put all arguments under keys "0", "1", "2" and so on. + // If there's an error attached, put it (as string) under "exception" like how structlog does it. + Log.send(level.convert(), name, messagePattern ?: "") { + arguments?.forEachIndexed { i, v -> put(i.toString(), v) } + throwable?.run { put("exception", stackTraceToString()) } + } + } + + override fun isTraceEnabled(marker: Marker?) = Log.isEnabledForLevel(Level.NOTSET, marker?.name) + + override fun isTraceEnabled() = Log.isEnabledForLevel(Level.NOTSET, null) + + override fun isDebugEnabled(marker: Marker?) = Log.isEnabledForLevel(Level.DEBUG, marker?.name) + + override fun isDebugEnabled() = Log.isEnabledForLevel(Level.DEBUG, null) + + override fun isInfoEnabled(marker: Marker?) = Log.isEnabledForLevel(Level.INFO, marker?.name) + + override fun isInfoEnabled() = Log.isEnabledForLevel(Level.INFO, null) + + override fun isWarnEnabled(marker: Marker?) = Log.isEnabledForLevel(Level.WARNING, marker?.name) + + override fun isWarnEnabled() = Log.isEnabledForLevel(Level.WARNING, null) + + override fun isErrorEnabled(marker: Marker?) = Log.isEnabledForLevel(Level.ERROR, marker?.name) + + override fun isErrorEnabled() = Log.isEnabledForLevel(Level.ERROR, null) Review Comment: These `is*Enabled` overrides feed the wrong key into `Log.isEnabledForLevel`, so per-logger overrides set via `AIRFLOW__LOGGING__NAMESPACE_LEVELS` do not take effect for SLF4J: - The no-marker variants pass `null`, so they always resolve to `globalThreshold` and never consult this logger's named threshold. - The marker variants pass `marker?.name` (the marker's name), not this logger's `name`. `AbstractLogger.debug(msg)` (and the other level methods) evaluate `isDebugEnabled()` *before* dispatching to `handleNormalizedLoggingCall`. So with global `INFO` and `AIRFLOW__LOGGING__NAMESPACE_LEVELS=com.example.Task=DEBUG`, `logger.debug(...)` on `com.example.Task` is rejected by the guard and never reaches the `Log.send` re-check that does use the real `name`. Net effect: named DEBUG overrides are silently ignored for SLF4J-backed tasks, while JUL / Log4j2 / JPL pass the real logger name and behave correctly. Suggest using this logger's `name` in the no-marker checks (e.g. `Log.isEnabledForLevel(Level.DEBUG, name)`), defining how marker filtering should compose with logger filtering, and adding a test where a named DEBUG threshold enables debug while the global threshold stays INFO. -- 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]
