[issue-180] Provide method for deserialization with metadata

[crazyzhou]

Signed-off-by: Brian Zhou [email protected]

Change log description

• [issue-180] Provide method for deserialization with metadata

Purpose of the change
Fixes #180

What the code does
Provide a method deserializeWithMetadata for user to override that can leverage EventRead in the deserialization.
Added unit test for the case.

How to verify it
./gradlew clean build passes
Target to master, should cherry-pick to all 0.7 branch

[codecov]

Codecov Report

❗ No coverage uploaded for pull request base (master@cce935d). Click here to learn what that means.
The diff coverage is n/a.

@@            Coverage Diff            @@
##             master     #305   +/-   ##
=========================================
  Coverage          ?   84.53%           
  Complexity        ?      345           
=========================================
  Files             ?       36           
  Lines             ?     1623           
  Branches          ?      177           
=========================================
  Hits              ?     1372           
  Misses            ?      139           
  Partials          ?      112

Impacted Files	Coverage Δ	Complexity Δ
...tors/flink/serialization/PravegaSerialization.java	0% <ø> (ø)	0 <0> (?)

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update cce935d...0038211. Read the comment docs.

[EronWright]

Good job!

[claudiofahey]

This name suggests that it replaces deserialize which is not the case. This method does not actually deserialize. I recommending renaming it to insertMetadata or something similar.

[crazyzhou]

It's true that this one does not do deserialization, but it really "get" the data. It has a default implementation eventRead.getEvent() and gets called in the FlinkPravegaReader here: https://github.com/pravega/flink-connectors/pull/305/files#diff-005207bb16199a05351084fea79fd9acR271

[crazyzhou]

I'm also not a fan of the current name, but I don't think insertMetadata is correct because if it's not overriden which is most cases, it won't insert metadata. Maybe getData , getEvent is better? Any suggestions of a better name?

[maddisondavid]

When reviewing this code I also got confused by the name of this method because it's not actually obvious that both deserialize methods will get called, one at the Pravega level and one at the Flink Connector level.

It's almost like the method should be called extractEvent since it's extracting it from the Pravega level API (and providing the chance to inject Metadata into the event). I considered perhaps it should be called getEvent or retrieveEvent but this would get confusing with the Pravega semantics.

[crazyzhou]

extractEvent is a good one, I can buy in this name

[maddisondavid]

Type T here is overloaded in that it expresses both the type of object expected from the Pravega stream (used by the Pravega API) and the type of object expected from the connector (with possible Metadata injected). This means that the object being stored in the stream MUST serialize to a Java class that expects that Metadata. For some formats (such as JSON) this may be easy but for other formats such as Java it's difficult and messy. In fact this can be seen in the Unit tests where IntegerWithEventPointer is the type being "stored" in the stream.

I'm not sure if it's possible, but it feels like there should be two types, a type T to represent the type from the stream and a type C to represent the type from the connector. In most cases these could be the same but in other cases where MetaData needs to be injected these could be different.

Or maybe (just thinking allowed) we're talking about two completely different interfaces, i.e. a PravegaMetaDataSchema<T, C> that extends PravegaDeserializationSchema and just has the method

public C deserializeWithMetadata(EventRead<T> eventRead) {

[crazyzhou]

Yes, this was my initial thought, and I also tried with it, but finally chose this. This is also what Kafka's approach to keep the same format of the event: https://github.com/apache/flink/blob/c6d9b37a8362317cc364c132ced5ac10dd3e10f2/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/util/serialization/JSONKeyValueDeserializationSchema.java.

I wrote some investigation and the reason here: https://docs.google.com/document/d/1Dp8KYA6N81869dFjhD7w3z2hXd2QkU-vS1DNkz5aRC4/edit#

[maddisondavid]

LGTM