You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2020/09/15 16:08:36 UTC
[GitHub] [flink-statefun] galenwarren opened a new pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
galenwarren opened a new pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152
This PR implements support for pluggable message payload serializers. It adds a new MessageFactoryType -- WITH_CUSTOM_PAYLOADS -- and a new class (MessageFactoryKey) that combines together MessageFactoryType and the name of a class to be used for payload serialization. This class must implement MessagePayloadSerializer.
This is pretty straightforward change, mostly replacing uses of MessageFactoryType with MessageFactoryKey. The changes to the MessageTypeSerializer.Snapshot class are a bit interesting, in that the serialized format for the snapshot had to change, to include the name of the custom serializer class. The snapshot version number was bumped from 1 to 2, and the new version of the class can read snapshots of both version 1 and 2, which is confirmed via new unit tests in MessageTypeSerializerSnapshotTest.
Any feedback or suggestions are welcome, and thanks to @igalshilman for pointing me in the right direction here.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] tzulitai edited a comment on pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
tzulitai edited a comment on pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152#issuecomment-698127300
LGTM, merging this now to `master`!
> Also, I was wondering if you know when this might make it into a release. No hurry, I can use my fork until then, just wondering.
@galenwarren
Since the feature freeze for 2.2 already happened, and this counts as a new feature, this will have to be included only in the next release (2.3).
The release cycles for StateFun are relatively shorter than Flink, so I guess you can expect this to be officially available within ~6 weeks when the next release is ready.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] galenwarren commented on a change in pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
galenwarren commented on a change in pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152#discussion_r491706635
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/message/MessageTypeSerializer.java
##########
@@ -101,45 +102,67 @@ public int hashCode() {
@Override
public TypeSerializerSnapshot<Message> snapshotConfiguration() {
- return new Snapshot(messageFactoryType);
+ return new Snapshot(messageFactoryKey);
}
private MessageFactory factory() {
if (factory == null) {
- factory = MessageFactory.forType(messageFactoryType);
+ factory = MessageFactory.forKey(messageFactoryKey);
}
return factory;
}
public static final class Snapshot implements TypeSerializerSnapshot<Message> {
- private MessageFactoryType messageFactoryType;
+ private MessageFactoryKey messageFactoryKey;
@SuppressWarnings("unused")
public Snapshot() {}
- Snapshot(MessageFactoryType messageFactoryType) {
- this.messageFactoryType = messageFactoryType;
+ Snapshot(MessageFactoryKey messageFactoryKey) {
+ this.messageFactoryKey = messageFactoryKey;
+ }
+
+ // used in unit tests
+ MessageFactoryKey getMessageFactoryKey() {
Review comment:
Done
##########
File path: statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/message/JavaPayloadSerializer.java
##########
@@ -0,0 +1,50 @@
+package org.apache.flink.statefun.flink.core.message;
Review comment:
Fixed
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/StatefulFunctionsConfigValidator.java
##########
@@ -61,4 +64,26 @@ private static void validateParentFirstClassloaderPatterns(Configuration configu
}
return parentFirstClassloaderPatterns;
}
+
+ private static void validateCustomPayloadSerializerClassName(Configuration configuration) {
+
+ MessageFactoryType factoryType =
+ configuration.get(StatefulFunctionsConfig.USER_MESSAGE_SERIALIZER);
+ String customPayloadSerializerClassName =
+ configuration.get(StatefulFunctionsConfig.USER_MESSAGE_CUSTOM_PAYLOAD_SERIALIZER_CLASS);
+
+ if (factoryType == MessageFactoryType.WITH_CUSTOM_PAYLOADS) {
+ if (StringUtils.isNullOrWhitespaceOnly(customPayloadSerializerClassName)) {
Review comment:
A question here -- as it stands now, the USER_MESSAGE_CUSTOM_PAYLOAD_SERIALIZER_CLASS config option has no default value, so if not set, its value will be null. So I believe I'd have to first check for null and then check length on the trimmed string. I'm happy to do that, but I thought I'd check first ...
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/message/MessageFactoryKey.java
##########
@@ -0,0 +1,43 @@
+package org.apache.flink.statefun.flink.core.message;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+public class MessageFactoryKey implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ private final MessageFactoryType type;
+ private final String customPayloadSerializerClassName;
+
+ private MessageFactoryKey(MessageFactoryType type, String customPayloadSerializerClassName) {
+ this.type = type;
+ this.customPayloadSerializerClassName = customPayloadSerializerClassName;
+ }
+
+ public static MessageFactoryKey forType(
+ MessageFactoryType type, String customPayloadSerializerClassName) {
+ return new MessageFactoryKey(type, customPayloadSerializerClassName);
+ }
+
+ public MessageFactoryType getType() {
+ return this.type;
+ }
+
+ public String getCustomPayloadSerializerClassName() {
Review comment:
As it stands now, this method gets called from MessageTypeSerializer.Snapshot.writeSnapshot, for all types of message serializers, not just custom ones. For non-custom ones, a null is returned and this is the value that is written into the snapshot, after the MesssageFactoryType. So, the version 2 serialized format is always a MessageFactoryType value and then the custom payload serializer class name (string, which can be null), for all types of serializers.
The writeSnapshot method could be changed to only call getCustomPayloadSerializerClassName in the event that the serializer is custom, and then to either 1) force a null serializer class name to be written to the snapshot or 2) skip the write of the serializer class name altogether in the non-custom case. This would make the read/writeSnapshot code and testing a bit more complicated but would definitely be doable, would one of those options be better?
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/StatefulFunctionsConfigValidator.java
##########
@@ -61,4 +64,26 @@ private static void validateParentFirstClassloaderPatterns(Configuration configu
}
return parentFirstClassloaderPatterns;
}
+
+ private static void validateCustomPayloadSerializerClassName(Configuration configuration) {
+
+ MessageFactoryType factoryType =
+ configuration.get(StatefulFunctionsConfig.USER_MESSAGE_SERIALIZER);
+ String customPayloadSerializerClassName =
+ configuration.get(StatefulFunctionsConfig.USER_MESSAGE_CUSTOM_PAYLOAD_SERIALIZER_CLASS);
+
+ if (factoryType == MessageFactoryType.WITH_CUSTOM_PAYLOADS) {
+ if (StringUtils.isNullOrWhitespaceOnly(customPayloadSerializerClassName)) {
Review comment:
A question here -- as it stands now, the USER_MESSAGE_CUSTOM_PAYLOAD_SERIALIZER_CLASS config option has no default value, so if not set, its value will be null. So I believe I'd have to first check for null and then check length on the trimmed string. I'm happy to do that, but I thought I'd check first in case that affects things ...
Or should USER_MESSAGE_CUSTOM_PAYLOAD_SERIALIZER_CLASS have an empty string as the default value?
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/StatefulFunctionsConfigValidator.java
##########
@@ -61,4 +64,26 @@ private static void validateParentFirstClassloaderPatterns(Configuration configu
}
return parentFirstClassloaderPatterns;
}
+
+ private static void validateCustomPayloadSerializerClassName(Configuration configuration) {
+
+ MessageFactoryType factoryType =
+ configuration.get(StatefulFunctionsConfig.USER_MESSAGE_SERIALIZER);
+ String customPayloadSerializerClassName =
+ configuration.get(StatefulFunctionsConfig.USER_MESSAGE_CUSTOM_PAYLOAD_SERIALIZER_CLASS);
+
+ if (factoryType == MessageFactoryType.WITH_CUSTOM_PAYLOADS) {
+ if (StringUtils.isNullOrWhitespaceOnly(customPayloadSerializerClassName)) {
Review comment:
OK will do
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/message/MessageFactoryKey.java
##########
@@ -0,0 +1,43 @@
+package org.apache.flink.statefun.flink.core.message;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+public class MessageFactoryKey implements Serializable {
Review comment:
Yes, and it will be included in the next push
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/message/MessageFactoryKey.java
##########
@@ -0,0 +1,43 @@
+package org.apache.flink.statefun.flink.core.message;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+public class MessageFactoryKey implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ private final MessageFactoryType type;
+ private final String customPayloadSerializerClassName;
+
+ private MessageFactoryKey(MessageFactoryType type, String customPayloadSerializerClassName) {
+ this.type = type;
+ this.customPayloadSerializerClassName = customPayloadSerializerClassName;
+ }
+
+ public static MessageFactoryKey forType(
+ MessageFactoryType type, String customPayloadSerializerClassName) {
+ return new MessageFactoryKey(type, customPayloadSerializerClassName);
+ }
+
+ public MessageFactoryType getType() {
+ return this.type;
+ }
+
+ public String getCustomPayloadSerializerClassName() {
Review comment:
For some reason, before I couldn't comment here, but now I can. Weird glitch. Anyway, I left a top-level comment/question for you on this one, just FYI.
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/message/MessageFactoryKey.java
##########
@@ -0,0 +1,43 @@
+package org.apache.flink.statefun.flink.core.message;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+public class MessageFactoryKey implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ private final MessageFactoryType type;
+ private final String customPayloadSerializerClassName;
+
+ private MessageFactoryKey(MessageFactoryType type, String customPayloadSerializerClassName) {
+ this.type = type;
Review comment:
Yes, will be fixed in next push
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] tzulitai commented on pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
tzulitai commented on pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152#issuecomment-698127300
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] igalshilman commented on a change in pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
igalshilman commented on a change in pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152#discussion_r490279224
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/StatefulFunctionsConfigValidator.java
##########
@@ -61,4 +64,26 @@ private static void validateParentFirstClassloaderPatterns(Configuration configu
}
return parentFirstClassloaderPatterns;
}
+
+ private static void validateCustomPayloadSerializerClassName(Configuration configuration) {
+
+ MessageFactoryType factoryType =
+ configuration.get(StatefulFunctionsConfig.USER_MESSAGE_SERIALIZER);
+ String customPayloadSerializerClassName =
+ configuration.get(StatefulFunctionsConfig.USER_MESSAGE_CUSTOM_PAYLOAD_SERIALIZER_CLASS);
+
+ if (factoryType == MessageFactoryType.WITH_CUSTOM_PAYLOADS) {
+ if (StringUtils.isNullOrWhitespaceOnly(customPayloadSerializerClassName)) {
Review comment:
I think that it could be just replaced with a call to ‘.trim()’ and then checking if the string is empty.
To avoid this dependency (I know it is currently there)
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/message/MessageFactoryKey.java
##########
@@ -0,0 +1,43 @@
+package org.apache.flink.statefun.flink.core.message;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+public class MessageFactoryKey implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ private final MessageFactoryType type;
+ private final String customPayloadSerializerClassName;
+
+ private MessageFactoryKey(MessageFactoryType type, String customPayloadSerializerClassName) {
+ this.type = type;
+ this.customPayloadSerializerClassName = customPayloadSerializerClassName;
+ }
+
+ public static MessageFactoryKey forType(
+ MessageFactoryType type, String customPayloadSerializerClassName) {
+ return new MessageFactoryKey(type, customPayloadSerializerClassName);
+ }
+
+ public MessageFactoryType getType() {
+ return this.type;
+ }
+
+ public String getCustomPayloadSerializerClassName() {
Review comment:
Can you add here a Precondtions.checkState that indeed the serializer is custom?
##########
File path: statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/message/JavaPayloadSerializer.java
##########
@@ -0,0 +1,50 @@
+package org.apache.flink.statefun.flink.core.message;
Review comment:
Missing license header.
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/message/MessageTypeSerializer.java
##########
@@ -101,45 +102,67 @@ public int hashCode() {
@Override
public TypeSerializerSnapshot<Message> snapshotConfiguration() {
- return new Snapshot(messageFactoryType);
+ return new Snapshot(messageFactoryKey);
}
private MessageFactory factory() {
if (factory == null) {
- factory = MessageFactory.forType(messageFactoryType);
+ factory = MessageFactory.forKey(messageFactoryKey);
}
return factory;
}
public static final class Snapshot implements TypeSerializerSnapshot<Message> {
- private MessageFactoryType messageFactoryType;
+ private MessageFactoryKey messageFactoryKey;
@SuppressWarnings("unused")
public Snapshot() {}
- Snapshot(MessageFactoryType messageFactoryType) {
- this.messageFactoryType = messageFactoryType;
+ Snapshot(MessageFactoryKey messageFactoryKey) {
+ this.messageFactoryKey = messageFactoryKey;
+ }
+
+ // used in unit tests
+ MessageFactoryKey getMessageFactoryKey() {
Review comment:
Can you annotate this method with @VisibleForTesting annotation?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] tzulitai commented on pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
tzulitai commented on pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152#issuecomment-693213790
@galenwarren before we start reviewing this, @galenwarren can you squash your commits and rebase on latest master?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] galenwarren commented on pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
galenwarren commented on pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152#issuecomment-695807576
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] galenwarren commented on pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
galenwarren commented on pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152#issuecomment-697390016
Great -- I've resolved the ones that are done and will work on the remaining item today. Thanks.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] galenwarren commented on a change in pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
galenwarren commented on a change in pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152#discussion_r491706635
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/message/MessageTypeSerializer.java
##########
@@ -101,45 +102,67 @@ public int hashCode() {
@Override
public TypeSerializerSnapshot<Message> snapshotConfiguration() {
- return new Snapshot(messageFactoryType);
+ return new Snapshot(messageFactoryKey);
}
private MessageFactory factory() {
if (factory == null) {
- factory = MessageFactory.forType(messageFactoryType);
+ factory = MessageFactory.forKey(messageFactoryKey);
}
return factory;
}
public static final class Snapshot implements TypeSerializerSnapshot<Message> {
- private MessageFactoryType messageFactoryType;
+ private MessageFactoryKey messageFactoryKey;
@SuppressWarnings("unused")
public Snapshot() {}
- Snapshot(MessageFactoryType messageFactoryType) {
- this.messageFactoryType = messageFactoryType;
+ Snapshot(MessageFactoryKey messageFactoryKey) {
+ this.messageFactoryKey = messageFactoryKey;
+ }
+
+ // used in unit tests
+ MessageFactoryKey getMessageFactoryKey() {
Review comment:
Done
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] galenwarren commented on a change in pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
galenwarren commented on a change in pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152#discussion_r492357907
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/message/MessageFactoryKey.java
##########
@@ -0,0 +1,43 @@
+package org.apache.flink.statefun.flink.core.message;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+public class MessageFactoryKey implements Serializable {
Review comment:
Yes, and it will be included in the next push
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] galenwarren commented on a change in pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
galenwarren commented on a change in pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152#discussion_r491706662
##########
File path: statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/message/JavaPayloadSerializer.java
##########
@@ -0,0 +1,50 @@
+package org.apache.flink.statefun.flink.core.message;
Review comment:
Fixed
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] galenwarren commented on a change in pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
galenwarren commented on a change in pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152#discussion_r492357429
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/StatefulFunctionsConfigValidator.java
##########
@@ -61,4 +64,26 @@ private static void validateParentFirstClassloaderPatterns(Configuration configu
}
return parentFirstClassloaderPatterns;
}
+
+ private static void validateCustomPayloadSerializerClassName(Configuration configuration) {
+
+ MessageFactoryType factoryType =
+ configuration.get(StatefulFunctionsConfig.USER_MESSAGE_SERIALIZER);
+ String customPayloadSerializerClassName =
+ configuration.get(StatefulFunctionsConfig.USER_MESSAGE_CUSTOM_PAYLOAD_SERIALIZER_CLASS);
+
+ if (factoryType == MessageFactoryType.WITH_CUSTOM_PAYLOADS) {
+ if (StringUtils.isNullOrWhitespaceOnly(customPayloadSerializerClassName)) {
Review comment:
OK will do
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/message/MessageFactoryKey.java
##########
@@ -0,0 +1,43 @@
+package org.apache.flink.statefun.flink.core.message;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+public class MessageFactoryKey implements Serializable {
Review comment:
Yes, and it will be included in the next push
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/message/MessageFactoryKey.java
##########
@@ -0,0 +1,43 @@
+package org.apache.flink.statefun.flink.core.message;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+public class MessageFactoryKey implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ private final MessageFactoryType type;
+ private final String customPayloadSerializerClassName;
+
+ private MessageFactoryKey(MessageFactoryType type, String customPayloadSerializerClassName) {
+ this.type = type;
+ this.customPayloadSerializerClassName = customPayloadSerializerClassName;
+ }
+
+ public static MessageFactoryKey forType(
+ MessageFactoryType type, String customPayloadSerializerClassName) {
+ return new MessageFactoryKey(type, customPayloadSerializerClassName);
+ }
+
+ public MessageFactoryType getType() {
+ return this.type;
+ }
+
+ public String getCustomPayloadSerializerClassName() {
Review comment:
For some reason, before I couldn't comment here, but now I can. Weird glitch. Anyway, I left a top-level comment/question for you on this one, just FYI.
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/message/MessageFactoryKey.java
##########
@@ -0,0 +1,43 @@
+package org.apache.flink.statefun.flink.core.message;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+public class MessageFactoryKey implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ private final MessageFactoryType type;
+ private final String customPayloadSerializerClassName;
+
+ private MessageFactoryKey(MessageFactoryType type, String customPayloadSerializerClassName) {
+ this.type = type;
Review comment:
Yes, will be fixed in next push
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] galenwarren commented on a change in pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
galenwarren commented on a change in pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152#discussion_r492357429
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/StatefulFunctionsConfigValidator.java
##########
@@ -61,4 +64,26 @@ private static void validateParentFirstClassloaderPatterns(Configuration configu
}
return parentFirstClassloaderPatterns;
}
+
+ private static void validateCustomPayloadSerializerClassName(Configuration configuration) {
+
+ MessageFactoryType factoryType =
+ configuration.get(StatefulFunctionsConfig.USER_MESSAGE_SERIALIZER);
+ String customPayloadSerializerClassName =
+ configuration.get(StatefulFunctionsConfig.USER_MESSAGE_CUSTOM_PAYLOAD_SERIALIZER_CLASS);
+
+ if (factoryType == MessageFactoryType.WITH_CUSTOM_PAYLOADS) {
+ if (StringUtils.isNullOrWhitespaceOnly(customPayloadSerializerClassName)) {
Review comment:
OK will do
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] igalshilman commented on a change in pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
igalshilman commented on a change in pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152#discussion_r492616516
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/message/MessageFactoryKey.java
##########
@@ -0,0 +1,43 @@
+package org.apache.flink.statefun.flink.core.message;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+public class MessageFactoryKey implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ private final MessageFactoryType type;
+ private final String customPayloadSerializerClassName;
+
+ private MessageFactoryKey(MessageFactoryType type, String customPayloadSerializerClassName) {
+ this.type = type;
+ this.customPayloadSerializerClassName = customPayloadSerializerClassName;
+ }
+
+ public static MessageFactoryKey forType(
+ MessageFactoryType type, String customPayloadSerializerClassName) {
+ return new MessageFactoryKey(type, customPayloadSerializerClassName);
+ }
+
+ public MessageFactoryType getType() {
+ return this.type;
+ }
+
+ public String getCustomPayloadSerializerClassName() {
Review comment:
>
> So the serialized format would already seem to be what you've requested! But if you'd prefer it get written without using `StringUtils.writeNullableString`, I'm happy to make that change.
Oh I see. My original thinking here was to prevent a future NPE by someone who would expect (mistakenly) that `messageFactoryKey.getCustomPayloadSerializerClassName()` would always return a class name. Making the method return `Optional` would signal to the user of that class that it is not necessarily true.
And regarding `StringUtils` I'd rather to keep that logic local to the `Snapshot` class, to protect from accidental change in `StringUtils`.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] galenwarren commented on pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
galenwarren commented on pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152#issuecomment-695807576
@igalshilman
Thanks for the review! I've rebased on master and fixed a couple of the issues you identified, and I have questions on the other two, in the comments.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] galenwarren commented on pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
galenwarren commented on pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152#issuecomment-698391699
>I find it best (and also recommended by the Flink contribution guide) to address PR comments with follow-up commits on the PR branch, instead of squashing + rebasing + force-pushing.
Thanks for the feedback -- will do that next time.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] galenwarren commented on pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
galenwarren commented on pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152#issuecomment-695335939
Hi, I just wanted to check if you needed anything from me at this point?
I see a conflict has emerged on ```statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/StatefulFunctionsConfigTest.java```, let me know if you need me to do something there. Thanks.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] tzulitai commented on pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
tzulitai commented on pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152#issuecomment-697375778
> Also, a question about process -- am I supposed to "Resolve Conversation" on items I've taken care of, or am I supposed to leave those open for someone else to resolve? Just not sure what the protocol is.
You can resolve them yourself if you find them to be in a resolved state already.
Either way, committers will usually manually check that there are no outstanding discussions before merging.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] galenwarren commented on a change in pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
galenwarren commented on a change in pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152#discussion_r492358772
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/message/MessageFactoryKey.java
##########
@@ -0,0 +1,43 @@
+package org.apache.flink.statefun.flink.core.message;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+public class MessageFactoryKey implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ private final MessageFactoryType type;
+ private final String customPayloadSerializerClassName;
+
+ private MessageFactoryKey(MessageFactoryType type, String customPayloadSerializerClassName) {
+ this.type = type;
Review comment:
Yes, will be fixed in next push
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] galenwarren commented on a change in pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
galenwarren commented on a change in pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152#discussion_r493716907
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/message/MessageFactoryKey.java
##########
@@ -0,0 +1,43 @@
+package org.apache.flink.statefun.flink.core.message;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+public class MessageFactoryKey implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ private final MessageFactoryType type;
+ private final String customPayloadSerializerClassName;
+
+ private MessageFactoryKey(MessageFactoryType type, String customPayloadSerializerClassName) {
+ this.type = type;
+ this.customPayloadSerializerClassName = customPayloadSerializerClassName;
+ }
+
+ public static MessageFactoryKey forType(
+ MessageFactoryType type, String customPayloadSerializerClassName) {
+ return new MessageFactoryKey(type, customPayloadSerializerClassName);
+ }
+
+ public MessageFactoryType getType() {
+ return this.type;
+ }
+
+ public String getCustomPayloadSerializerClassName() {
Review comment:
These changes are in the code I just pushed, please let me know if it looks OK or if you need anything else. Thanks.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] galenwarren commented on a change in pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
galenwarren commented on a change in pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152#discussion_r491708474
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/message/MessageFactoryKey.java
##########
@@ -0,0 +1,43 @@
+package org.apache.flink.statefun.flink.core.message;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+public class MessageFactoryKey implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ private final MessageFactoryType type;
+ private final String customPayloadSerializerClassName;
+
+ private MessageFactoryKey(MessageFactoryType type, String customPayloadSerializerClassName) {
+ this.type = type;
+ this.customPayloadSerializerClassName = customPayloadSerializerClassName;
+ }
+
+ public static MessageFactoryKey forType(
+ MessageFactoryType type, String customPayloadSerializerClassName) {
+ return new MessageFactoryKey(type, customPayloadSerializerClassName);
+ }
+
+ public MessageFactoryType getType() {
+ return this.type;
+ }
+
+ public String getCustomPayloadSerializerClassName() {
Review comment:
As it stands now, this method gets called from MessageTypeSerializer.Snapshot.writeSnapshot, for all types of message serializers, not just custom ones. For non-custom ones, a null is returned and this is the value that is written into the snapshot, after the MesssageFactoryType. So, the version 2 serialized format is always a MessageFactoryType value and then the custom payload serializer class name (string, which can be null), for all types of serializers.
The writeSnapshot method could be changed to only call getCustomPayloadSerializerClassName in the event that the serializer is custom, and then to either 1) force a null serializer class name to be written to the snapshot or 2) skip the write of the serializer class name altogether in the non-custom case. This would make the read/writeSnapshot code and testing a bit more complicated but would definitely be doable, would one of those options be better?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] galenwarren commented on a change in pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
galenwarren commented on a change in pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152#discussion_r492358386
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/message/MessageFactoryKey.java
##########
@@ -0,0 +1,43 @@
+package org.apache.flink.statefun.flink.core.message;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+public class MessageFactoryKey implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ private final MessageFactoryType type;
+ private final String customPayloadSerializerClassName;
+
+ private MessageFactoryKey(MessageFactoryType type, String customPayloadSerializerClassName) {
+ this.type = type;
+ this.customPayloadSerializerClassName = customPayloadSerializerClassName;
+ }
+
+ public static MessageFactoryKey forType(
+ MessageFactoryType type, String customPayloadSerializerClassName) {
+ return new MessageFactoryKey(type, customPayloadSerializerClassName);
+ }
+
+ public MessageFactoryType getType() {
+ return this.type;
+ }
+
+ public String getCustomPayloadSerializerClassName() {
Review comment:
For some reason, before I couldn't comment here, but now I can. Weird glitch. Anyway, I left a top-level comment/question for you on this one, just FYI.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] galenwarren commented on pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
galenwarren commented on pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152#issuecomment-693484878
@tzulitai
Yes, of course. Sorry about that. Is this better?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] galenwarren commented on a change in pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
galenwarren commented on a change in pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152#discussion_r493610448
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/message/MessageFactoryKey.java
##########
@@ -0,0 +1,43 @@
+package org.apache.flink.statefun.flink.core.message;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+public class MessageFactoryKey implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ private final MessageFactoryType type;
+ private final String customPayloadSerializerClassName;
+
+ private MessageFactoryKey(MessageFactoryType type, String customPayloadSerializerClassName) {
+ this.type = type;
+ this.customPayloadSerializerClassName = customPayloadSerializerClassName;
+ }
+
+ public static MessageFactoryKey forType(
+ MessageFactoryType type, String customPayloadSerializerClassName) {
+ return new MessageFactoryKey(type, customPayloadSerializerClassName);
+ }
+
+ public MessageFactoryType getType() {
+ return this.type;
+ }
+
+ public String getCustomPayloadSerializerClassName() {
Review comment:
Great -- understood, I'll make that change and push a new version.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] igalshilman commented on a change in pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
igalshilman commented on a change in pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152#discussion_r492318196
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/message/MessageFactoryKey.java
##########
@@ -0,0 +1,43 @@
+package org.apache.flink.statefun.flink.core.message;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+public class MessageFactoryKey implements Serializable {
Review comment:
Can this be final?
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/message/MessageFactoryKey.java
##########
@@ -0,0 +1,43 @@
+package org.apache.flink.statefun.flink.core.message;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+public class MessageFactoryKey implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ private final MessageFactoryType type;
+ private final String customPayloadSerializerClassName;
+
+ private MessageFactoryKey(MessageFactoryType type, String customPayloadSerializerClassName) {
+ this.type = type;
Review comment:
Can this be: `this.type = Objects.requireNonNull(type);`
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/StatefulFunctionsConfigValidator.java
##########
@@ -61,4 +64,26 @@ private static void validateParentFirstClassloaderPatterns(Configuration configu
}
return parentFirstClassloaderPatterns;
}
+
+ private static void validateCustomPayloadSerializerClassName(Configuration configuration) {
+
+ MessageFactoryType factoryType =
+ configuration.get(StatefulFunctionsConfig.USER_MESSAGE_SERIALIZER);
+ String customPayloadSerializerClassName =
+ configuration.get(StatefulFunctionsConfig.USER_MESSAGE_CUSTOM_PAYLOAD_SERIALIZER_CLASS);
+
+ if (factoryType == MessageFactoryType.WITH_CUSTOM_PAYLOADS) {
+ if (StringUtils.isNullOrWhitespaceOnly(customPayloadSerializerClassName)) {
Review comment:
I think that you are right, let's leave that as it is 👍
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/message/MessageFactoryKey.java
##########
@@ -0,0 +1,43 @@
+package org.apache.flink.statefun.flink.core.message;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+public class MessageFactoryKey implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ private final MessageFactoryType type;
+ private final String customPayloadSerializerClassName;
+
+ private MessageFactoryKey(MessageFactoryType type, String customPayloadSerializerClassName) {
+ this.type = type;
+ this.customPayloadSerializerClassName = customPayloadSerializerClassName;
+ }
+
+ public static MessageFactoryKey forType(
+ MessageFactoryType type, String customPayloadSerializerClassName) {
+ return new MessageFactoryKey(type, customPayloadSerializerClassName);
+ }
+
+ public MessageFactoryType getType() {
+ return this.type;
+ }
+
+ public String getCustomPayloadSerializerClassName() {
Review comment:
That is a good point!
In that case, what do you think about making this method return an `Optional<String>`,
and version 2 would write an extra byte to indicate if the optional is present or not.
Something along the lines:
```
if (className.isPresent()) {
out.writeBooolean(true);
out.writeUtf8String(className.get());
} else {
out.writeBooolean(false);
}
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] tzulitai commented on pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
tzulitai commented on pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152#issuecomment-698128903
@galenwarren one comment about addressing PR reviews, for consideration in future contributions:
I find it best (and also recommended by the Flink contribution guide) to address PR comments with follow-up commits on the PR branch, instead of squashing + rebasing + force-pushing.
That way changes since the last review will be separate commits and will allow the reviewer to more easily track the changes.
Usually committers will be rebasing + squashing the follow-up commits when merging for you.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] galenwarren commented on pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
galenwarren commented on pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152#issuecomment-698391699
>I find it best (and also recommended by the Flink contribution guide) to address PR comments with follow-up commits on the PR branch, instead of squashing + rebasing + force-pushing.
Thanks for the feedback -- will do that next time.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] galenwarren commented on a change in pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
galenwarren commented on a change in pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152#discussion_r491707641
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/StatefulFunctionsConfigValidator.java
##########
@@ -61,4 +64,26 @@ private static void validateParentFirstClassloaderPatterns(Configuration configu
}
return parentFirstClassloaderPatterns;
}
+
+ private static void validateCustomPayloadSerializerClassName(Configuration configuration) {
+
+ MessageFactoryType factoryType =
+ configuration.get(StatefulFunctionsConfig.USER_MESSAGE_SERIALIZER);
+ String customPayloadSerializerClassName =
+ configuration.get(StatefulFunctionsConfig.USER_MESSAGE_CUSTOM_PAYLOAD_SERIALIZER_CLASS);
+
+ if (factoryType == MessageFactoryType.WITH_CUSTOM_PAYLOADS) {
+ if (StringUtils.isNullOrWhitespaceOnly(customPayloadSerializerClassName)) {
Review comment:
A question here -- as it stands now, the USER_MESSAGE_CUSTOM_PAYLOAD_SERIALIZER_CLASS config option has no default value, so if not set, its value will be null. So I believe I'd have to first check for null and then check length on the trimmed string. I'm happy to do that, but I thought I'd check first ...
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] galenwarren commented on pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
galenwarren commented on pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152#issuecomment-696388005
@igalshilman Thanks!
I'll take a look at the others shortly, but wanted to quickly reply regarding the question about Optional<String> and the serialized snapshot format. (I can't reply to the comment itself for some reason ...) What you propose is absolutely doable. I just wanted to make sure you're aware that the suggested code is essentially identical to the implementation of StringUtils.writeNullableString, which is what is currently used to write the version 2 snapshot in MessageTypeSerializer.Snapshot.writeSnapshot, i.e.:
```
@Override
public void writeSnapshot(DataOutputView dataOutputView) throws IOException {
// version 1
dataOutputView.writeUTF(messageFactoryKey.getType().name());
// added in version 2
StringUtils.writeNullableString(
messageFactoryKey.getCustomPayloadSerializerClassName(), dataOutputView);
}
```
... and ...
```
public static void writeNullableString(@Nullable String str, DataOutputView out) throws IOException {
if (str != null) {
out.writeBoolean(true);
writeString(str, out);
} else {
out.writeBoolean(false);
}
}
```
So the serialized format would already seem to be what you've requested! But if you'd prefer it get written without using ```StringUtils.writeNullableString```, I'm happy to make that change.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] galenwarren commented on pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
galenwarren commented on pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152#issuecomment-696388005
@igalshilman Thanks!
I'll take a look at the others shortly, but wanted to quickly reply regarding the question about Optional<String> and the serialized snapshot format. (I can't reply to the comment itself for some reason ...) What you propose is absolutely doable. I just wanted to make sure you're aware that the suggested code is essentially identical to the implementation of StringUtils.writeNullableString, which is what is currently used to write the version 2 snapshot in MessageTypeSerializer.Snapshot.writeSnapshot, i.e.:
```
@Override
public void writeSnapshot(DataOutputView dataOutputView) throws IOException {
// version 1
dataOutputView.writeUTF(messageFactoryKey.getType().name());
// added in version 2
StringUtils.writeNullableString(
messageFactoryKey.getCustomPayloadSerializerClassName(), dataOutputView);
}
```
... and ...
```
public static void writeNullableString(@Nullable String str, DataOutputView out) throws IOException {
if (str != null) {
out.writeBoolean(true);
writeString(str, out);
} else {
out.writeBoolean(false);
}
}
```
So the serialized format would already seem to be what you've requested! But if you'd prefer it get written without using ```StringUtils.writeNullableString```, I'm happy to make that change.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] tzulitai commented on pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
tzulitai commented on pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152#issuecomment-698127300
LGTM, merging this now to `master`!
> Also, I was wondering if you know when this might make it into a release. No hurry, I can use my fork until then, just wondering.
Since the feature freeze for 2.2 already happened, and this counts as a new feature, this will have to be included only in the next release (2.3).
The release cycles for StateFun are relatively shorter than Flink, so you can expect this to be officially available within 1~2 months when the next release is out.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] igalshilman commented on a change in pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
igalshilman commented on a change in pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152#discussion_r492318196
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/message/MessageFactoryKey.java
##########
@@ -0,0 +1,43 @@
+package org.apache.flink.statefun.flink.core.message;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+public class MessageFactoryKey implements Serializable {
Review comment:
Can this be final?
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/message/MessageFactoryKey.java
##########
@@ -0,0 +1,43 @@
+package org.apache.flink.statefun.flink.core.message;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+public class MessageFactoryKey implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ private final MessageFactoryType type;
+ private final String customPayloadSerializerClassName;
+
+ private MessageFactoryKey(MessageFactoryType type, String customPayloadSerializerClassName) {
+ this.type = type;
Review comment:
Can this be: `this.type = Objects.requireNonNull(type);`
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/StatefulFunctionsConfigValidator.java
##########
@@ -61,4 +64,26 @@ private static void validateParentFirstClassloaderPatterns(Configuration configu
}
return parentFirstClassloaderPatterns;
}
+
+ private static void validateCustomPayloadSerializerClassName(Configuration configuration) {
+
+ MessageFactoryType factoryType =
+ configuration.get(StatefulFunctionsConfig.USER_MESSAGE_SERIALIZER);
+ String customPayloadSerializerClassName =
+ configuration.get(StatefulFunctionsConfig.USER_MESSAGE_CUSTOM_PAYLOAD_SERIALIZER_CLASS);
+
+ if (factoryType == MessageFactoryType.WITH_CUSTOM_PAYLOADS) {
+ if (StringUtils.isNullOrWhitespaceOnly(customPayloadSerializerClassName)) {
Review comment:
I think that you are right, let's leave that as it is 👍
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/message/MessageFactoryKey.java
##########
@@ -0,0 +1,43 @@
+package org.apache.flink.statefun.flink.core.message;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+public class MessageFactoryKey implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ private final MessageFactoryType type;
+ private final String customPayloadSerializerClassName;
+
+ private MessageFactoryKey(MessageFactoryType type, String customPayloadSerializerClassName) {
+ this.type = type;
+ this.customPayloadSerializerClassName = customPayloadSerializerClassName;
+ }
+
+ public static MessageFactoryKey forType(
+ MessageFactoryType type, String customPayloadSerializerClassName) {
+ return new MessageFactoryKey(type, customPayloadSerializerClassName);
+ }
+
+ public MessageFactoryType getType() {
+ return this.type;
+ }
+
+ public String getCustomPayloadSerializerClassName() {
Review comment:
That is a good point!
In that case, what do you think about making this method return an `Optional<String>`,
and version 2 would write an extra byte to indicate if the optional is present or not.
Something along the lines:
```
if (className.isPresent()) {
out.writeBooolean(true);
out.writeUtf8String(className.get());
} else {
out.writeBooolean(false);
}
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] galenwarren commented on pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
galenwarren commented on pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152#issuecomment-697363116
@tzulitai Thanks! I'm excited to be able to make a contribution here.
I pushed an update yesterday that addresses all of @igalshilman 's remaining comments except the one about writeSnapshot and the Snapshot serialization format, which I have a question about (above). Depending on the answer to that, I might have a bit more work to do, or we might be done.
Also, a question about process -- am I supposed to "Resolve Conversation" on items I've taken care of, or am I supposed to leave those open for someone else to resolve? Just not sure what the protocol is.
Thanks both of you for your help on this.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] tzulitai closed pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
tzulitai closed pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] igalshilman commented on a change in pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
igalshilman commented on a change in pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152#discussion_r492318196
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/message/MessageFactoryKey.java
##########
@@ -0,0 +1,43 @@
+package org.apache.flink.statefun.flink.core.message;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+public class MessageFactoryKey implements Serializable {
Review comment:
Can this be final?
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/message/MessageFactoryKey.java
##########
@@ -0,0 +1,43 @@
+package org.apache.flink.statefun.flink.core.message;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+public class MessageFactoryKey implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ private final MessageFactoryType type;
+ private final String customPayloadSerializerClassName;
+
+ private MessageFactoryKey(MessageFactoryType type, String customPayloadSerializerClassName) {
+ this.type = type;
Review comment:
Can this be: `this.type = Objects.requireNonNull(type);`
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/StatefulFunctionsConfigValidator.java
##########
@@ -61,4 +64,26 @@ private static void validateParentFirstClassloaderPatterns(Configuration configu
}
return parentFirstClassloaderPatterns;
}
+
+ private static void validateCustomPayloadSerializerClassName(Configuration configuration) {
+
+ MessageFactoryType factoryType =
+ configuration.get(StatefulFunctionsConfig.USER_MESSAGE_SERIALIZER);
+ String customPayloadSerializerClassName =
+ configuration.get(StatefulFunctionsConfig.USER_MESSAGE_CUSTOM_PAYLOAD_SERIALIZER_CLASS);
+
+ if (factoryType == MessageFactoryType.WITH_CUSTOM_PAYLOADS) {
+ if (StringUtils.isNullOrWhitespaceOnly(customPayloadSerializerClassName)) {
Review comment:
I think that you are right, let's leave that as it is 👍
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/message/MessageFactoryKey.java
##########
@@ -0,0 +1,43 @@
+package org.apache.flink.statefun.flink.core.message;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+public class MessageFactoryKey implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ private final MessageFactoryType type;
+ private final String customPayloadSerializerClassName;
+
+ private MessageFactoryKey(MessageFactoryType type, String customPayloadSerializerClassName) {
+ this.type = type;
+ this.customPayloadSerializerClassName = customPayloadSerializerClassName;
+ }
+
+ public static MessageFactoryKey forType(
+ MessageFactoryType type, String customPayloadSerializerClassName) {
+ return new MessageFactoryKey(type, customPayloadSerializerClassName);
+ }
+
+ public MessageFactoryType getType() {
+ return this.type;
+ }
+
+ public String getCustomPayloadSerializerClassName() {
Review comment:
That is a good point!
In that case, what do you think about making this method return an `Optional<String>`,
and version 2 would write an extra byte to indicate if the optional is present or not.
Something along the lines:
```
if (className.isPresent()) {
out.writeBooolean(true);
out.writeUtf8String(className.get());
} else {
out.writeBooolean(false);
}
```
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/message/MessageFactoryKey.java
##########
@@ -0,0 +1,43 @@
+package org.apache.flink.statefun.flink.core.message;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+public class MessageFactoryKey implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ private final MessageFactoryType type;
+ private final String customPayloadSerializerClassName;
+
+ private MessageFactoryKey(MessageFactoryType type, String customPayloadSerializerClassName) {
+ this.type = type;
+ this.customPayloadSerializerClassName = customPayloadSerializerClassName;
+ }
+
+ public static MessageFactoryKey forType(
+ MessageFactoryType type, String customPayloadSerializerClassName) {
+ return new MessageFactoryKey(type, customPayloadSerializerClassName);
+ }
+
+ public MessageFactoryType getType() {
+ return this.type;
+ }
+
+ public String getCustomPayloadSerializerClassName() {
Review comment:
>
> So the serialized format would already seem to be what you've requested! But if you'd prefer it get written without using `StringUtils.writeNullableString`, I'm happy to make that change.
Oh I see. My original thinking here was to prevent a future NPE by someone who would expect (mistakenly) that `messageFactoryKey.getCustomPayloadSerializerClassName()` would always return a class name. Making the method return `Optional` would signal to the user of that class that it is not necessarily true.
And regarding `StringUtils` I'd rather to keep that logic local to the `Snapshot` class, to protect from accidental change in `StringUtils`.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] galenwarren commented on pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
galenwarren commented on pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152#issuecomment-697933207
@igalshilman
Great! Thanks for your help and the kind words. So does that mean we're done?
Also, I was wondering if you know when this might make it into a release. No hurry, I can use my fork until then, just wondering.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] galenwarren commented on pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
galenwarren commented on pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152#issuecomment-697804936
The remaining item is resolved with the latest push.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] galenwarren commented on a change in pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
galenwarren commented on a change in pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152#discussion_r491707641
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/StatefulFunctionsConfigValidator.java
##########
@@ -61,4 +64,26 @@ private static void validateParentFirstClassloaderPatterns(Configuration configu
}
return parentFirstClassloaderPatterns;
}
+
+ private static void validateCustomPayloadSerializerClassName(Configuration configuration) {
+
+ MessageFactoryType factoryType =
+ configuration.get(StatefulFunctionsConfig.USER_MESSAGE_SERIALIZER);
+ String customPayloadSerializerClassName =
+ configuration.get(StatefulFunctionsConfig.USER_MESSAGE_CUSTOM_PAYLOAD_SERIALIZER_CLASS);
+
+ if (factoryType == MessageFactoryType.WITH_CUSTOM_PAYLOADS) {
+ if (StringUtils.isNullOrWhitespaceOnly(customPayloadSerializerClassName)) {
Review comment:
A question here -- as it stands now, the USER_MESSAGE_CUSTOM_PAYLOAD_SERIALIZER_CLASS config option has no default value, so if not set, its value will be null. So I believe I'd have to first check for null and then check length on the trimmed string. I'm happy to do that, but I thought I'd check first in case that affects things ...
Or should USER_MESSAGE_CUSTOM_PAYLOAD_SERIALIZER_CLASS have an empty string as the default value?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] tzulitai edited a comment on pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
tzulitai edited a comment on pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152#issuecomment-698127300
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] tzulitai edited a comment on pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
tzulitai edited a comment on pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152#issuecomment-698127300
LGTM, merging this now to `master`!
> Also, I was wondering if you know when this might make it into a release. No hurry, I can use my fork until then, just wondering.
Since the feature freeze for 2.2 already happened, and this counts as a new feature, this will have to be included only in the next release (2.3).
The release cycles for StateFun are relatively shorter than Flink, so I guess you can expect this to be officially available within ~6 weeks when the next release is ready.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] tzulitai edited a comment on pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
tzulitai edited a comment on pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152#issuecomment-698127300
LGTM, merging this now to `master`!
> Also, I was wondering if you know when this might make it into a release. No hurry, I can use my fork until then, just wondering.
Since the feature freeze for 2.2 already happened, and this counts as a new feature, this will have to be included only in the next release (2.3).
The release cycles for StateFun are relatively shorter than Flink, so you can expect this to be officially available within 4~6 weeks when the next release is ready.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] tzulitai closed pull request #152: [FLINK-19176] Add pluggable statefun payload serializer
Posted by GitBox <gi...@apache.org>.
tzulitai closed pull request #152:
URL: https://github.com/apache/flink-statefun/pull/152
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org