FrankYang0529 · Oct 5, 2026
@@ -165,8 +165,11 @@ class Client internal constructor(
    * Pushes an XCom value for downstream tasks to read.
    *
    * @param key XCom key; defaults to [XCOM_RETURN_KEY].
-   * @param value Value to push. Must be JSON-serializable.
+   * @param value Value to push. Must be JSON-serializable. The SDK rejects byte
+   *    arrays, so encode binary data as a string first, for example with Base64.
    * @throws ApiError if the API call fails.
+   * @throws IllegalArgumentException if [value] is or contains a type that the
+   *    SDK cannot send, such as a byte array.
    */
   @JvmOverloads fun setXCom(
     key: String = XCOM_RETURN_KEY,

@@ -41,11 +41,6 @@ import java.math.BigInteger
 import java.time.OffsetDateTime
 import java.time.ZoneOffset
 
-private fun MessagePacker.packByteArray(data: ByteArray) {
-  packBinaryHeader(data.size)
-  data.forEach { packByte(it) }
-}
-
 private fun MessagePacker.packMap(data: Map<*, *>) {
   packMapHeader(data.size)
   data.forEach { (k, v) ->
@@ -71,7 +66,13 @@ fun MessagePacker.packAny(data: Any?) {
     is BigInteger -> packBigInteger(data)
     is Float -> packFloat(data)
     is Double -> packDouble(data)
-    is ByteArray -> packByteArray(data)
+    // The supervisor rejects bytes in values such as XCom values. It does not reply
+    // to a request it cannot decode, so the task that sent the request would wait forever.
+    is ByteArray ->
+      throw IllegalArgumentException(
+        "Airflow does not accept byte arrays because they are not JSON values. " +
+          "Encode the bytes as a string first, for example with java.util.Base64.",
+      )
     is String -> packString(data)
     is Map<*, *> -> packMap(data)
     is Collection<*> -> packCollection(data)

@@ -20,6 +20,7 @@
 package org.apache.airflow.sdk.execution
 
 import io.ktor.utils.io.ByteChannel
+import io.ktor.utils.io.exhausted
 import io.ktor.utils.io.readByteArray
 import io.ktor.utils.io.writeByteArray
 import kotlinx.coroutines.async
@@ -386,6 +387,39 @@ class CommsTest {
     Assertions.assertInstanceOf(ApiError::class.java, failure)
   }
 
+  @Test
+  @DisplayName("setXCom rejects a byte array nested in the value without writing anything to the supervisor")
+  @Timeout(value = 30, unit = TimeUnit.SECONDS)
+  fun setXComRejectsByteArrayBeforeWriting() {
+    val toClient = ByteChannel(autoFlush = true)
+    val fromClient = ByteChannel(autoFlush = true)
+    val comm = CoordinatorComm(toClient, fromClient)
+    val details =
+      StartupDetails().also {
+        it.ti =
+          TaskInstance().also { ti ->
+            ti.dagId = "d"
+            ti.taskId = "t"
+            ti.runId = "r"
+          }
+      }
+    val client = PublicClient(details, CoordinatorClient(comm))
+    // Answer the first request in advance. If setXCom sent the byte array anyway,
+    // this reply would let setXCom return. Then assertThrows would fail at once
+    // instead of the test waiting for its 30-second timeout.
+    runBlocking { toClient.writeFrame(emptyResponseFrame(0)) }
+
+    val error =
+      Assertions.assertThrows(IllegalArgumentException::class.java) {
+        client.setXCom(value = mapOf("blob" to byteArrayOf(1)))
+      }
+
+    Assertions.assertTrue("Base64" in error.message.orEmpty(), "the error should suggest an encoding: $error")
+    runBlocking { fromClient.flushAndClose() }
+    Assertions.assertTrue(runBlocking { fromClient.exhausted() }, "nothing should be written to the supervisor")
+    comm.close()
+  }
+
   @Test
   @DisplayName("Should fail a pending call when the coordinator socket closes")
   @Timeout(value = 30, unit = TimeUnit.SECONDS)