[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.
• Preserve explicit durations and existing falsey-argument behavior.
• Include the test-only calc-variant savepoint refresh from FLINK-40631 / PR #29156, with its upstream attribution, so the unrelated master table test can deserialize the current Variant type.
• Add Python regression tests for gateway-free construction, pickle/protobuf round trips, and accessor compatibility, plus a RocksDB thread-mode regression that guards gateway access inside the worker and checks state updates across four cleanup policies.

Verifying this change

• Azure build 78968 at current head 2932b88ed1ababb9abb6dd9bfd37a54ae00b90c1 passed all 13 jobs: compilation, Python, table, core, connectors, misc, tests, licensing, documentation, and all four E2E groups.
• Python CI exercised both test_state_ttl_config.py and the checked-in EmbeddedDataStreamStreamTests.test_state_ttl_without_gateway. The previously failing CalcRestoreTest also passed with zero failures or errors after the fixture refresh. No tests are disabled or skipped by this change.
• The local regression suite passed 5 tests and 11 subtests; Flake8, syntax, and diff checks passed.
• Standalone execution of the exact TTL worker UDF in public Flink 2.3.0 / Pemja 0.5.7 / Python 3.11 / Java 17 with RocksDB failed before the TTL fix at the gateway guard and passed afterward with all four cases and eight expected outputs.

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
• 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:

• 2932b88 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?