[runtime] Preserve typed Flink key identity in action state

[rob-9]

Closes #1099.

This branch is stacked on PR #1094, which makes the action identifier stable across JVM restarts. The two changes fix separate parts of the action-state key.

Purpose of change

We currently store action progress under the current key so recovery can find completed work and avoid repeating its side effects. The existing format turns that key into text. Different typed keys with the same text, such as Long(1) and String("1"), can therefore point to the same action state. A lookup or prune for one key can remove the other key's completed state, allowing recovery to run that action again.

This change gives each typed key a durable identity derived from its serialized bytes. Kafka and Fluss use that identity consistently for writes, lookups, cleanup, partitioning, and recovery.

Runtime flow

1. During operator initialization, the store receives the keyed backend's serializer and maximum parallelism.
2. ActionStateKeyEncoder serializes the typed key and hashes the bytes with SHA-256. The versioned state key contains that digest, the key-group, sequence number, event ID, action ID, and serializer fingerprint.
3. Kafka partitions by the business-key digest. Fluss uses it as the table's distribution identity. Both stores use it for lookup, divergence cleanup, and pruning.
4. Recovery validates the version, fields, key-group range, and serializer fingerprint before applying the subtask ownership filter and caching each owned record.

Key decisions

• Serialized bytes preserve type information and align the identity with keyed state. A fixed-length digest keeps raw key data out of backend keys and gives every record a bounded key size.
• The serializer-snapshot fingerprint turns a potentially silent lookup miss into a clear recovery error. This rule also covers serializer changes that Flink classifies as compatible after migration when their bytes differ.
• Earlier unversioned records contain an ambiguous textual key, so recovery rejects them with upgrade guidance.

Related work

Issue #1034 manages the Kafka offset boundary for deleting an older log prefix, while this PR defines the identity of each record. PR #885 manages Kafka tombstones. When those changes are combined, replay must validate the record key before applying a value or tombstone, and pruning must derive the same typed identity.

Behavioral Semantics

Interaction decisions

Record during recovery	Ownership	Result
Current format and expected serializer	Owned	Cache the record
Current format and expected serializer	Foreign	Skip the record
Different serializer fingerprint	Any	Stop with a compatibility error
Malformed field or out-of-range key-group	Any	Stop and identify the invalid field and key
Earlier unversioned format	Any	Stop with upgrade guidance

Behavioral contracts

• Long(1), String("1"), and custom keys sharing a string representation receive separate identities, including at maximum parallelism 1.
• Equivalent instances of the same Flink serializer produce the same identity and fingerprint.
• Kafka and Fluss share one typed identity across storage, lookup, cleanup, distribution, and recovery.
• Recovery validates each record first, then caches records owned by the current subtask.
• Generated keys use canonical fields and a nonnegative sequence number.

Failure behavior

• Key serialization and serializer-snapshot fingerprinting failures raise immediately and retain their original causes.
• Legacy records, malformed fields, noncanonical values, out-of-range key-groups, and serializer mismatches stop recovery with a specific error.
• Pruning preserves records whose format or sequence number prevents safe attribution.

API

The supported user-facing API and configuration remain unchanged. The stores, key utilities, and Kafka partitioner now carry explicit @Internal annotations. Their construction path receives an ActionStateKeyEncoder, so direct callers of these implementation classes must update their constructor calls.

The durable record format changes incompatibly. A recovery range containing unversioned records requires a fresh Kafka topic or Fluss table and a fresh job state. Versioned recovery also requires the same key type and a byte-for-byte-equivalent serializer snapshot configuration.

Documentation

• doc-needed
• doc-not-needed
• doc-included

The deployment and configuration documentation explains the format transition, serializer compatibility, backend isolation, and digest security boundary.

Was this patch authored or co-authored using generative AI tooling?

• Yes
• No