[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

Flink Agents stores 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 state key contains that digest, the key-group, sequence number, event ID, and action ID.
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 fields and key-group range 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.
• Serializer snapshots can change during normal operation as POJO subclass caches grow. The key therefore uses serialized key bytes without a snapshot fingerprint.
• The store supports one record format. Existing records and changes to key types or serializer configuration during recovery are unsupported; field validation cannot reliably detect every incompatible record.

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
Valid current-format fields	Owned	Cache the record
Valid current-format fields	Foreign	Skip the record
Malformed field or out-of-range key-group	Any	Stop and identify the invalid field and key

Behavioral contracts

• Long(1), String("1"), and custom keys sharing a string representation receive separate identities, including at maximum parallelism 1.
• Equivalent serializers produce the same key identity.
• Unchanged POJO jobs recover completed action state even when runtime subclass caching changes their serializer snapshots.
• 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 failures raise immediately and retain their original causes.
• Malformed fields, noncanonical values, and out-of-range key-groups stop recovery with a specific error. Recovery errors bound every echoed key and field.
• 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. Existing action-state records are unsupported; use a fresh Kafka topic or Fluss table and start without an older checkpoint or savepoint when upgrading. For subsequent recovery, preserve the key type and serializer configuration. Flink may accept a serializer change that produces different key bytes; the action-state store does not detect that change and may miss completed work.

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

[wenjin272]

Thanks for working on this and for adding the key-isolation coverage! I left two suggestions to simplify the state-key format and avoid rejecting valid recovery.

[wenjin272]

Could we remove the serializer snapshot fingerprint from both the state key and recovery validation? Flink’s PojoSerializer snapshots include a runtime subclass cache, so an unchanged job can produce a different fingerprint after checkpoint recovery even though the key bytes remain identical. I reproduced this with a POJO subclass: Flink keyed state restored successfully, but Kafka action-state recovery failed. For this PR, documenting that changes to key types or serializer configurations are unsupported during recovery seems sufficient.

[rob-9]

removed. I added a Pojo recovery test that fails before this fix, and documented that key type and serializer config changes are unsupported during recovery.

[wenjin272]

Could we also remove the v2 prefix and keep a single state-key format? Since we’re still in 0.x and don’t need backward compatibility for existing action-state records, the version marker adds unnecessary complexity and may suggest a compatibility mechanism we don’t intend to provide.

[rob-9]

yep, makes sense!