wenjin272 commented on code in PR #874:
URL: https://github.com/apache/flink-agents/pull/874#discussion_r3541046130
##########
runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateSerde.java:
##########
@@ -88,4 +150,62 @@ public void serialize(ActionTask value, JsonGenerator gen,
SerializerProvider se
gen.writeNull();
}
}
+
+ /**
+ * Binds the Kryo envelope (de)serializer to the {@code value} property of
{@link MemoryUpdate}.
+ */
+ abstract static class MemoryUpdateValueMixin {
+ @JsonSerialize(using = MemoryUpdateValueSerializer.class)
+ @JsonDeserialize(using = MemoryUpdateValueDeserializer.class)
+ Object value;
+ }
+
+ /**
+ * Writes a non-null {@link MemoryUpdate} value as the {@code
{serde,version,payload}} envelope.
+ */
+ static class MemoryUpdateValueSerializer extends JsonSerializer<Object> {
+ @Override
+ public void serialize(Object value, JsonGenerator gen,
SerializerProvider provider)
+ throws IOException {
+ gen.writeStartObject();
+ gen.writeStringField(ENVELOPE_SERDE_FIELD, MEMORY_UPDATE_SERDE);
+ gen.writeNumberField(ENVELOPE_VERSION_FIELD,
MEMORY_UPDATE_VERSION);
+ gen.writeStringField(
+ ENVELOPE_PAYLOAD_FIELD,
+ Base64.getEncoder().encodeToString(kryoSerialize(value)));
+ gen.writeEndObject();
+ }
+ }
+
+ /**
+ * Reads the Kryo envelope back to the concrete value; version-checks;
falls back for a
+ * non-envelope node.
+ */
+ static class MemoryUpdateValueDeserializer extends
JsonDeserializer<Object> {
+ @Override
+ public Object deserialize(JsonParser p, DeserializationContext ctxt)
throws IOException {
+ JsonNode node = ctxt.readTree(p);
+ if (isEnvelope(node)) {
Review Comment:
IIUC, the isEnvelope flag is used to maintain backward compatibility with
legacy MemoryUpdate instances that were serialized using JSON.
Is it essential to maintain backward compatibility in this case?
* For new jobs, both serialization and deserialization should be performed
using Kryo, with `isEnvelope` set to true.
* For old jobs, both serialization and deserialization should be performed
using Jackson.
* The `isEnvelope` can only become false after a job upgrade and restart.
Since flink agents are still in beta, I consider this incompatibility
acceptable.
--
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]
