jason810496 commented on code in PR #69827:
URL: https://github.com/apache/airflow/pull/69827#discussion_r3608400217
##########
java-sdk/processor/src/test/kotlin/org/apache/airflow/sdk/BuilderTest.kt:
##########
@@ -297,4 +297,21 @@ class BuilderTest {
"Cannot create task from vararg function t1",
)
}
+
+ @Test
+ @DisplayName("fail compilation when dag class is a non-static inner class")
+ fun failCompilationWhenDagClassIsNonStaticInner() {
+ val compilation =
+ compile(
+ """
+ package org.apache.airflow.example;
+ import org.apache.airflow.sdk.Builder;
+ public class TestExample {
+ @Builder.Dag
+ public class InnerDag { @Builder.Task(id = "foo") public void t1()
{} }
+ }
+ """,
+ )
+ assertThat(compilation).failed()
Review Comment:
It would be nice to assert the error message with `.hadErrorContaining(` as
well. Thanks.
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Task.kt:
##########
@@ -71,17 +73,22 @@ internal object TaskRunner {
client: Client,
): Any {
val task = bundle.dags[request.ti.dagId]?.tasks[request.ti.taskId] ?:
return TaskResult.of(TaskState.State.REMOVED)
+ val instance =
+ try {
+ task.getDeclaredConstructor().newInstance()
+ } catch (e: Throwable) {
+ logger.error(
+ "Cannot instantiate task class. A task class must be concrete and
declare a public no-argument constructor",
+ mapOf("ti" to request.ti, "taskClass" to task.name, "error" to e,
"trace" to e.stackTraceToString()),
+ )
+ return TaskResult.failure(request.tiContext.shouldRetry)
+ }
Review Comment:
I think we need to handle the error raise by the constructor body itself
separately with different helper message.
```suggestion
val instance =
try {
task.getDeclaredConstructor().newInstance()
} catch (e: InvocationTargetException) {
val cause = e.cause ?: e
logger.error(
"Task class constructor threw an exception",
mapOf("ti" to request.ti, "taskClass" to task.name, "error" to
cause, "trace" to cause.stackTraceToString()),
)
return TaskResult.failure(request.tiContext.shouldRetry)
} catch (e: Throwable) {
logger.error(
"Cannot instantiate task class. A task class must be concrete and
declare a public no-argument constructor",
mapOf("ti" to request.ti, "taskClass" to task.name, "error" to e,
"trace" to e.stackTraceToString()),
)
return TaskResult.failure(request.tiContext.shouldRetry)
}
```
##########
airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst:
##########
@@ -218,11 +228,21 @@ read.
}
}
-Register tasks manually in a ``BundleBuilder``:
+Register tasks manually in a ``BundleBuilder``. A task class can be top-level
like ``FetchTask``, or
+nested ``static`` class like ``ProcessTask``:
.. code-block:: java
public class MyBundle implements BundleBuilder {
+ public static class ProcessTask implements Task {
+ @Override
+ public void execute(Context context, Client client) throws Exception {
+ var fetched = (String) client.getXCom("fetch");
+ // implement task logic
+ client.setXCom(count);
Review Comment:
IIUC, do we mean `fetched` variable here?
```suggestion
client.setXCom(fetched);
```
--
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]
