You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "FrankYang0529 (via GitHub)" <gi...@apache.org> on 2026/06/27 08:29:06 UTC
[PR] Java SDK: Serialize coordinator client send/receive for thread safety [airflow]
FrankYang0529 opened a new pull request, #69080:
URL: https://github.com/apache/airflow/pull/69080
Guard the whole send/receive pair in `communicateImpl` with a `Mutex` so concurrent task threads are serialized over the single channel, and verify the response id echoes the request id to reject a mis-paired response.
<!-- SPDX-License-Identifier: Apache-2.0
https://www.apache.org/licenses/LICENSE-2.0 -->
<!--
Thank you for contributing!
Please provide above a brief description of the changes made in this pull request.
Write a good git commit message following this guide: http://chris.beams.io/posts/git-commit/
Please make sure that your code changes are covered with tests.
And in case of new features or big changes remember to adjust the documentation.
For user-facing UI changes, please attach before/after screenshots (or a short
screen recording) so reviewers can assess the visual impact.
Feel free to ping (in general) for the review if you do not see reaction for a few days
(72 Hours is the minimum reaction time you can expect from volunteers) - we sometimes miss notifications.
In case of an existing issue, reference it using one of the following:
* closes: #ISSUE
* related: #ISSUE
-->
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR, please
change below checkbox to `[X]` followed by the name of the tool, uncomment the "Generated-by".
-->
- [X] Yes - Claude Code with Opus 4.8
<!--
Generated-by: [Tool Name] following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
-->
---
* Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
* For fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment: `{pr_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments). You can add this file in a follow-up commit after the PR is created so you know the PR number.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Java SDK: Serialize coordinator client send/receive for thread safety [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #69080:
URL: https://github.com/apache/airflow/pull/69080#discussion_r3489008734
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Comm.kt:
##########
@@ -109,19 +112,22 @@ class CoordinatorComm(
}
@Throws(ApiError::class)
- suspend fun communicateImpl(body: Any): Any {
- var frame: IncomingFrame? = null
+ suspend fun communicateImpl(body: Any): Any =
+ commMutex.withLock {
+ val requestId = nextId.fetchAndAdd(1)
+ var frame: IncomingFrame? = null
- suspend fun handle(f: IncomingFrame) {
- frame = f
- }
- sendMessage(nextId.fetchAndAdd(1), body)
- processOnce(::handle)
- if (frame == null) {
- throw ApiError("No response received")
+ suspend fun handle(f: IncomingFrame) {
+ frame = f
+ }
+ sendMessage(requestId, body)
+ processOnce(::handle)
+ val received = frame ?: throw ApiError("No response received")
+ if (received.id != requestId) {
+ throw ApiError("response id ${received.id} does not match request id $requestId")
+ }
+ received.body ?: Unit
Review Comment:
But for Python Task execution side, we're fully serializing the request. So it's fine to have current patch here instead of the mutiplexing approach IMO.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Java SDK: Serialize coordinator client send/receive for thread safety [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #69080:
URL: https://github.com/apache/airflow/pull/69080#discussion_r3488029977
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Comm.kt:
##########
@@ -109,19 +112,22 @@ class CoordinatorComm(
}
@Throws(ApiError::class)
- suspend fun communicateImpl(body: Any): Any {
- var frame: IncomingFrame? = null
+ suspend fun communicateImpl(body: Any): Any =
+ commMutex.withLock {
+ val requestId = nextId.fetchAndAdd(1)
Review Comment:
I think this should be outside of the lock since `nextId` handles concurrency itself?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Java SDK: Serialize coordinator client send/receive for thread safety [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 merged PR #69080:
URL: https://github.com/apache/airflow/pull/69080
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Java SDK: Serialize coordinator client send/receive for thread safety [airflow]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #69080:
URL: https://github.com/apache/airflow/pull/69080#discussion_r3490115362
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Comm.kt:
##########
@@ -109,19 +112,22 @@ class CoordinatorComm(
}
@Throws(ApiError::class)
- suspend fun communicateImpl(body: Any): Any {
- var frame: IncomingFrame? = null
+ suspend fun communicateImpl(body: Any): Any =
+ commMutex.withLock {
+ val requestId = nextId.fetchAndAdd(1)
+ var frame: IncomingFrame? = null
- suspend fun handle(f: IncomingFrame) {
- frame = f
- }
- sendMessage(nextId.fetchAndAdd(1), body)
- processOnce(::handle)
- if (frame == null) {
- throw ApiError("No response received")
+ suspend fun handle(f: IncomingFrame) {
+ frame = f
+ }
+ sendMessage(requestId, body)
+ processOnce(::handle)
+ val received = frame ?: throw ApiError("No response received")
+ if (received.id != requestId) {
+ throw ApiError("response id ${received.id} does not match request id $requestId")
+ }
+ received.body ?: Unit
Review Comment:
Yes, Python Task also uses lock.
https://github.com/apache/airflow/blob/fc5f0441034176e2db333885855957e051cab5eb/task-sdk/src/airflow/sdk/execution_time/comms.py#L217-L223
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Java SDK: Serialize coordinator client send/receive for thread safety [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #69080:
URL: https://github.com/apache/airflow/pull/69080#discussion_r3488984282
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Comm.kt:
##########
@@ -109,19 +112,22 @@ class CoordinatorComm(
}
@Throws(ApiError::class)
- suspend fun communicateImpl(body: Any): Any {
- var frame: IncomingFrame? = null
+ suspend fun communicateImpl(body: Any): Any =
+ commMutex.withLock {
+ val requestId = nextId.fetchAndAdd(1)
+ var frame: IncomingFrame? = null
- suspend fun handle(f: IncomingFrame) {
- frame = f
- }
- sendMessage(nextId.fetchAndAdd(1), body)
- processOnce(::handle)
- if (frame == null) {
- throw ApiError("No response received")
+ suspend fun handle(f: IncomingFrame) {
+ frame = f
+ }
+ sendMessage(requestId, body)
+ processOnce(::handle)
+ val received = frame ?: throw ApiError("No response received")
+ if (received.id != requestId) {
+ throw ApiError("response id ${received.id} does not match request id $requestId")
+ }
+ received.body ?: Unit
Review Comment:
Yes, the Triggerer and Go-SDK side implemented I/O multiplexing instead of fully serializing the requests.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Java SDK: Serialize coordinator client send/receive for thread safety [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #69080:
URL: https://github.com/apache/airflow/pull/69080#discussion_r3489020889
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Comm.kt:
##########
@@ -109,19 +112,22 @@ class CoordinatorComm(
}
@Throws(ApiError::class)
- suspend fun communicateImpl(body: Any): Any {
- var frame: IncomingFrame? = null
+ suspend fun communicateImpl(body: Any): Any =
+ commMutex.withLock {
+ val requestId = nextId.fetchAndAdd(1)
Review Comment:
Yes, the nextID itself is concurrent-safe. IIRC we didn't use the "id" for ordering, we only treat it as identifier so I think we can move outside of the lock.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Java SDK: Serialize coordinator client send/receive for thread safety [airflow]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #69080:
URL: https://github.com/apache/airflow/pull/69080#discussion_r3490105125
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Comm.kt:
##########
@@ -109,19 +112,22 @@ class CoordinatorComm(
}
@Throws(ApiError::class)
- suspend fun communicateImpl(body: Any): Any {
- var frame: IncomingFrame? = null
+ suspend fun communicateImpl(body: Any): Any =
+ commMutex.withLock {
+ val requestId = nextId.fetchAndAdd(1)
Review Comment:
Move `requestId` outside the lock. Thank you.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Java SDK: Serialize coordinator client send/receive for thread safety [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #69080:
URL: https://github.com/apache/airflow/pull/69080#discussion_r3496550842
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Comm.kt:
##########
@@ -109,19 +112,22 @@ class CoordinatorComm(
}
@Throws(ApiError::class)
- suspend fun communicateImpl(body: Any): Any {
- var frame: IncomingFrame? = null
+ suspend fun communicateImpl(body: Any): Any =
+ commMutex.withLock {
+ val requestId = nextId.fetchAndAdd(1)
+ var frame: IncomingFrame? = null
- suspend fun handle(f: IncomingFrame) {
- frame = f
- }
- sendMessage(nextId.fetchAndAdd(1), body)
- processOnce(::handle)
- if (frame == null) {
- throw ApiError("No response received")
+ suspend fun handle(f: IncomingFrame) {
+ frame = f
+ }
+ sendMessage(requestId, body)
+ processOnce(::handle)
+ val received = frame ?: throw ApiError("No response received")
+ if (received.id != requestId) {
+ throw ApiError("response id ${received.id} does not match request id $requestId")
+ }
+ received.body ?: Unit
Review Comment:
I’m not saying we shouldn’t use locks, but if we do multiplexing we should be able to reduce the locking surface quite a bit, essentially only needs to lock during send and during receive, not the entire communication (serialize-send-receive-deserialize) process.
This can be explored in a different PR if needed.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Java SDK: Serialize coordinator client send/receive for thread safety [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #69080:
URL: https://github.com/apache/airflow/pull/69080#discussion_r3496550842
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Comm.kt:
##########
@@ -109,19 +112,22 @@ class CoordinatorComm(
}
@Throws(ApiError::class)
- suspend fun communicateImpl(body: Any): Any {
- var frame: IncomingFrame? = null
+ suspend fun communicateImpl(body: Any): Any =
+ commMutex.withLock {
+ val requestId = nextId.fetchAndAdd(1)
+ var frame: IncomingFrame? = null
- suspend fun handle(f: IncomingFrame) {
- frame = f
- }
- sendMessage(nextId.fetchAndAdd(1), body)
- processOnce(::handle)
- if (frame == null) {
- throw ApiError("No response received")
+ suspend fun handle(f: IncomingFrame) {
+ frame = f
+ }
+ sendMessage(requestId, body)
+ processOnce(::handle)
+ val received = frame ?: throw ApiError("No response received")
+ if (received.id != requestId) {
+ throw ApiError("response id ${received.id} does not match request id $requestId")
+ }
+ received.body ?: Unit
Review Comment:
I’m not saying we shouldn’t use locks, but if we do multiplexing we should be able to reduce the locking surface quite a bit, essentially only needs to lock during send and during receive, not the entire communication (serialize-send-receive-deserialize) process.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Java SDK: Serialize coordinator client send/receive for thread safety [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #69080:
URL: https://github.com/apache/airflow/pull/69080#discussion_r3488031939
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Comm.kt:
##########
@@ -109,19 +112,22 @@ class CoordinatorComm(
}
@Throws(ApiError::class)
- suspend fun communicateImpl(body: Any): Any {
- var frame: IncomingFrame? = null
+ suspend fun communicateImpl(body: Any): Any =
+ commMutex.withLock {
+ val requestId = nextId.fetchAndAdd(1)
+ var frame: IncomingFrame? = null
- suspend fun handle(f: IncomingFrame) {
- frame = f
- }
- sendMessage(nextId.fetchAndAdd(1), body)
- processOnce(::handle)
- if (frame == null) {
- throw ApiError("No response received")
+ suspend fun handle(f: IncomingFrame) {
+ frame = f
+ }
+ sendMessage(requestId, body)
+ processOnce(::handle)
+ val received = frame ?: throw ApiError("No response received")
+ if (received.id != requestId) {
+ throw ApiError("response id ${received.id} does not match request id $requestId")
+ }
+ received.body ?: Unit
Review Comment:
I wonder if there’s a way to do though outside of a lock since we have the request ID to identify which response if for which. IIRC the Python implementation does this.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org