mxm commented on a change in pull request #12062:
URL: https://github.com/apache/beam/pull/12062#discussion_r445131904
##########
File path:
runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/state/InMemoryBagUserStateFactory.java
##########
@@ -37,13 +37,19 @@
/**
* Holds user state in memory. Only one key is active at a time due to the
GroupReduceFunction being
* called once per key. Needs to be reset via {@code resetForNewKey()} before
processing a new key.
+ *
+ * <p>In case of any failures, this factory must be discarded. Otherwise, the
contained state cache
+ * token would be reused which would corrupt the state cache.
*/
public class InMemoryBagUserStateFactory<K, V, W extends BoundedWindow>
implements StateRequestHandlers.BagUserStateHandlerFactory<K, V, W> {
+ private final ByteString cacheToken;
+
private List<InMemorySingleKeyBagState> handlers;
public InMemoryBagUserStateFactory() {
+ cacheToken =
ByteString.copyFrom(UUID.randomUUID().toString().getBytes(Charsets.UTF_8));
Review comment:
I see, makes sense. Let me update the PR.
----------------------------------------------------------------
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:
[email protected]
