[runtime] Preserve typed Flink key identity in action state

docs/content/docs/operations/configuration.md

@@ -179,7 +179,7 @@ Here are the configuration options for Kafka-based Action State Store.
 | Key                                 | Default                  | Type    | Description                                                                 |
 |-------------------------------------|--------------------------|---------|-----------------------------------------------------------------------------|
 | `kafkaBootstrapServers`             | "localhost:9092"         | String  | The config parameter specifies the Kafka bootstrap server.                  |
-| `kafkaActionStateTopic`             | (none)                   | String  | The config parameter specifies the Kafka topic for action state.            |
+| `kafkaActionStateTopic`             | (none)                   | String  | The Kafka topic for action state. Dedicate it to one logical Flink Agents operator, shared by that operator's subtasks. |
 | `kafkaActionStateTopicNumPartitions`| 64                       | Integer | The config parameter specifies the number of partitions for the Kafka action state topic. |
 | `kafkaActionStateTopicReplicationFactor` | 1                     | Integer | The config parameter specifies the replication factor for the Kafka action state topic. |
 
@@ -191,7 +191,7 @@ Here are the configuration options for Fluss-based Action State Store.
 |------------------------------|------------------|---------|------------------------------------------------------------------------------------------|
 | `flussBootstrapServers`      | "localhost:9123" | String  | The Fluss bootstrap servers address.                                                     |
 | `flussActionStateDatabase`   | "flink_agents"   | String  | The Fluss database name for storing action state.                                        |
-| `flussActionStateTable`      | (none)           | String  | The Fluss table name for storing action state.                                           |
+| `flussActionStateTable`      | (none)           | String  | The Fluss table for action state. Dedicate it to one logical Flink Agents operator, shared by that operator's subtasks. |
 | `flussActionStateTableBuckets` | 64             | Integer | The number of buckets for the Fluss action state table.                                  |
 | `flussSecurityProtocol`      | "PLAINTEXT"      | String  | The authentication protocol for Fluss client. Valid values: `PLAINTEXT` (default, no authentication), `SASL` (SASL/PLAIN authentication). |
 | `flussSaslMechanism`         | "PLAIN"          | String  | The SASL mechanism for Fluss authentication.                                             |

docs/content/docs/operations/deployment.md

@@ -102,6 +102,16 @@ The same persisted action state is also used by fine-grained durable execution.
 **Note**: Currently, Kafka and Fluss are supported as the external action state store.
 {{< /hint >}}
 
+{{< hint warning >}}
+The current versioned action-state key format is not compatible with records written in the earlier unversioned format. When upgrading a job that still has unversioned action-state records in its recovery range, use a fresh Kafka topic or Fluss table and start without an older checkpoint or savepoint. Recovery rejects the old format instead of guessing which typed Flink key it represents.
+
+Action-state keys contain a digest of the serialized key and a fingerprint of the key serializer's snapshot. Recovery requires the same key type and a byte-for-byte-equivalent serializer snapshot. Flink can accept a serializer that reads old bytes but writes different bytes for the same key; registering a previously unregistered key class with Kryo is one example. The fingerprint check rejects a changed snapshot before a lookup can miss completed action state and repeat its side effects. This conservative check also rejects snapshot changes that leave a particular key's bytes unchanged, including unrelated Kryo registrations. To recover existing state, use the original serializer configuration. To change it, use a fresh Kafka topic or Fluss table and start without an older checkpoint or savepoint. Custom key serializers must produce deterministic bytes and record encoding changes in their snapshots.
+
+Dedicate each Kafka topic or Fluss table to one logical Flink Agents operator, shared by that operator's subtasks. Action-state keys do not contain a job or operator namespace, so sharing a backend between logical operators can allow otherwise identical records to collide.
+{{< /hint >}}
+
+The business-key component stored in the backend is a SHA-256 digest rather than the serialized key itself. This keeps record keys bounded and avoids embedding raw key bytes, but it is not encryption; protect the action-state backend with appropriate access controls.
+
 See [Action State Store Configuration]({{< ref "docs/operations/configuration#action-state-store" >}}) for configuration options.
 
 {{< hint info >}}

runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateKeyEncoder.java

@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.agents.runtime.actionstate;
+
+import org.apache.flink.agents.api.Event;
+import org.apache.flink.agents.plan.actions.Action;
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.util.Preconditions;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.function.IntPredicate;
+
+/**
+ * Encodes and validates versioned action-state keys using an operator's keyed-state serializer.
+ *
+ * <p>Every key carries a fingerprint of the serializer snapshot. Recovery requires an identical
+ * snapshot even when Flink accepts the new serializer: reading old bytes successfully does not
+ * guarantee that serializing the same key produces the same digest. Custom key serializers must
+ * produce deterministic bytes and describe encoding changes in their snapshots.
+ */
+@Internal
+public final class ActionStateKeyEncoder {
+
+    private final int maxParallelism;
+    private final TypeSerializer<Object> keySerializer;
+    private final String serializerFingerprint;
+
+    public ActionStateKeyEncoder(int maxParallelism, TypeSerializer<?> keySerializer) {
+        Preconditions.checkArgument(
+                maxParallelism > 0,
+                "maxParallelism must be positive but was %s; it must match the operator's maximum parallelism.",
+                maxParallelism);
+        this.maxParallelism = maxParallelism;
+        this.keySerializer = duplicateKeySerializer(keySerializer);
+        this.serializerFingerprint =
+                ActionStateUtil.generateSerializerFingerprint(this.keySerializer);
+    }
+
+    public String generateKey(Object key, long seqNum, Action action, Event event)
+            throws IOException {
+        return ActionStateUtil.generateKey(
+                key, seqNum, action, event, maxParallelism, keySerializer, serializerFingerprint);
+    }
+
+    public String generateBusinessKeyIdentity(Object key) {
+        return ActionStateUtil.generateBusinessKeyIdentity(key, keySerializer);
+    }
+
+    public boolean isKeyRetained(@Nullable IntPredicate ownershipFilter, String stateKey) {
+        return ActionStateUtil.isKeyRetained(
+                ownershipFilter, stateKey, maxParallelism, serializerFingerprint);
+    }
+
+    @VisibleForTesting
+    String getSerializerFingerprint() {
+        return serializerFingerprint;
+    }
+
+    @SuppressWarnings("unchecked")
+    private static TypeSerializer<Object> duplicateKeySerializer(TypeSerializer<?> keySerializer) {
+        return (TypeSerializer<Object>)
+                Preconditions.checkNotNull(keySerializer, "keySerializer cannot be null")
+                        .duplicate();
+    }
+}

runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateKeyPartitioner.java

@@ -17,12 +17,15 @@
  */
 package org.apache.flink.agents.runtime.actionstate;
 
+import org.apache.flink.annotation.Internal;
 import org.apache.flink.util.MathUtils;
 import org.apache.kafka.clients.producer.Partitioner;
 import org.apache.kafka.common.Cluster;
 
 import java.util.Map;
 
+/** Partitions versioned action-state records by their encoded business-key identity. */
+@Internal
 public class ActionStateKeyPartitioner implements Partitioner {
 
     @Override
@@ -41,15 +44,15 @@ public int partition(
             throw new IllegalArgumentException("Key must be a String");
         }
 
-        String businessKey = ActionStateUtil.businessKeyOf((String) key);
-        if (businessKey == null) {
+        String businessKeyIdentity = ActionStateUtil.businessKeyIdentityOf((String) key);
+        if (businessKeyIdentity == null) {
             throw new IllegalArgumentException("Key format is invalid");
         }
-        if (businessKey.isEmpty()) {
-            throw new IllegalArgumentException("Business key part of the key cannot be empty");
+        if (businessKeyIdentity.isEmpty()) {
+            throw new IllegalArgumentException("Business key identity cannot be empty");
         }
 
-        return MathUtils.murmurHash(businessKey.hashCode()) % numPartitions;
+        return MathUtils.murmurHash(businessKeyIdentity.hashCode()) % numPartitions;
     }
 
     @Override