This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new a1f1a491db1 Return null from Java SDK getVariable for missing
Variables (#75920)
a1f1a491db1 is described below
commit a1f1a491db1831aed7c82ae3a0f3aa74c58c30fe
Author: Aaron Chen <[email protected]>
AuthorDate: Sun Nov 1 15:34:07 2026 -0800
Return null from Java SDK getVariable for missing Variables (#75920)
* Return null from Java SDK getVariable for missing Variables
* Refactor error response handling in CommTest to use a unified
typedErrorFrame method
---
.../org/apache/airflow/e2e/TestBundleBuilder.java | 6 +-
.../main/kotlin/org/apache/airflow/sdk/Client.kt | 2 +-
.../org/apache/airflow/sdk/execution/Client.kt | 8 +-
.../org/apache/airflow/sdk/execution/CommTest.kt | 91 +++++++++++++---------
4 files changed, 65 insertions(+), 42 deletions(-)
diff --git
a/airflow-e2e-tests/java-test-bundle/src/java/org/apache/airflow/e2e/TestBundleBuilder.java
b/airflow-e2e-tests/java-test-bundle/src/java/org/apache/airflow/e2e/TestBundleBuilder.java
index 39c4bd46afc..342b1dd3354 100644
---
a/airflow-e2e-tests/java-test-bundle/src/java/org/apache/airflow/e2e/TestBundleBuilder.java
+++
b/airflow-e2e-tests/java-test-bundle/src/java/org/apache/airflow/e2e/TestBundleBuilder.java
@@ -51,7 +51,8 @@ public class TestBundleBuilder implements BundleBuilder {
/**
* Stores this run's id where the E2E test can read it back through the REST
API, then writes
- * and deletes a scratch variable to exercise the delete path.
+ * and deletes a scratch variable to exercise the delete path, and reads it
back to check that a
+ * missing variable reads as null.
*/
public static class WriteAndDeleteVariable implements Task {
public void execute(@NotNull Context context, Client client) {
@@ -59,6 +60,9 @@ public class TestBundleBuilder implements BundleBuilder {
"java_e2e_variable", context.dagRun.runId, "written by the Java SDK
e2e test");
client.setVariable("java_e2e_scratch", "scratch");
client.deleteVariable("java_e2e_scratch");
+ if (client.getVariable("java_e2e_scratch") != null) {
+ throw new IllegalStateException("java_e2e_scratch survived its
delete");
+ }
}
}
diff --git a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Client.kt
b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Client.kt
index d8cb35734fe..8b3311382b5 100644
--- a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Client.kt
+++ b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Client.kt
@@ -109,7 +109,7 @@ class Client internal constructor(
* @return The variable value, or `null` if the variable is not set.
* @throws ApiError if the API call fails.
*/
- fun getVariable(key: String): Any? = impl.getVariable(key).value
+ fun getVariable(key: String): Any? = impl.getVariable(key)?.value
/**
* Stores an Airflow variable, replacing any existing value.
diff --git
a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Client.kt
b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Client.kt
index 317dac7ebb0..7bcef266297 100644
--- a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Client.kt
+++ b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Client.kt
@@ -61,7 +61,8 @@ import java.util.UUID
interface Client {
fun getConnection(id: String): ConnectionResult
- fun getVariable(key: String): VariableResult
+ /** Returns `null` when the variable does not exist. */
+ fun getVariable(key: String): VariableResult?
fun setVariable(
key: String,
@@ -164,7 +165,10 @@ class CoordinatorClient(
override fun getVariable(key: String) =
runBlocking {
- exec.communicate<VariableResult>(GetVariable().also { it.key = key })
+ exec.communicateOrNullIf<VariableResult>(
+ GetVariable().also { it.key = key },
+ ErrorResponse.ErrorType.VARIABLE_NOT_FOUND,
+ )
}
override fun setVariable(
diff --git
a/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/execution/CommTest.kt
b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/execution/CommTest.kt
index b6c12c36a57..68b4efa38dc 100644
--- a/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/execution/CommTest.kt
+++ b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/execution/CommTest.kt
@@ -183,24 +183,30 @@ class CommsTest {
return out.toByteArray()
}
- private fun notFoundResponseFrame(id: Int): ByteArray {
- val out = ByteArrayOutputStream()
- MessagePack.newDefaultPacker(out).use { packer ->
- packer.packArrayHeader(3)
- packer.packInt(id)
- packer.packNil()
- packer.packMapHeader(3)
- packer.packString("type")
- packer.packString("ErrorResponse")
- packer.packString("error")
- packer.packString("TASK_STORE_NOT_FOUND")
- packer.packString("detail")
- packer.packMapHeader(1)
- packer.packString("key")
- packer.packString("job_id")
+ // An ErrorResponse carrying a typed `error` code and a one-entry `detail`,
as the supervisor sends it.
+ private fun typedErrorFrame(
+ error: String,
+ detailKey: String,
+ detailValue: String,
+ ): (Int) -> ByteArray =
+ { id ->
+ val out = ByteArrayOutputStream()
+ MessagePack.newDefaultPacker(out).use { packer ->
+ packer.packArrayHeader(3)
+ packer.packInt(id)
+ packer.packNil()
+ packer.packMapHeader(3)
+ packer.packString("type")
+ packer.packString("ErrorResponse")
+ packer.packString("error")
+ packer.packString(error)
+ packer.packString("detail")
+ packer.packMapHeader(1)
+ packer.packString(detailKey)
+ packer.packString(detailValue)
+ }
+ out.toByteArray()
}
- return out.toByteArray()
- }
private fun taskStateStoreResultFrame(id: Int): ByteArray {
val out = ByteArrayOutputStream()
@@ -282,25 +288,6 @@ class CommsTest {
return Triple(requests.single().rawBody as Map<*, *>, outcome.getOrNull(),
outcome.exceptionOrNull())
}
- private fun dagRunExistsResponseFrame(id: Int): ByteArray {
- val out = ByteArrayOutputStream()
- MessagePack.newDefaultPacker(out).use { packer ->
- packer.packArrayHeader(3)
- packer.packInt(id)
- packer.packNil()
- packer.packMapHeader(3)
- packer.packString("type")
- packer.packString("ErrorResponse")
- packer.packString("error")
- packer.packString("DAGRUN_ALREADY_EXISTS")
- packer.packString("detail")
- packer.packMapHeader(1)
- packer.packString("dag_id")
- packer.packString("downstream")
- }
- return out.toByteArray()
- }
-
private fun dagRunStateResultFrame(id: Int): ByteArray {
val out = ByteArrayOutputStream()
MessagePack.newDefaultPacker(out).use { packer ->
@@ -491,7 +478,7 @@ class CommsTest {
@Timeout(value = 30, unit = TimeUnit.SECONDS)
fun triggerDagRunReportsAnExistingRun() {
val (_, alreadyExists, failure) =
- transportRoundTrip(::dagRunExistsResponseFrame) {
+ transportRoundTrip(typedErrorFrame("DAGRUN_ALREADY_EXISTS", "dag_id",
"downstream")) {
it.triggerDagRun("downstream", "run", null, null, null, false, null)
}
@@ -525,6 +512,31 @@ class CommsTest {
Assertions.assertEquals("downstream", body["dag_id"])
}
+ @Test
+ @DisplayName("getVariable returns null on VARIABLE_NOT_FOUND instead of
raising")
+ @Timeout(value = 30, unit = TimeUnit.SECONDS)
+ fun getVariableReturnsNullWhenNotFound() {
+ var value: Any? = "unset"
+ val (body, failure) =
+ roundTrip(typedErrorFrame("VARIABLE_NOT_FOUND", "key",
"optional_setting")) {
+ value = it.getVariable("optional_setting")
+ }
+
+ Assertions.assertNull(failure, "getVariable should return normally on
VARIABLE_NOT_FOUND, got $failure")
+ Assertions.assertEquals("GetVariable", body["type"])
+ Assertions.assertEquals("optional_setting", body["key"])
+ Assertions.assertNull(value)
+ }
+
+ @Test
+ @DisplayName("getVariable raises ApiError for any other error response")
+ @Timeout(value = 30, unit = TimeUnit.SECONDS)
+ fun getVariableRaisesApiErrorOnOtherErrors() {
+ val (_, failure) = roundTrip(::errorResponseFrame) {
it.getVariable("optional_setting") }
+
+ Assertions.assertInstanceOf(ApiError::class.java, failure)
+ }
+
@Test
@DisplayName("setVariable keeps a null description on the wire so the
supervisor accepts the request")
@Timeout(value = 30, unit = TimeUnit.SECONDS)
@@ -592,7 +604,10 @@ class CommsTest {
@Timeout(value = 30, unit = TimeUnit.SECONDS)
fun taskStateStoreGetReturnsNullWhenNotFound() {
var value: Any? = "unset"
- val (_, failure) = roundTrip(::notFoundResponseFrame,
currentTaskInstance()) { value = it.taskStateStore.get("job_id") }
+ val (_, failure) =
+ roundTrip(typedErrorFrame("TASK_STORE_NOT_FOUND", "key", "job_id"),
currentTaskInstance()) {
+ value = it.taskStateStore.get("job_id")
+ }
Assertions.assertNull(failure, "get should return normally on
TASK_STORE_NOT_FOUND, got $failure")
Assertions.assertNull(value)
