dmvk commented on a change in pull request #18901:
URL: https://github.com/apache/flink/pull/18901#discussion_r815884834
##########
File path:
flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesStateHandleStore.java
##########
@@ -69,13 +75,82 @@
* the leader could update the store. Then we will completely get rid of the
lock-and-release in
* Zookeeper implementation.
*
- * @param <T> Type of state
+ * @param <T> Type of the state we're storing.
*/
public class KubernetesStateHandleStore<T extends Serializable>
implements StateHandleStore<T, StringResourceVersion> {
private static final Logger LOG =
LoggerFactory.getLogger(KubernetesStateHandleStore.class);
+ private static <T extends Serializable> StateHandleWithDeleteMarker<T>
deserializeStateHandle(
+ String content) throws IOException {
+ checkNotNull(content, "Content should not be null.");
+ final byte[] data = Base64.getDecoder().decode(content);
+ try {
+ return deserialize(data);
+ } catch (IOException | ClassNotFoundException e) {
+ throw new IOException(
+ String.format(
+ "Failed to deserialize state handle from ConfigMap
data %s.", content),
+ e);
+ }
+ }
+
+ private static String toBase64(byte[] bytes) {
+ return Base64.getEncoder().encodeToString(bytes);
+ }
+
+ @VisibleForTesting
+ static String serializeStateHandle(StateHandleWithDeleteMarker<?>
stateHandle)
Review comment:
What you're suggesting is basically an equivalent of
`InstantiationUtil#serializeObject`. The base64 encode thing is local to the
k8s implementation
--
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: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
