Apache Flink master (2.4-SNAPSHOT) picked up 28 commits: 124 files, 8087 insertions, 406 deletions. The changes that affect a running job are the TLS defaults, RocksDB prefix scans, and Table API cast nullability. PyFlink DataFrame I/O and tests are a large share of that diff.
TLS 1.3 is negotiated when both peers can use it ¶
Enable TLS 1.3 in the default protocol and cipher list changes two strings in SecurityOptions. security.ssl.protocol moves from TLSv1.2 to TLSv1.2,TLSv1.3. security.ssl.algorithms keeps the two TLS 1.2 GCM suites and adds TLS_AES_128_GCM_SHA256 and TLS_AES_256_GCM_SHA384. The SSL docs say each side picks the highest protocol both support that also has a matching cipher. A peer without TLS 1.3 stays on TLS 1.2 when a TLS 1.2 suite is still listed. A value you already set for either option is left as written.
The new default depends on comma separated protocol support. Pekko ConfigSSLEngineProvider fed the raw string to SSLContext.getInstance and setEnabledProtocols, and a comma is not a protocol name. CustomSSLEngineProvider now uses SSLContext.getInstance("TLS"), splits the list, and sets engine protocols and ciphers. The generated HOCON quotes the protocol so the comma survives parsing.
That commit sets endpointIdentificationAlgorithm to null in createInternalNettySSLContext. From Netty 4.2, endpoint identification is on by default for the JDK and OpenSSL providers. Internal SSL uses one shared certificate, so the peer hostname is not on it and the check drops the connection. REST TLS stays on createRestNettySSLContext.
security.ssl.protocol: TLSv1.2,TLSv1.3
security.ssl.algorithms: TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256,TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384,TLS_AES_128_GCM_SHA256,TLS_AES_256_GCM_SHA384
RocksDB scans stop at the prefix ¶
Three commits under FLINK-40584 change RocksDBMapState and the timer queue. Bound MapState and timer prefix iterators stops a native seek at the end of the key and namespace, or at the end of the key group, so the scan stays off tombstones from other state. Timer initialization uses the same bound in RocksDBCachingPriorityQueueSet. The upper bound lives on a private ReadOptions copy in RocksDBOperationUtils. Auto prefix mode stays on, because a prefix extractor otherwise honors an upper bound only when it shares the seek key prefix.
Clear inherited seek restrictions removes prefixSameAsStart and iterate_lower_bound from that copy. prefixSameAsStart stops at the next extracted prefix, which can fall inside a map or a key group. A lower bound above the seek key drops entries with no error. Caller options remain only when the prefix has no exclusive upper bound.
Check iterator validity before advancing covers MapState.remove. A cache refill can seek past the last entry while the deletion flag stays stale, and the next advance then runs on an exhausted iterator. RocksDB rejects that call. Factories that set either knob on the shared options will not see it on these bounded iterators.
CAST nullability follows the input ¶
Derive CAST nullability from the input changes CastConverter. The planner used the target type literal, so a not null target could mark a nullable input as not null. The converter now uses the resolved call output type. The result is nullable when the input may be null. Top level nullability on the target type is ignored. Nullability of nested types, such as an array element, still comes from the target.
Preserve nullability when casting VARIANT is the strategy under that converter. SpecificTypeStrategies.CAST marks the result nullable when the source is nullable. It also forces a nullable result when the source is VARIANT and the target is another type, even if the variant column itself is not null. A not null variant cast to INT comes out nullable. Variant.toJson now prints Z for a zero offset on TIMESTAMP_LTZ and TIMESTAMP_LTZ_NS, replacing +00:00 in BinaryVariantUtil. Other offsets stay numeric. A matcher for the +00:00 suffix will miss UTC values.
Two Avro fixes are in the same batch. Nested rows now receive legacyTimestampMapping. The old createRowConverter path always assumed the legacy mapping, so a nested TIMESTAMP_LTZ threw Unsupported type: TIMESTAMP_LTZ with that mapping off. Enum symbols are checked before CHAR or VARCHAR becomes an EnumSymbol. Unknown symbols take the schema default when one exists, and otherwise fail in the converter with the value and the allowed symbols.
Savepoint bootstrap shares the task memory manager ¶
Use the task IO and memory manager deletes the private setup in SavepointEnvironment. That class built its own IOManagerAsync, a MemoryManager fixed at 64 MiB, and a stub SavepointTaskManagerRuntimeInfo. It now requires a StreamingRuntimeContext and reuses the task IOManager, MemoryManager, SharedResources, and TaskManagerRuntimeInfo.
StateBootstrapTransformation no longer calls stream.keyBy. keyBy marked BootstrapStreamTaskRunner as keyed, so the runner took managed memory the nested operator needed. Partitioning is a KeyGroupStreamPartitioner on a PartitionTransformation. The isolated 64 MiB is gone. Size the slot for the operator you bootstrap. Skipping keyBy also skipped the key check. Reject key types that cannot be hashed reliably puts KeySelectorUtil.validateKeyType back on that path. Arrays, enums, and POJOs whose hashCode is still Object.hashCode fail with InvalidProgramException while the transformation is built, before a savepoint is written. Tuple fields are included in the walk.
PyFlink DataFrames read Parquet and JSON files ¶
Parquet and JSON filesystem I/O adds read_parquet and read_json in dataframe.py. Both are PublicEvolving and call the filesystem connector, which must be on the classpath with the format. The default scan is bounded. monitor_interval makes an unbounded streaming source that records each path once and does not tail appended bytes. With partition_by, only partitions seen at planning time are visible. New partition directories stay unread, and a partitioned source that is empty at planning stays empty. Default partitions that stand for null or empty values are still unreadable. utc_timezone defaults to false, so Parquet timestamps use the JVM zone, not the session zone.
DataFrame.__getattr__ resolves an existing column without running a job. The name has to be a public identifier and not a Python keyword, and real DataFrame attributes win. flat_map is the other new method. MapWriter resets key and value writers between batches, so later Arrow batches keep aligned offsets.
What to watch ¶
Leaving both SSL options unset is what picks up TLS 1.3. A config that already sets both keeps its handshake. If you customized only the cipher list, keep a TLS 1.2 suite in it for peers that cannot do TLS 1.3.
RocksDB options factories that set prefixSameAsStart or iterate_lower_bound no longer apply those to bounded MapState and timer iterators. Retest timer recovery if those knobs were there to limit scans.
Casts to an explicit not null type, and jobs that match +00:00 in variant JSON, both change on this master. Nested Avro TIMESTAMP_LTZ should pass when the legacy mapping is off. Unknown enum strings fail in the converter, or take the schema default, before GenericDatumWriter.