sangkyoonnam opened a new pull request, #29538:
URL: https://github.com/apache/flink/pull/29538
## What is the purpose of the change
This change fixes a shutdown race that can make
`CommonDataStreamTests.test_logging_msg` fail with `external log debug-level
msg not found`. The same race can drop application logs outside tests.
Before this change, `WrappedSdkHarnessClient.close()` closes the logging
server before the control server. The Python SDK harness sends queued log
entries on a background thread and drains the rest in
`FnApiLogRecordHandler.close()` after the control stream ends. By then the
logging stream is already closed, so entries the worker hasn't sent yet are
lost.
## Brief change log
- Close the control server first, so the Python SDK harness leaves its
control loop and drains its log queue.
- Stop accepting new logging calls and wait up to 1 s per environment for
the open logging stream to end, then close the logging server and the remaining
resources in their existing order. Entries still queued after 1 s can be lost,
as before.
[Beam's own
copy](https://github.com/apache/beam/blob/v2.75.0/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/control/DefaultJobBundleFactory.java)
also closes logging first, with a comment that this avoids error messages
during shutdown. This change accepts that noise so pending logs arrive. When I
cancelled a job with a bundle in flight, the harness's `Python sdk harness
failed` traceback (`Runner closed connection`) reached the TaskManager log only
with the change. Both versions reached `CANCELED`.
## Verifying this change
This change added tests and can be verified as follows:
- Added two tests to `DefaultJobBundleFactoryTest` with a fake harness on
real gRPC streams. One sends a log entry after the control stream ends and
checks that it is written; the other leaves its logging stream open and checks
that the runner ends it. Without the fix, only the first test failed, in 5 of 5
runs. With the fix, all three tests in the class passed in 5 of 5 runs.
- `mvn -o -pl flink-python verify` passes with JDK 17.
- I compiled the same change into the Java JAR of a PyFlink 2.3.0
installation and ran a copy of `test_logging_msg` with each log batch carrying
the test's messages delayed by 50 ms. Without the change both messages were
missing right after the job and still missing 20 s later; with it both arrived.
With a 3 s delay the logging stream ended about 1 s after the harness started
shutting down its log handler, and the delayed messages were lost.
The change can be backported to `release-2.3`, where the failure was
reported. The new gRPC imports in the code and the test need that branch's
`v1p60p1` namespace, as in FLINK-40707.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with
`@Public(Evolving)`: no
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its
components), Checkpointing, Kubernetes/Yarn, ZooKeeper: yes, closing a Python
environment can wait up to 1 s longer for its logging stream
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? no
- If yes, how is the feature documented? not applicable
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (please specify the tool below)
Generated-by: Claude Code 2.1.303 (Claude Opus 5.5)
--
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]
