rob-9 opened a new pull request, #1100:
URL: https://github.com/apache/flink-agents/pull/1100
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`
- [x] `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?
- [x] Yes
- [ ] No
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
