weiqingy commented on code in PR #874:
URL: https://github.com/apache/flink-agents/pull/874#discussion_r3541600905
##########
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:
Thanks for the review, @wenjin272. I've updated the PR accordingly.
--
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]
