wenjin272 commented on code in PR #1100:
URL: https://github.com/apache/flink-agents/pull/1100#discussion_r3986109690
##########
runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateUtil.java:
##########
@@ -22,57 +22,75 @@
import com.fasterxml.jackson.databind.json.JsonMapper;
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.api.common.typeutils.TypeSerializer;
+import
org.apache.flink.api.common.typeutils.TypeSerializerSnapshotSerializationUtil;
+import org.apache.flink.core.memory.DataOutputSerializer;
import org.apache.flink.runtime.state.KeyGroupRangeAssignment;
import org.apache.flink.util.Preconditions;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.Base64;
import java.util.List;
import java.util.UUID;
import java.util.function.IntPredicate;
import java.util.function.LongPredicate;
/** Utility class for action state related operations. */
-public class ActionStateUtil {
-
- private static final Logger LOG =
LoggerFactory.getLogger(ActionStateUtil.class);
+@Internal
+public final class ActionStateUtil {
private static final JsonMapper MAPPER =
JsonMapper.builder()
.configure(SerializationFeature.ORDER_MAP_ENTRIES_BY_KEYS,
true)
.configure(MapperFeature.SORT_PROPERTIES_ALPHABETICALLY,
true)
.build();
private static final String KEY_SEPARATOR = "_";
Review Comment:
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.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateUtil.java:
##########
@@ -81,16 +99,49 @@ public static String generateKey(
int keyGroup = KeyGroupRangeAssignment.assignToKeyGroup(key,
maxParallelism);
return String.join(
KEY_SEPARATOR,
- String.valueOf(keyGroup),
+ KEY_GROUP_PREFIX + keyGroup,
String.valueOf(seqNum),
generateUUIDForEvent(event),
generateUUIDForAction(action),
- key.toString());
+ serializerFingerprint,
+ generateBusinessKeyIdentity(key, keySerializer));
+ }
+
+ /** Returns a stable digest of a Flink key's serialized, type-preserving
representation. */
+ public static <K> String generateBusinessKeyIdentity(
+ @Nonnull K key, @Nonnull TypeSerializer<K> keySerializer) {
+ Preconditions.checkNotNull(key, "key cannot be null.");
+ Preconditions.checkNotNull(keySerializer, "keySerializer cannot be
null.");
+ DataOutputSerializer output = new DataOutputSerializer(64);
+ try {
+ keySerializer.serialize(key, output);
+ } catch (IOException e) {
+ throw new IllegalStateException(
+ "Failed to serialize the Flink key for durable action
state", e);
+ }
+ return sha256Base64(output.getCopyOfBuffer());
+ }
+
+ /**
+ * Fingerprints the serializer's snapshot, including its version and
configuration, once per
+ * store. Custom serializers must describe all encoding changes in their
snapshots.
+ */
+ static String generateSerializerFingerprint(TypeSerializer<?>
keySerializer) {
+ DataOutputSerializer output = new DataOutputSerializer(128);
+ try {
+ TypeSerializerSnapshotSerializationUtil.writeSerializerSnapshot(
+ output, keySerializer.snapshotConfiguration());
Review Comment:
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.
--
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]
