rob-9 commented on code in PR #1100:
URL: https://github.com/apache/flink-agents/pull/1100#discussion_r3988534403
##########
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:
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.
##########
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:
yep, makes sense!
--
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]
