[FLINK-40628][python] Avoid gateway launch during embedded TTL initialization

[bowenli86]

What is the purpose of the change

Fix FLINK-40628. Initializing TTL-enabled state in a Python thread-mode operator can request a Py4J gateway while constructing the default RocksDB cleanup strategy. Starting that subprocess from a multithreaded TaskManager can leave the task stuck in INITIALIZING.

Brief change log

• Defer the default 30-day Duration until the public accessor is called. The embedded converter continues to use Java's existing default; explicit durations and existing falsey-argument behavior are preserved.
• Reject get_gateway() and launch_gateway() in embedded workers before locking, connecting, or launching a subprocess. Detect the native Pemja interpreter and direct callers to pemja.findClass in the error message.
• Add focused gateway and TTL regressions, plus embedded-worker tests for both gateway entry points, Java access through Pemja, and RocksDB state updates across four TTL cleanup policies.

Verifying this change

• Focused Python suite: 9 tests and 15 subtests passed. The new guard tests fail against the implementation without the guard. Flake8, Python syntax, and diff checks passed.
• Standalone Linux validation used the exact checked-in worker functions and production Python files with public Flink 2.3.0 / Pemja 0.5.7 / Python 3.11 / Java 17. Both embedded regressions failed safely before their respective fixes and passed afterward; the TTL case produced all eight expected outputs. Ordinary client gateway creation, reuse, and direct launch also passed. This supplements the repository CI run below.
• Azure build 79012 passed all 13 jobs on rebased head 7db42d567feb3699f1b36201aa51cef7e926146f: compilation, Python, table, core, connectors, misc, tests, licensing, documentation, and all four E2E groups. The gateway and TTL Python suites passed; the previously failing CalcRestoreTest reported zero failures and errors (one pre-existing skip). No tests are disabled or skipped by this change.

Does this pull request potentially affect one of the following parts:

• Dependencies: no
• Public API: no signature or annotation changes
• Serializers: no format changes
• Runtime per-record code paths: no change to normal record processing; unsupported Py4J calls from embedded workers now fail immediately
• Deployment or recovery: Python operator state initialization; no checkpoint/savepoint format or behavior changes
• 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?

• Yes

Generated-by: Codex (GPT-6)

[flinkbot]

CI report:

• 7db42d5 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[bowenli86]

@flinkbot run azure

[bowenli86]

@flinkbot run azure

[dianfu]

@bowenli86 Good catch! Thanks for this fix.

Do you think it makes sense to add some centralized guard (raise some meaningful error messages to tell users to use pemja.findClass if they want to call Java from Python in thread mode) for embedded workers. Users can accidentally‌ call Py4J and potentially fall through to launch_gateway().

Besides, I saw that we have included changes of FLINK-40631 in this PR. Are there any special reasons to do so?

[bowenli86]

Thanks @dianfu, agreed. I’ve added a centralized guard to both get_gateway() and launch_gateway(). In embedded workers, these now fail immediately with an error directing users to pemja.findClass. The checks run before gateway locking, connection, or launch.

FLINK-40631 was included temporarily to unblock the unrelated CalcRestoreTest failure. Latest master now contains the proper fix, so I’ve rebased onto it and removed the fixture and revert commits. The PR now contains two focused FLINK-40628 commits.

I also added regression coverage for the guard and verified Java access through Pemja. The focused tests and standalone embedded runtime checks pass; fresh Azure CI is pending.

[bowenli86]

@flinkbot run azure

[dianfu]

@bowenli86 Thanks for the update. LGTM.