You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by ke...@apache.org on 2017/02/06 22:44:10 UTC
[09/15] beam git commit: Move StateNamespace and related to
runners/core-java
Move StateNamespace and related to runners/core-java
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/e77e7f0d
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/e77e7f0d
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/e77e7f0d
Branch: refs/heads/master
Commit: e77e7f0d535eb8ad309f3210e3061566b55e7a21
Parents: 144b1df
Author: Kenneth Knowles <kl...@google.com>
Authored: Thu Jan 26 21:26:04 2017 -0800
Committer: Kenneth Knowles <kl...@google.com>
Committed: Mon Feb 6 09:26:56 2017 -0800
----------------------------------------------------------------------
.../operators/ApexGroupByKeyOperator.java | 2 +-
.../translation/utils/ApexStateInternals.java | 2 +-
.../utils/ApexStateInternalsTest.java | 6 +-
.../runners/core/InMemoryStateInternals.java | 1 -
.../runners/core/InMemoryTimerInternals.java | 1 -
.../runners/core/MergingActiveWindowSet.java | 1 -
.../runners/core/ReduceFnContextFactory.java | 4 +-
.../beam/runners/core/ReduceFnRunner.java | 2 +-
.../beam/runners/core/SideInputHandler.java | 1 -
.../beam/runners/core/SimpleDoFnRunner.java | 2 -
.../beam/runners/core/SplittableParDo.java | 2 -
.../beam/runners/core/StateInternals.java | 1 -
.../beam/runners/core/StateNamespace.java | 56 ++++
.../runners/core/StateNamespaceForTest.java | 65 +++++
.../beam/runners/core/StateNamespaces.java | 278 +++++++++++++++++++
.../apache/beam/runners/core/StateTable.java | 1 -
.../core/TestInMemoryStateInternals.java | 1 -
.../beam/runners/core/TimerInternals.java | 2 -
.../TriggerStateMachineContextFactory.java | 4 +-
.../core/InMemoryStateInternalsTest.java | 2 -
.../core/InMemoryTimerInternalsTest.java | 2 -
.../runners/core/KeyedWorkItemCoderTest.java | 1 -
.../core/PushbackSideInputDoFnRunnerTest.java | 1 -
.../beam/runners/core/ReduceFnTester.java | 2 -
.../beam/runners/core/SimpleDoFnRunnerTest.java | 1 -
.../beam/runners/core/StateNamespacesTest.java | 130 +++++++++
.../beam/runners/core/TimerInternalsTest.java | 2 -
.../triggers/TriggerStateMachineTester.java | 8 +-
.../CopyOnAccessInMemoryStateInternals.java | 2 +-
.../runners/direct/DirectTimerInternals.java | 2 +-
.../direct/StatefulParDoEvaluatorFactory.java | 4 +-
.../beam/runners/direct/WatermarkManager.java | 2 +-
.../CopyOnAccessInMemoryStateInternalsTest.java | 6 +-
.../direct/DirectTimerInternalsTest.java | 2 +-
...leManagerRemovingTransformEvaluatorTest.java | 2 +-
.../runners/direct/EvaluationContextTest.java | 2 +-
.../StatefulParDoEvaluatorFactoryTest.java | 4 +-
.../runners/direct/WatermarkManagerTest.java | 2 +-
.../wrappers/streaming/FlinkStateInternals.java | 2 +-
.../wrappers/streaming/WindowDoFnOperator.java | 2 +-
.../streaming/FlinkStateInternalsTest.java | 4 +-
.../beam/sdk/util/state/StateNamespace.java | 56 ----
.../sdk/util/state/StateNamespaceForTest.java | 65 -----
.../beam/sdk/util/state/StateNamespaces.java | 278 -------------------
.../sdk/util/state/StateNamespacesTest.java | 130 ---------
45 files changed, 559 insertions(+), 587 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/operators/ApexGroupByKeyOperator.java
----------------------------------------------------------------------
diff --git a/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/operators/ApexGroupByKeyOperator.java b/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/operators/ApexGroupByKeyOperator.java
index 274e807..3508c3e 100644
--- a/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/operators/ApexGroupByKeyOperator.java
+++ b/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/operators/ApexGroupByKeyOperator.java
@@ -47,6 +47,7 @@ import org.apache.beam.runners.core.KeyedWorkItems;
import org.apache.beam.runners.core.OldDoFn;
import org.apache.beam.runners.core.StateInternals;
import org.apache.beam.runners.core.StateInternalsFactory;
+import org.apache.beam.runners.core.StateNamespace;
import org.apache.beam.runners.core.SystemReduceFn;
import org.apache.beam.runners.core.TimerInternals;
import org.apache.beam.runners.core.WindowingInternals;
@@ -63,7 +64,6 @@ import org.apache.beam.sdk.util.CoderUtils;
import org.apache.beam.sdk.util.TimeDomain;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.util.WindowingStrategy;
-import org.apache.beam.sdk.util.state.StateNamespace;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionView;
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/utils/ApexStateInternals.java
----------------------------------------------------------------------
diff --git a/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/utils/ApexStateInternals.java b/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/utils/ApexStateInternals.java
index c9f2e39..544000d 100644
--- a/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/utils/ApexStateInternals.java
+++ b/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/utils/ApexStateInternals.java
@@ -30,6 +30,7 @@ import java.util.Arrays;
import java.util.List;
import org.apache.beam.runners.core.StateInternals;
import org.apache.beam.runners.core.StateInternalsFactory;
+import org.apache.beam.runners.core.StateNamespace;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.coders.Coder.Context;
import org.apache.beam.sdk.coders.InstantCoder;
@@ -46,7 +47,6 @@ import org.apache.beam.sdk.util.state.ReadableState;
import org.apache.beam.sdk.util.state.State;
import org.apache.beam.sdk.util.state.StateContext;
import org.apache.beam.sdk.util.state.StateContexts;
-import org.apache.beam.sdk.util.state.StateNamespace;
import org.apache.beam.sdk.util.state.StateTag;
import org.apache.beam.sdk.util.state.StateTag.StateBinder;
import org.apache.beam.sdk.util.state.ValueState;
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/apex/src/test/java/org/apache/beam/runners/apex/translation/utils/ApexStateInternalsTest.java
----------------------------------------------------------------------
diff --git a/runners/apex/src/test/java/org/apache/beam/runners/apex/translation/utils/ApexStateInternalsTest.java b/runners/apex/src/test/java/org/apache/beam/runners/apex/translation/utils/ApexStateInternalsTest.java
index 4d797f1..75f648b 100644
--- a/runners/apex/src/test/java/org/apache/beam/runners/apex/translation/utils/ApexStateInternalsTest.java
+++ b/runners/apex/src/test/java/org/apache/beam/runners/apex/translation/utils/ApexStateInternalsTest.java
@@ -23,9 +23,9 @@ import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import com.datatorrent.lib.util.KryoCloneUtils;
-
import java.util.Arrays;
-
+import org.apache.beam.runners.core.StateNamespace;
+import org.apache.beam.runners.core.StateNamespaceForTest;
import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.coders.VarIntCoder;
import org.apache.beam.sdk.transforms.Sum;
@@ -37,8 +37,6 @@ import org.apache.beam.sdk.util.state.BagState;
import org.apache.beam.sdk.util.state.CombiningState;
import org.apache.beam.sdk.util.state.ReadableState;
import org.apache.beam.sdk.util.state.StateMerging;
-import org.apache.beam.sdk.util.state.StateNamespace;
-import org.apache.beam.sdk.util.state.StateNamespaceForTest;
import org.apache.beam.sdk.util.state.StateTag;
import org.apache.beam.sdk.util.state.StateTags;
import org.apache.beam.sdk.util.state.ValueState;
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/core-java/src/main/java/org/apache/beam/runners/core/InMemoryStateInternals.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/InMemoryStateInternals.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/InMemoryStateInternals.java
index cf03750..4779954 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/InMemoryStateInternals.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/InMemoryStateInternals.java
@@ -37,7 +37,6 @@ import org.apache.beam.sdk.util.state.ReadableState;
import org.apache.beam.sdk.util.state.State;
import org.apache.beam.sdk.util.state.StateContext;
import org.apache.beam.sdk.util.state.StateContexts;
-import org.apache.beam.sdk.util.state.StateNamespace;
import org.apache.beam.sdk.util.state.StateTag;
import org.apache.beam.sdk.util.state.StateTag.StateBinder;
import org.apache.beam.sdk.util.state.ValueState;
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/core-java/src/main/java/org/apache/beam/runners/core/InMemoryTimerInternals.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/InMemoryTimerInternals.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/InMemoryTimerInternals.java
index bbe1a24..f8aaa09 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/InMemoryTimerInternals.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/InMemoryTimerInternals.java
@@ -30,7 +30,6 @@ import javax.annotation.Nullable;
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
import org.apache.beam.sdk.util.TimeDomain;
import org.apache.beam.sdk.util.WindowTracing;
-import org.apache.beam.sdk.util.state.StateNamespace;
import org.joda.time.Instant;
/** {@link TimerInternals} with all watermarks and processing clock simulated in-memory. */
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/core-java/src/main/java/org/apache/beam/runners/core/MergingActiveWindowSet.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/MergingActiveWindowSet.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/MergingActiveWindowSet.java
index e3480fa..806591d 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/MergingActiveWindowSet.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/MergingActiveWindowSet.java
@@ -38,7 +38,6 @@ import org.apache.beam.sdk.coders.MapCoder;
import org.apache.beam.sdk.coders.SetCoder;
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
import org.apache.beam.sdk.transforms.windowing.WindowFn;
-import org.apache.beam.sdk.util.state.StateNamespaces;
import org.apache.beam.sdk.util.state.StateTag;
import org.apache.beam.sdk.util.state.StateTags;
import org.apache.beam.sdk.util.state.ValueState;
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnContextFactory.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnContextFactory.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnContextFactory.java
index cc0f560..0d82514 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnContextFactory.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnContextFactory.java
@@ -24,6 +24,7 @@ import com.google.common.collect.ImmutableMap;
import java.util.Collection;
import java.util.Map;
import javax.annotation.Nullable;
+import org.apache.beam.runners.core.StateNamespaces.WindowNamespace;
import org.apache.beam.runners.core.TimerInternals.TimerData;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.options.PipelineOptions;
@@ -38,9 +39,6 @@ import org.apache.beam.sdk.util.state.State;
import org.apache.beam.sdk.util.state.StateAccessor;
import org.apache.beam.sdk.util.state.StateContext;
import org.apache.beam.sdk.util.state.StateContexts;
-import org.apache.beam.sdk.util.state.StateNamespace;
-import org.apache.beam.sdk.util.state.StateNamespaces;
-import org.apache.beam.sdk.util.state.StateNamespaces.WindowNamespace;
import org.apache.beam.sdk.util.state.StateTag;
import org.apache.beam.sdk.values.PCollectionView;
import org.joda.time.Instant;
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnRunner.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnRunner.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnRunner.java
index 091adad..34db752 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnRunner.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnRunner.java
@@ -37,6 +37,7 @@ import javax.annotation.Nullable;
import org.apache.beam.runners.core.GroupByKeyViaGroupByKeyOnly.GroupByKeyOnly;
import org.apache.beam.runners.core.ReduceFnContextFactory.OnTriggerCallbacks;
import org.apache.beam.runners.core.ReduceFnContextFactory.StateStyle;
+import org.apache.beam.runners.core.StateNamespaces.WindowNamespace;
import org.apache.beam.runners.core.TimerInternals.TimerData;
import org.apache.beam.runners.core.triggers.ExecutableTriggerStateMachine;
import org.apache.beam.runners.core.triggers.TriggerStateMachineContextFactory;
@@ -57,7 +58,6 @@ import org.apache.beam.sdk.util.WindowTracing;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.util.WindowingStrategy;
import org.apache.beam.sdk.util.WindowingStrategy.AccumulationMode;
-import org.apache.beam.sdk.util.state.StateNamespaces.WindowNamespace;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
import org.joda.time.Duration;
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/core-java/src/main/java/org/apache/beam/runners/core/SideInputHandler.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/SideInputHandler.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/SideInputHandler.java
index 6a7d03d..16324c1 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/SideInputHandler.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/SideInputHandler.java
@@ -32,7 +32,6 @@ import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
import org.apache.beam.sdk.util.ReadyCheckingSideInputReader;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.util.state.AccumulatorCombiningState;
-import org.apache.beam.sdk.util.state.StateNamespaces;
import org.apache.beam.sdk.util.state.StateTag;
import org.apache.beam.sdk.util.state.StateTags;
import org.apache.beam.sdk.util.state.ValueState;
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
index 9b1b852..8c9b8b7 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
@@ -55,8 +55,6 @@ import org.apache.beam.sdk.util.UserCodeException;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.util.WindowingStrategy;
import org.apache.beam.sdk.util.state.State;
-import org.apache.beam.sdk.util.state.StateNamespace;
-import org.apache.beam.sdk.util.state.StateNamespaces;
import org.apache.beam.sdk.util.state.StateSpec;
import org.apache.beam.sdk.util.state.StateTags;
import org.apache.beam.sdk.values.PCollectionView;
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/core-java/src/main/java/org/apache/beam/runners/core/SplittableParDo.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/SplittableParDo.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/SplittableParDo.java
index 39d7c56..a5eacaf 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/SplittableParDo.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/SplittableParDo.java
@@ -51,8 +51,6 @@ import org.apache.beam.sdk.transforms.windowing.OutputTimeFns;
import org.apache.beam.sdk.util.TimeDomain;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.util.WindowingStrategy;
-import org.apache.beam.sdk.util.state.StateNamespace;
-import org.apache.beam.sdk.util.state.StateNamespaces;
import org.apache.beam.sdk.util.state.StateTag;
import org.apache.beam.sdk.util.state.StateTags;
import org.apache.beam.sdk.util.state.ValueState;
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/core-java/src/main/java/org/apache/beam/runners/core/StateInternals.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/StateInternals.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/StateInternals.java
index 0b3bd62..7490c20 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/StateInternals.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/StateInternals.java
@@ -22,7 +22,6 @@ import org.apache.beam.sdk.annotations.Experimental.Kind;
import org.apache.beam.sdk.transforms.GroupByKey;
import org.apache.beam.sdk.util.state.State;
import org.apache.beam.sdk.util.state.StateContext;
-import org.apache.beam.sdk.util.state.StateNamespace;
import org.apache.beam.sdk.util.state.StateTag;
/**
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/core-java/src/main/java/org/apache/beam/runners/core/StateNamespace.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/StateNamespace.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/StateNamespace.java
new file mode 100644
index 0000000..e8293a0
--- /dev/null
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/StateNamespace.java
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.core;
+
+import java.io.IOException;
+
+/**
+ * A namespace used for scoping state stored with {@link StateInternals}.
+ *
+ * <p>Instances of {@link StateNamespace} are guaranteed to have a {@link Object#hashCode} and
+ * {@link Object#equals} that uniquely identify the namespace.
+ */
+public interface StateNamespace {
+
+ /**
+ * Return a {@link String} representation of the key. It is guaranteed that this
+ * {@link String} will uniquely identify the key.
+ *
+ * <p>This will encode the actual namespace as a {@link String}. It is
+ * preferable to use the {@link StateNamespace} object when possible.
+ *
+ * <p>The string produced by the standard implementations will not contain a '+' character. This
+ * enables adding a '+' between the actual namespace and other information, if needed, to separate
+ * the two.
+ */
+ String stringKey();
+
+ /**
+ * Append the string representation of this key to the {@link Appendable}.
+ */
+ void appendTo(Appendable sb) throws IOException;
+
+ /**
+ * Return an {@link Object} to use as a key in a cache.
+ *
+ * <p>Different namespaces may use the same key in order to be treated as a unit in the cache.
+ * The {@link Object}'s {@link Object#hashCode} and {@link Object#equals} methods will be used to
+ * determine equality.
+ */
+ Object getCacheKey();
+}
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/core-java/src/main/java/org/apache/beam/runners/core/StateNamespaceForTest.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/StateNamespaceForTest.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/StateNamespaceForTest.java
new file mode 100644
index 0000000..529519c
--- /dev/null
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/StateNamespaceForTest.java
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.core;
+
+import java.io.IOException;
+import java.util.Objects;
+
+/**
+ * A simple {@link StateNamespace} used for testing.
+ */
+public class StateNamespaceForTest implements StateNamespace {
+ private String key;
+
+ public StateNamespaceForTest(String key) {
+ this.key = key;
+ }
+
+ @Override
+ public String stringKey() {
+ return key;
+ }
+
+ @Override
+ public Object getCacheKey() {
+ return key;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+
+ if (!(obj instanceof StateNamespaceForTest)) {
+ return false;
+ }
+
+ return Objects.equals(this.key, ((StateNamespaceForTest) obj).key);
+ }
+
+ @Override
+ public int hashCode() {
+ return key.hashCode();
+ }
+
+ @Override
+ public void appendTo(Appendable sb) throws IOException {
+ sb.append(key);
+ }
+}
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/core-java/src/main/java/org/apache/beam/runners/core/StateNamespaces.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/StateNamespaces.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/StateNamespaces.java
new file mode 100644
index 0000000..e13c912
--- /dev/null
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/StateNamespaces.java
@@ -0,0 +1,278 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.core;
+
+import com.google.common.base.Splitter;
+import java.io.IOException;
+import java.util.List;
+import java.util.Objects;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.coders.CoderException;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.util.CoderUtils;
+
+/**
+ * Factory methods for creating the {@link StateNamespace StateNamespaces}.
+ */
+public class StateNamespaces {
+
+ private enum Namespace {
+ GLOBAL,
+ WINDOW,
+ WINDOW_AND_TRIGGER
+ }
+
+ public static StateNamespace global() {
+ return new GlobalNamespace();
+ }
+
+ public static <W extends BoundedWindow> StateNamespace window(Coder<W> windowCoder, W window) {
+ return new WindowNamespace<>(windowCoder, window);
+ }
+
+ public static <W extends BoundedWindow>
+ StateNamespace windowAndTrigger(Coder<W> windowCoder, W window, int triggerIdx) {
+ return new WindowAndTriggerNamespace<>(windowCoder, window, triggerIdx);
+ }
+
+ private StateNamespaces() {}
+
+ /**
+ * {@link StateNamespace} that is global to the current key being processed.
+ */
+ public static class GlobalNamespace implements StateNamespace {
+
+ private static final String GLOBAL_STRING = "/";
+
+ @Override
+ public String stringKey() {
+ return GLOBAL_STRING;
+ }
+
+ @Override
+ public Object getCacheKey() {
+ return GLOBAL_STRING;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ return obj == this || obj instanceof GlobalNamespace;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(Namespace.GLOBAL);
+ }
+
+ @Override
+ public String toString() {
+ return "Global";
+ }
+
+ @Override
+ public void appendTo(Appendable sb) throws IOException {
+ sb.append(GLOBAL_STRING);
+ }
+ }
+
+ /**
+ * {@link StateNamespace} that is scoped to a specific window.
+ */
+ public static class WindowNamespace<W extends BoundedWindow> implements StateNamespace {
+
+ private static final String WINDOW_FORMAT = "/%s/";
+
+ private Coder<W> windowCoder;
+ private W window;
+
+ private WindowNamespace(Coder<W> windowCoder, W window) {
+ this.windowCoder = windowCoder;
+ this.window = window;
+ }
+
+ public W getWindow() {
+ return window;
+ }
+
+ @Override
+ public String stringKey() {
+ try {
+ return String.format(WINDOW_FORMAT, CoderUtils.encodeToBase64(windowCoder, window));
+ } catch (CoderException e) {
+ throw new RuntimeException("Unable to generate string key from window " + window, e);
+ }
+ }
+
+ @Override
+ public void appendTo(Appendable sb) throws IOException {
+ sb.append('/').append(CoderUtils.encodeToBase64(windowCoder, window)).append('/');
+ }
+
+ /**
+ * State in the same window will all be evicted together.
+ */
+ @Override
+ public Object getCacheKey() {
+ return window;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj == this) {
+ return true;
+ }
+
+ if (!(obj instanceof WindowNamespace)) {
+ return false;
+ }
+
+ WindowNamespace<?> that = (WindowNamespace<?>) obj;
+ return Objects.equals(this.window, that.window);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(Namespace.WINDOW, window);
+ }
+
+ @Override
+ public String toString() {
+ return "Window(" + window + ")";
+ }
+ }
+
+ /**
+ * {@link StateNamespace} that is scoped to a particular window and trigger index.
+ */
+ public static class WindowAndTriggerNamespace<W extends BoundedWindow>
+ implements StateNamespace {
+
+ private static final String WINDOW_AND_TRIGGER_FORMAT = "/%s/%s/";
+
+ private static final int TRIGGER_RADIX = 36;
+ private Coder<W> windowCoder;
+ private W window;
+ private int triggerIndex;
+
+ private WindowAndTriggerNamespace(Coder<W> windowCoder, W window, int triggerIndex) {
+ this.windowCoder = windowCoder;
+ this.window = window;
+ this.triggerIndex = triggerIndex;
+ }
+
+ public W getWindow() {
+ return window;
+ }
+
+ public int getTriggerIndex() {
+ return triggerIndex;
+ }
+
+ @Override
+ public String stringKey() {
+ try {
+ return String.format(WINDOW_AND_TRIGGER_FORMAT,
+ CoderUtils.encodeToBase64(windowCoder, window),
+ // Use base 36 so that can address 36 triggers in a single byte and still be human
+ // readable.
+ Integer.toString(triggerIndex, TRIGGER_RADIX).toUpperCase());
+ } catch (CoderException e) {
+ throw new RuntimeException("Unable to generate string key from window " + window, e);
+ }
+ }
+
+ @Override
+ public void appendTo(Appendable sb) throws IOException {
+ sb.append('/').append(CoderUtils.encodeToBase64(windowCoder, window));
+ sb.append('/').append(Integer.toString(triggerIndex, TRIGGER_RADIX).toUpperCase());
+ sb.append('/');
+ }
+
+ /**
+ * State in the same window will all be evicted together.
+ */
+ @Override
+ public Object getCacheKey() {
+ return window;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj == this) {
+ return true;
+ }
+
+ if (!(obj instanceof WindowAndTriggerNamespace)) {
+ return false;
+ }
+
+ WindowAndTriggerNamespace<?> that = (WindowAndTriggerNamespace<?>) obj;
+ return this.triggerIndex == that.triggerIndex
+ && Objects.equals(this.window, that.window);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(Namespace.WINDOW_AND_TRIGGER, window, triggerIndex);
+ }
+
+ @Override
+ public String toString() {
+ return "WindowAndTrigger(" + window + "," + triggerIndex + ")";
+ }
+ }
+
+ private static final Splitter SLASH_SPLITTER = Splitter.on('/');
+
+ /**
+ * Convert a {@code stringKey} produced using {@link StateNamespace#stringKey}
+ * on one of the namespaces produced by this class into the original
+ * {@link StateNamespace}.
+ */
+ public static <W extends BoundedWindow> StateNamespace fromString(
+ String stringKey, Coder<W> windowCoder) {
+ if (!stringKey.startsWith("/") || !stringKey.endsWith("/")) {
+ throw new RuntimeException("Invalid namespace string: '" + stringKey + "'");
+ }
+
+ if (GlobalNamespace.GLOBAL_STRING.equals(stringKey)) {
+ return global();
+ }
+
+ List<String> parts = SLASH_SPLITTER.splitToList(stringKey);
+ if (parts.size() != 3 && parts.size() != 4) {
+ throw new RuntimeException("Invalid namespace string: '" + stringKey + "'");
+ }
+ // Ends should be empty (we start and end with /)
+ if (!parts.get(0).isEmpty() || !parts.get(parts.size() - 1).isEmpty()) {
+ throw new RuntimeException("Invalid namespace string: '" + stringKey + "'");
+ }
+
+ try {
+ W window = CoderUtils.decodeFromBase64(windowCoder, parts.get(1));
+ if (parts.size() > 3) {
+ int index = Integer.parseInt(parts.get(2), WindowAndTriggerNamespace.TRIGGER_RADIX);
+ return windowAndTrigger(windowCoder, window, index);
+ } else {
+ return window(windowCoder, window);
+ }
+ } catch (Exception e) {
+ throw new RuntimeException("Invalid namespace string: '" + stringKey + "'", e);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/core-java/src/main/java/org/apache/beam/runners/core/StateTable.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/StateTable.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/StateTable.java
index 84b3016..48ebea3 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/StateTable.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/StateTable.java
@@ -23,7 +23,6 @@ import java.util.Map;
import java.util.Set;
import org.apache.beam.sdk.util.state.State;
import org.apache.beam.sdk.util.state.StateContext;
-import org.apache.beam.sdk.util.state.StateNamespace;
import org.apache.beam.sdk.util.state.StateTag;
import org.apache.beam.sdk.util.state.StateTag.StateBinder;
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/core-java/src/main/java/org/apache/beam/runners/core/TestInMemoryStateInternals.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/TestInMemoryStateInternals.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/TestInMemoryStateInternals.java
index 60754e2..ed852a5 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/TestInMemoryStateInternals.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/TestInMemoryStateInternals.java
@@ -21,7 +21,6 @@ import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import org.apache.beam.sdk.util.state.State;
-import org.apache.beam.sdk.util.state.StateNamespace;
import org.apache.beam.sdk.util.state.StateTag;
import org.apache.beam.sdk.util.state.WatermarkHoldState;
import org.joda.time.Instant;
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/core-java/src/main/java/org/apache/beam/runners/core/TimerInternals.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/TimerInternals.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/TimerInternals.java
index a50a622..9c7bd57 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/TimerInternals.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/TimerInternals.java
@@ -38,8 +38,6 @@ import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
import org.apache.beam.sdk.util.PropertyNames;
import org.apache.beam.sdk.util.TimeDomain;
-import org.apache.beam.sdk.util.state.StateNamespace;
-import org.apache.beam.sdk.util.state.StateNamespaces;
import org.joda.time.Instant;
/**
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/TriggerStateMachineContextFactory.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/TriggerStateMachineContextFactory.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/TriggerStateMachineContextFactory.java
index 2ac9065..1d96519 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/TriggerStateMachineContextFactory.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/TriggerStateMachineContextFactory.java
@@ -26,6 +26,8 @@ import java.util.Map;
import javax.annotation.Nullable;
import org.apache.beam.runners.core.ActiveWindowSet;
import org.apache.beam.runners.core.StateInternals;
+import org.apache.beam.runners.core.StateNamespace;
+import org.apache.beam.runners.core.StateNamespaces;
import org.apache.beam.runners.core.triggers.TriggerStateMachine.MergingTriggerInfo;
import org.apache.beam.runners.core.triggers.TriggerStateMachine.TriggerInfo;
import org.apache.beam.sdk.coders.Coder;
@@ -36,8 +38,6 @@ import org.apache.beam.sdk.util.Timers;
import org.apache.beam.sdk.util.state.MergingStateAccessor;
import org.apache.beam.sdk.util.state.State;
import org.apache.beam.sdk.util.state.StateAccessor;
-import org.apache.beam.sdk.util.state.StateNamespace;
-import org.apache.beam.sdk.util.state.StateNamespaces;
import org.apache.beam.sdk.util.state.StateTag;
import org.joda.time.Instant;
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/core-java/src/test/java/org/apache/beam/runners/core/InMemoryStateInternalsTest.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/InMemoryStateInternalsTest.java b/runners/core-java/src/test/java/org/apache/beam/runners/core/InMemoryStateInternalsTest.java
index c75399c..8bcd177 100644
--- a/runners/core-java/src/test/java/org/apache/beam/runners/core/InMemoryStateInternalsTest.java
+++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/InMemoryStateInternalsTest.java
@@ -33,8 +33,6 @@ import org.apache.beam.sdk.util.state.BagState;
import org.apache.beam.sdk.util.state.CombiningState;
import org.apache.beam.sdk.util.state.ReadableState;
import org.apache.beam.sdk.util.state.StateMerging;
-import org.apache.beam.sdk.util.state.StateNamespace;
-import org.apache.beam.sdk.util.state.StateNamespaceForTest;
import org.apache.beam.sdk.util.state.StateTag;
import org.apache.beam.sdk.util.state.StateTags;
import org.apache.beam.sdk.util.state.ValueState;
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/core-java/src/test/java/org/apache/beam/runners/core/InMemoryTimerInternalsTest.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/InMemoryTimerInternalsTest.java b/runners/core-java/src/test/java/org/apache/beam/runners/core/InMemoryTimerInternalsTest.java
index a060d48..0226260 100644
--- a/runners/core-java/src/test/java/org/apache/beam/runners/core/InMemoryTimerInternalsTest.java
+++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/InMemoryTimerInternalsTest.java
@@ -23,8 +23,6 @@ import static org.junit.Assert.assertThat;
import org.apache.beam.runners.core.TimerInternals.TimerData;
import org.apache.beam.sdk.util.TimeDomain;
-import org.apache.beam.sdk.util.state.StateNamespace;
-import org.apache.beam.sdk.util.state.StateNamespaceForTest;
import org.joda.time.Instant;
import org.junit.Test;
import org.junit.runner.RunWith;
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/core-java/src/test/java/org/apache/beam/runners/core/KeyedWorkItemCoderTest.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/KeyedWorkItemCoderTest.java b/runners/core-java/src/test/java/org/apache/beam/runners/core/KeyedWorkItemCoderTest.java
index 3e8edbd..4d59634 100644
--- a/runners/core-java/src/test/java/org/apache/beam/runners/core/KeyedWorkItemCoderTest.java
+++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/KeyedWorkItemCoderTest.java
@@ -25,7 +25,6 @@ import org.apache.beam.sdk.testing.CoderProperties;
import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
import org.apache.beam.sdk.util.TimeDomain;
import org.apache.beam.sdk.util.WindowedValue;
-import org.apache.beam.sdk.util.state.StateNamespaces;
import org.joda.time.Instant;
import org.junit.Test;
import org.junit.runner.RunWith;
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/core-java/src/test/java/org/apache/beam/runners/core/PushbackSideInputDoFnRunnerTest.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/PushbackSideInputDoFnRunnerTest.java b/runners/core-java/src/test/java/org/apache/beam/runners/core/PushbackSideInputDoFnRunnerTest.java
index f3327c2..cb057b8 100644
--- a/runners/core-java/src/test/java/org/apache/beam/runners/core/PushbackSideInputDoFnRunnerTest.java
+++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/PushbackSideInputDoFnRunnerTest.java
@@ -41,7 +41,6 @@ import org.apache.beam.sdk.util.IdentitySideInputWindowFn;
import org.apache.beam.sdk.util.ReadyCheckingSideInputReader;
import org.apache.beam.sdk.util.TimeDomain;
import org.apache.beam.sdk.util.WindowedValue;
-import org.apache.beam.sdk.util.state.StateNamespaces;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionView;
import org.hamcrest.Matchers;
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/core-java/src/test/java/org/apache/beam/runners/core/ReduceFnTester.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/ReduceFnTester.java b/runners/core-java/src/test/java/org/apache/beam/runners/core/ReduceFnTester.java
index 61707ad..bea0c39 100644
--- a/runners/core-java/src/test/java/org/apache/beam/runners/core/ReduceFnTester.java
+++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/ReduceFnTester.java
@@ -71,8 +71,6 @@ import org.apache.beam.sdk.util.WindowTracing;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.util.WindowingStrategy;
import org.apache.beam.sdk.util.WindowingStrategy.AccumulationMode;
-import org.apache.beam.sdk.util.state.StateNamespace;
-import org.apache.beam.sdk.util.state.StateNamespaces;
import org.apache.beam.sdk.util.state.StateTag;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.TimestampedValue;
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/core-java/src/test/java/org/apache/beam/runners/core/SimpleDoFnRunnerTest.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/SimpleDoFnRunnerTest.java b/runners/core-java/src/test/java/org/apache/beam/runners/core/SimpleDoFnRunnerTest.java
index 8deda5f..193722c 100644
--- a/runners/core-java/src/test/java/org/apache/beam/runners/core/SimpleDoFnRunnerTest.java
+++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/SimpleDoFnRunnerTest.java
@@ -42,7 +42,6 @@ import org.apache.beam.sdk.util.TimerSpecs;
import org.apache.beam.sdk.util.UserCodeException;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.util.WindowingStrategy;
-import org.apache.beam.sdk.util.state.StateNamespaces;
import org.apache.beam.sdk.values.TupleTag;
import org.joda.time.Duration;
import org.joda.time.Instant;
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/core-java/src/test/java/org/apache/beam/runners/core/StateNamespacesTest.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/StateNamespacesTest.java b/runners/core-java/src/test/java/org/apache/beam/runners/core/StateNamespacesTest.java
new file mode 100644
index 0000000..dee94cb
--- /dev/null
+++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/StateNamespacesTest.java
@@ -0,0 +1,130 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.core;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThat;
+
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
+import org.apache.beam.sdk.transforms.windowing.IntervalWindow;
+import org.hamcrest.Matchers;
+import org.joda.time.Instant;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/**
+ * Tests for {@link StateNamespaces}.
+ */
+@RunWith(JUnit4.class)
+public class StateNamespacesTest {
+
+ private final Coder<IntervalWindow> intervalCoder = IntervalWindow.getCoder();
+
+ private IntervalWindow intervalWindow(long start, long end) {
+ return new IntervalWindow(new Instant(start), new Instant(end));
+ }
+
+ /**
+ * This test should not be changed. It verifies that the stringKey matches certain expectations.
+ * If this changes, the ability to reload any pipeline that has persisted these namespaces will
+ * be impacted.
+ */
+ @Test
+ public void testStability() {
+ StateNamespace global = StateNamespaces.global();
+ StateNamespace intervalWindow =
+ StateNamespaces.window(intervalCoder, intervalWindow(1000, 87392));
+ StateNamespace intervalWindowAndTrigger =
+ StateNamespaces.windowAndTrigger(intervalCoder, intervalWindow(1000, 87392), 57);
+ StateNamespace globalWindow = StateNamespaces.window(
+ GlobalWindow.Coder.INSTANCE, GlobalWindow.INSTANCE);
+ StateNamespace globalWindowAndTrigger = StateNamespaces.windowAndTrigger(
+ GlobalWindow.Coder.INSTANCE, GlobalWindow.INSTANCE, 12);
+
+ assertEquals("/", global.stringKey());
+ assertEquals("/gAAAAAABVWD4ogU/", intervalWindow.stringKey());
+ assertEquals("/gAAAAAABVWD4ogU/1L/", intervalWindowAndTrigger.stringKey());
+ assertEquals("//", globalWindow.stringKey());
+ assertEquals("//C/", globalWindowAndTrigger.stringKey());
+ }
+
+ /**
+ * Test that WindowAndTrigger namespaces are prefixed by the related Window namespace.
+ */
+ @Test
+ public void testIntervalWindowPrefixing() {
+ StateNamespace window =
+ StateNamespaces.window(intervalCoder, intervalWindow(1000, 87392));
+ StateNamespace windowAndTrigger = StateNamespaces.windowAndTrigger(
+ intervalCoder, intervalWindow(1000, 87392), 57);
+ assertThat(windowAndTrigger.stringKey(), Matchers.startsWith(window.stringKey()));
+ assertThat(StateNamespaces.global().stringKey(),
+ Matchers.not(Matchers.startsWith(window.stringKey())));
+ }
+
+ /**
+ * Test that WindowAndTrigger namespaces are prefixed by the related Window namespace.
+ */
+ @Test
+ public void testGlobalWindowPrefixing() {
+ StateNamespace window =
+ StateNamespaces.window(GlobalWindow.Coder.INSTANCE, GlobalWindow.INSTANCE);
+ StateNamespace windowAndTrigger = StateNamespaces.windowAndTrigger(
+ GlobalWindow.Coder.INSTANCE, GlobalWindow.INSTANCE, 57);
+ assertThat(windowAndTrigger.stringKey(), Matchers.startsWith(window.stringKey()));
+ assertThat(StateNamespaces.global().stringKey(),
+ Matchers.not(Matchers.startsWith(window.stringKey())));
+ }
+
+ @Test
+ public void testFromStringGlobal() {
+ assertStringKeyRoundTrips(intervalCoder, StateNamespaces.global());
+ }
+
+ @Test
+ public void testFromStringIntervalWindow() {
+ assertStringKeyRoundTrips(
+ intervalCoder, StateNamespaces.window(intervalCoder, intervalWindow(1000, 8000)));
+ assertStringKeyRoundTrips(
+ intervalCoder, StateNamespaces.window(intervalCoder, intervalWindow(1000, 8000)));
+
+ assertStringKeyRoundTrips(intervalCoder,
+ StateNamespaces.windowAndTrigger(intervalCoder, intervalWindow(1000, 8000), 18));
+ assertStringKeyRoundTrips(intervalCoder,
+ StateNamespaces.windowAndTrigger(intervalCoder, intervalWindow(1000, 8000), 19));
+ assertStringKeyRoundTrips(intervalCoder,
+ StateNamespaces.windowAndTrigger(intervalCoder, intervalWindow(2000, 8000), 19));
+ }
+
+ @Test
+ public void testFromStringGlobalWindow() {
+ assertStringKeyRoundTrips(GlobalWindow.Coder.INSTANCE, StateNamespaces.global());
+ assertStringKeyRoundTrips(GlobalWindow.Coder.INSTANCE,
+ StateNamespaces.window(GlobalWindow.Coder.INSTANCE, GlobalWindow.INSTANCE));
+ assertStringKeyRoundTrips(GlobalWindow.Coder.INSTANCE,
+ StateNamespaces.windowAndTrigger(GlobalWindow.Coder.INSTANCE, GlobalWindow.INSTANCE, 18));
+ }
+
+ private void assertStringKeyRoundTrips(
+ Coder<? extends BoundedWindow> coder, StateNamespace namespace) {
+ assertEquals(namespace, StateNamespaces.fromString(namespace.stringKey(), coder));
+ }
+}
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/core-java/src/test/java/org/apache/beam/runners/core/TimerInternalsTest.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/TimerInternalsTest.java b/runners/core-java/src/test/java/org/apache/beam/runners/core/TimerInternalsTest.java
index a0ef505..768c974 100644
--- a/runners/core-java/src/test/java/org/apache/beam/runners/core/TimerInternalsTest.java
+++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/TimerInternalsTest.java
@@ -29,8 +29,6 @@ import org.apache.beam.sdk.testing.CoderProperties;
import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
import org.apache.beam.sdk.transforms.windowing.IntervalWindow;
import org.apache.beam.sdk.util.TimeDomain;
-import org.apache.beam.sdk.util.state.StateNamespace;
-import org.apache.beam.sdk.util.state.StateNamespaces;
import org.joda.time.Instant;
import org.junit.Test;
import org.junit.runner.RunWith;
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/core-java/src/test/java/org/apache/beam/runners/core/triggers/TriggerStateMachineTester.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/triggers/TriggerStateMachineTester.java b/runners/core-java/src/test/java/org/apache/beam/runners/core/triggers/TriggerStateMachineTester.java
index cd6d394..1bdcff2 100644
--- a/runners/core-java/src/test/java/org/apache/beam/runners/core/triggers/TriggerStateMachineTester.java
+++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/triggers/TriggerStateMachineTester.java
@@ -38,6 +38,10 @@ import org.apache.beam.runners.core.InMemoryTimerInternals;
import org.apache.beam.runners.core.MergingActiveWindowSet;
import org.apache.beam.runners.core.NonMergingActiveWindowSet;
import org.apache.beam.runners.core.StateInternals;
+import org.apache.beam.runners.core.StateNamespace;
+import org.apache.beam.runners.core.StateNamespaces;
+import org.apache.beam.runners.core.StateNamespaces.WindowAndTriggerNamespace;
+import org.apache.beam.runners.core.StateNamespaces.WindowNamespace;
import org.apache.beam.runners.core.TestInMemoryStateInternals;
import org.apache.beam.runners.core.TimerInternals.TimerData;
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
@@ -49,10 +53,6 @@ import org.apache.beam.sdk.util.Timers;
import org.apache.beam.sdk.util.WindowTracing;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.util.WindowingStrategy.AccumulationMode;
-import org.apache.beam.sdk.util.state.StateNamespace;
-import org.apache.beam.sdk.util.state.StateNamespaces;
-import org.apache.beam.sdk.util.state.StateNamespaces.WindowAndTriggerNamespace;
-import org.apache.beam.sdk.util.state.StateNamespaces.WindowNamespace;
import org.apache.beam.sdk.values.TimestampedValue;
import org.joda.time.Duration;
import org.joda.time.Instant;
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/direct-java/src/main/java/org/apache/beam/runners/direct/CopyOnAccessInMemoryStateInternals.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/CopyOnAccessInMemoryStateInternals.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/CopyOnAccessInMemoryStateInternals.java
index e809823..4b59bf9 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/CopyOnAccessInMemoryStateInternals.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/CopyOnAccessInMemoryStateInternals.java
@@ -32,6 +32,7 @@ import org.apache.beam.runners.core.InMemoryStateInternals.InMemoryStateBinder;
import org.apache.beam.runners.core.InMemoryStateInternals.InMemoryValue;
import org.apache.beam.runners.core.InMemoryStateInternals.InMemoryWatermarkHold;
import org.apache.beam.runners.core.StateInternals;
+import org.apache.beam.runners.core.StateNamespace;
import org.apache.beam.runners.core.StateTable;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.transforms.Combine.CombineFn;
@@ -45,7 +46,6 @@ import org.apache.beam.sdk.util.state.BagState;
import org.apache.beam.sdk.util.state.State;
import org.apache.beam.sdk.util.state.StateContext;
import org.apache.beam.sdk.util.state.StateContexts;
-import org.apache.beam.sdk.util.state.StateNamespace;
import org.apache.beam.sdk.util.state.StateTag;
import org.apache.beam.sdk.util.state.StateTag.StateBinder;
import org.apache.beam.sdk.util.state.ValueState;
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectTimerInternals.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectTimerInternals.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectTimerInternals.java
index 233c3b8..67d568d 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectTimerInternals.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectTimerInternals.java
@@ -18,12 +18,12 @@
package org.apache.beam.runners.direct;
import javax.annotation.Nullable;
+import org.apache.beam.runners.core.StateNamespace;
import org.apache.beam.runners.core.TimerInternals;
import org.apache.beam.runners.direct.WatermarkManager.TimerUpdate;
import org.apache.beam.runners.direct.WatermarkManager.TimerUpdate.TimerUpdateBuilder;
import org.apache.beam.runners.direct.WatermarkManager.TransformWatermarks;
import org.apache.beam.sdk.util.TimeDomain;
-import org.apache.beam.sdk.util.state.StateNamespace;
import org.joda.time.Instant;
/**
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/direct-java/src/main/java/org/apache/beam/runners/direct/StatefulParDoEvaluatorFactory.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/StatefulParDoEvaluatorFactory.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/StatefulParDoEvaluatorFactory.java
index 8dd1657..b3a2156 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/StatefulParDoEvaluatorFactory.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/StatefulParDoEvaluatorFactory.java
@@ -28,6 +28,8 @@ import java.util.HashMap;
import java.util.Map;
import org.apache.beam.runners.core.KeyedWorkItem;
import org.apache.beam.runners.core.KeyedWorkItems;
+import org.apache.beam.runners.core.StateNamespace;
+import org.apache.beam.runners.core.StateNamespaces;
import org.apache.beam.runners.core.TimerInternals.TimerData;
import org.apache.beam.runners.direct.DirectExecutionContext.DirectStepContext;
import org.apache.beam.runners.direct.DirectRunner.CommittedBundle;
@@ -42,8 +44,6 @@ import org.apache.beam.sdk.transforms.reflect.DoFnSignatures;
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.util.WindowingStrategy;
-import org.apache.beam.sdk.util.state.StateNamespace;
-import org.apache.beam.sdk.util.state.StateNamespaces;
import org.apache.beam.sdk.util.state.StateSpec;
import org.apache.beam.sdk.util.state.StateTag;
import org.apache.beam.sdk.util.state.StateTags;
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WatermarkManager.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WatermarkManager.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WatermarkManager.java
index a653858..2526d2d 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WatermarkManager.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WatermarkManager.java
@@ -50,6 +50,7 @@ import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import javax.annotation.Nullable;
import javax.annotation.concurrent.GuardedBy;
+import org.apache.beam.runners.core.StateNamespace;
import org.apache.beam.runners.core.TimerInternals;
import org.apache.beam.runners.core.TimerInternals.TimerData;
import org.apache.beam.runners.direct.DirectRunner.CommittedBundle;
@@ -58,7 +59,6 @@ import org.apache.beam.sdk.transforms.AppliedPTransform;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
import org.apache.beam.sdk.util.TimeDomain;
-import org.apache.beam.sdk.util.state.StateNamespace;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.TaggedPValue;
import org.joda.time.Instant;
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/direct-java/src/test/java/org/apache/beam/runners/direct/CopyOnAccessInMemoryStateInternalsTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/CopyOnAccessInMemoryStateInternalsTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/CopyOnAccessInMemoryStateInternalsTest.java
index 4284f3e..25cd252 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/CopyOnAccessInMemoryStateInternalsTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/CopyOnAccessInMemoryStateInternalsTest.java
@@ -28,6 +28,9 @@ import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
+import org.apache.beam.runners.core.StateNamespace;
+import org.apache.beam.runners.core.StateNamespaceForTest;
+import org.apache.beam.runners.core.StateNamespaces;
import org.apache.beam.sdk.coders.CannotProvideCoderException;
import org.apache.beam.sdk.coders.CoderRegistry;
import org.apache.beam.sdk.coders.StringUtf8Coder;
@@ -42,9 +45,6 @@ import org.apache.beam.sdk.transforms.windowing.OutputTimeFns;
import org.apache.beam.sdk.util.state.AccumulatorCombiningState;
import org.apache.beam.sdk.util.state.BagState;
import org.apache.beam.sdk.util.state.CombiningState;
-import org.apache.beam.sdk.util.state.StateNamespace;
-import org.apache.beam.sdk.util.state.StateNamespaceForTest;
-import org.apache.beam.sdk.util.state.StateNamespaces;
import org.apache.beam.sdk.util.state.StateTag;
import org.apache.beam.sdk.util.state.StateTags;
import org.apache.beam.sdk.util.state.ValueState;
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectTimerInternalsTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectTimerInternalsTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectTimerInternalsTest.java
index 4d48489..c8ac695 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectTimerInternalsTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectTimerInternalsTest.java
@@ -22,13 +22,13 @@ import static org.hamcrest.Matchers.equalTo;
import static org.junit.Assert.assertThat;
import static org.mockito.Mockito.when;
+import org.apache.beam.runners.core.StateNamespaces;
import org.apache.beam.runners.core.TimerInternals.TimerData;
import org.apache.beam.runners.direct.WatermarkManager.TimerUpdate;
import org.apache.beam.runners.direct.WatermarkManager.TimerUpdate.TimerUpdateBuilder;
import org.apache.beam.runners.direct.WatermarkManager.TransformWatermarks;
import org.apache.beam.sdk.coders.VarIntCoder;
import org.apache.beam.sdk.util.TimeDomain;
-import org.apache.beam.sdk.util.state.StateNamespaces;
import org.joda.time.Duration;
import org.joda.time.Instant;
import org.junit.Before;
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DoFnLifecycleManagerRemovingTransformEvaluatorTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DoFnLifecycleManagerRemovingTransformEvaluatorTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DoFnLifecycleManagerRemovingTransformEvaluatorTest.java
index 26b4ef1..d046ce5 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DoFnLifecycleManagerRemovingTransformEvaluatorTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DoFnLifecycleManagerRemovingTransformEvaluatorTest.java
@@ -27,13 +27,13 @@ import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
+import org.apache.beam.runners.core.StateNamespaces;
import org.apache.beam.runners.core.TimerInternals.TimerData;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
import org.apache.beam.sdk.util.TimeDomain;
import org.apache.beam.sdk.util.WindowedValue;
-import org.apache.beam.sdk.util.state.StateNamespaces;
import org.hamcrest.Matchers;
import org.joda.time.Instant;
import org.junit.Before;
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/direct-java/src/test/java/org/apache/beam/runners/direct/EvaluationContextTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/EvaluationContextTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/EvaluationContextTest.java
index 6360f34..6dd2ea4 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/EvaluationContextTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/EvaluationContextTest.java
@@ -31,6 +31,7 @@ import java.util.Collection;
import java.util.Collections;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
+import org.apache.beam.runners.core.StateNamespaces;
import org.apache.beam.runners.core.TimerInternals.TimerData;
import org.apache.beam.runners.direct.DirectExecutionContext.DirectStepContext;
import org.apache.beam.runners.direct.DirectRunner.CommittedBundle;
@@ -58,7 +59,6 @@ import org.apache.beam.sdk.util.TimeDomain;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.util.WindowingStrategy;
import org.apache.beam.sdk.util.state.BagState;
-import org.apache.beam.sdk.util.state.StateNamespaces;
import org.apache.beam.sdk.util.state.StateTag;
import org.apache.beam.sdk.util.state.StateTags;
import org.apache.beam.sdk.values.KV;
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/direct-java/src/test/java/org/apache/beam/runners/direct/StatefulParDoEvaluatorFactoryTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/StatefulParDoEvaluatorFactoryTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/StatefulParDoEvaluatorFactoryTest.java
index 62fc102..cc5ee74 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/StatefulParDoEvaluatorFactoryTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/StatefulParDoEvaluatorFactoryTest.java
@@ -36,6 +36,8 @@ import java.util.List;
import org.apache.beam.runners.core.KeyedWorkItem;
import org.apache.beam.runners.core.KeyedWorkItems;
import org.apache.beam.runners.core.StateInternals;
+import org.apache.beam.runners.core.StateNamespace;
+import org.apache.beam.runners.core.StateNamespaces;
import org.apache.beam.runners.direct.DirectRunner.CommittedBundle;
import org.apache.beam.runners.direct.DirectRunner.UncommittedBundle;
import org.apache.beam.runners.direct.ParDoMultiOverrideFactory.StatefulParDo;
@@ -55,8 +57,6 @@ import org.apache.beam.sdk.transforms.windowing.Window;
import org.apache.beam.sdk.util.ReadyCheckingSideInputReader;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.util.WindowingStrategy;
-import org.apache.beam.sdk.util.state.StateNamespace;
-import org.apache.beam.sdk.util.state.StateNamespaces;
import org.apache.beam.sdk.util.state.StateSpec;
import org.apache.beam.sdk.util.state.StateSpecs;
import org.apache.beam.sdk.util.state.StateTag;
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/direct-java/src/test/java/org/apache/beam/runners/direct/WatermarkManagerTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/WatermarkManagerTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/WatermarkManagerTest.java
index ee1e5bf..493edbb 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/WatermarkManagerTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/WatermarkManagerTest.java
@@ -33,6 +33,7 @@ import java.util.Collections;
import java.util.EnumSet;
import java.util.Map;
import javax.annotation.Nullable;
+import org.apache.beam.runners.core.StateNamespaces;
import org.apache.beam.runners.core.TimerInternals.TimerData;
import org.apache.beam.runners.direct.CommittedResult.OutputType;
import org.apache.beam.runners.direct.DirectRunner.CommittedBundle;
@@ -58,7 +59,6 @@ import org.apache.beam.sdk.transforms.windowing.IntervalWindow;
import org.apache.beam.sdk.transforms.windowing.PaneInfo;
import org.apache.beam.sdk.util.TimeDomain;
import org.apache.beam.sdk.util.WindowedValue;
-import org.apache.beam.sdk.util.state.StateNamespaces;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionList;
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/FlinkStateInternals.java
----------------------------------------------------------------------
diff --git a/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/FlinkStateInternals.java b/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/FlinkStateInternals.java
index c992d40..5731a38 100644
--- a/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/FlinkStateInternals.java
+++ b/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/FlinkStateInternals.java
@@ -23,6 +23,7 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import org.apache.beam.runners.core.StateInternals;
+import org.apache.beam.runners.core.StateNamespace;
import org.apache.beam.runners.flink.translation.types.CoderTypeInformation;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.coders.CoderException;
@@ -39,7 +40,6 @@ import org.apache.beam.sdk.util.state.ReadableState;
import org.apache.beam.sdk.util.state.State;
import org.apache.beam.sdk.util.state.StateContext;
import org.apache.beam.sdk.util.state.StateContexts;
-import org.apache.beam.sdk.util.state.StateNamespace;
import org.apache.beam.sdk.util.state.StateTag;
import org.apache.beam.sdk.util.state.ValueState;
import org.apache.beam.sdk.util.state.WatermarkHoldState;
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/WindowDoFnOperator.java
----------------------------------------------------------------------
diff --git a/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/WindowDoFnOperator.java b/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/WindowDoFnOperator.java
index f1eee84..5ef7896 100644
--- a/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/WindowDoFnOperator.java
+++ b/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/WindowDoFnOperator.java
@@ -44,6 +44,7 @@ import org.apache.beam.runners.core.KeyedWorkItem;
import org.apache.beam.runners.core.KeyedWorkItems;
import org.apache.beam.runners.core.StateInternals;
import org.apache.beam.runners.core.StateInternalsFactory;
+import org.apache.beam.runners.core.StateNamespace;
import org.apache.beam.runners.core.SystemReduceFn;
import org.apache.beam.runners.core.TimerInternals;
import org.apache.beam.runners.core.TimerInternalsFactory;
@@ -55,7 +56,6 @@ import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
import org.apache.beam.sdk.util.TimeDomain;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.util.WindowingStrategy;
-import org.apache.beam.sdk.util.state.StateNamespace;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollectionView;
import org.apache.beam.sdk.values.TupleTag;
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/runners/flink/runner/src/test/java/org/apache/beam/runners/flink/streaming/FlinkStateInternalsTest.java
----------------------------------------------------------------------
diff --git a/runners/flink/runner/src/test/java/org/apache/beam/runners/flink/streaming/FlinkStateInternalsTest.java b/runners/flink/runner/src/test/java/org/apache/beam/runners/flink/streaming/FlinkStateInternalsTest.java
index 126f611..5784b68 100644
--- a/runners/flink/runner/src/test/java/org/apache/beam/runners/flink/streaming/FlinkStateInternalsTest.java
+++ b/runners/flink/runner/src/test/java/org/apache/beam/runners/flink/streaming/FlinkStateInternalsTest.java
@@ -24,6 +24,8 @@ import static org.junit.Assert.assertThat;
import java.nio.ByteBuffer;
import java.util.Arrays;
+import org.apache.beam.runners.core.StateNamespace;
+import org.apache.beam.runners.core.StateNamespaceForTest;
import org.apache.beam.runners.flink.translation.wrappers.streaming.FlinkStateInternals;
import org.apache.beam.sdk.coders.CoderException;
import org.apache.beam.sdk.coders.StringUtf8Coder;
@@ -38,8 +40,6 @@ import org.apache.beam.sdk.util.state.BagState;
import org.apache.beam.sdk.util.state.CombiningState;
import org.apache.beam.sdk.util.state.ReadableState;
import org.apache.beam.sdk.util.state.StateMerging;
-import org.apache.beam.sdk.util.state.StateNamespace;
-import org.apache.beam.sdk.util.state.StateNamespaceForTest;
import org.apache.beam.sdk.util.state.StateTag;
import org.apache.beam.sdk.util.state.StateTags;
import org.apache.beam.sdk.util.state.ValueState;
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/StateNamespace.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/StateNamespace.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/StateNamespace.java
deleted file mode 100644
index 04440cb..0000000
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/StateNamespace.java
+++ /dev/null
@@ -1,56 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.beam.sdk.util.state;
-
-import java.io.IOException;
-
-/**
- * A namespace used for scoping state stored with {@link StateInternals}.
- *
- * <p>Instances of {@link StateNamespace} are guaranteed to have a {@link Object#hashCode} and
- * {@link Object#equals} that uniquely identify the namespace.
- */
-public interface StateNamespace {
-
- /**
- * Return a {@link String} representation of the key. It is guaranteed that this
- * {@link String} will uniquely identify the key.
- *
- * <p>This will encode the actual namespace as a {@link String}. It is
- * preferable to use the {@link StateNamespace} object when possible.
- *
- * <p>The string produced by the standard implementations will not contain a '+' character. This
- * enables adding a '+' between the actual namespace and other information, if needed, to separate
- * the two.
- */
- String stringKey();
-
- /**
- * Append the string representation of this key to the {@link Appendable}.
- */
- void appendTo(Appendable sb) throws IOException;
-
- /**
- * Return an {@link Object} to use as a key in a cache.
- *
- * <p>Different namespaces may use the same key in order to be treated as a unit in the cache.
- * The {@link Object}'s {@link Object#hashCode} and {@link Object#equals} methods will be used to
- * determine equality.
- */
- Object getCacheKey();
-}
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/StateNamespaceForTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/StateNamespaceForTest.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/StateNamespaceForTest.java
deleted file mode 100644
index f1050ae..0000000
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/StateNamespaceForTest.java
+++ /dev/null
@@ -1,65 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.beam.sdk.util.state;
-
-import java.io.IOException;
-import java.util.Objects;
-
-/**
- * A simple {@link StateNamespace} used for testing.
- */
-public class StateNamespaceForTest implements StateNamespace {
- private String key;
-
- public StateNamespaceForTest(String key) {
- this.key = key;
- }
-
- @Override
- public String stringKey() {
- return key;
- }
-
- @Override
- public Object getCacheKey() {
- return key;
- }
-
- @Override
- public boolean equals(Object obj) {
- if (this == obj) {
- return true;
- }
-
- if (!(obj instanceof StateNamespaceForTest)) {
- return false;
- }
-
- return Objects.equals(this.key, ((StateNamespaceForTest) obj).key);
- }
-
- @Override
- public int hashCode() {
- return key.hashCode();
- }
-
- @Override
- public void appendTo(Appendable sb) throws IOException {
- sb.append(key);
- }
-}
http://git-wip-us.apache.org/repos/asf/beam/blob/e77e7f0d/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/StateNamespaces.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/StateNamespaces.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/StateNamespaces.java
deleted file mode 100644
index 75f555c..0000000
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/StateNamespaces.java
+++ /dev/null
@@ -1,278 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.beam.sdk.util.state;
-
-import com.google.common.base.Splitter;
-import java.io.IOException;
-import java.util.List;
-import java.util.Objects;
-import org.apache.beam.sdk.coders.Coder;
-import org.apache.beam.sdk.coders.CoderException;
-import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
-import org.apache.beam.sdk.util.CoderUtils;
-
-/**
- * Factory methods for creating the {@link StateNamespace StateNamespaces}.
- */
-public class StateNamespaces {
-
- private enum Namespace {
- GLOBAL,
- WINDOW,
- WINDOW_AND_TRIGGER
- }
-
- public static StateNamespace global() {
- return new GlobalNamespace();
- }
-
- public static <W extends BoundedWindow> StateNamespace window(Coder<W> windowCoder, W window) {
- return new WindowNamespace<>(windowCoder, window);
- }
-
- public static <W extends BoundedWindow>
- StateNamespace windowAndTrigger(Coder<W> windowCoder, W window, int triggerIdx) {
- return new WindowAndTriggerNamespace<>(windowCoder, window, triggerIdx);
- }
-
- private StateNamespaces() {}
-
- /**
- * {@link StateNamespace} that is global to the current key being processed.
- */
- public static class GlobalNamespace implements StateNamespace {
-
- private static final String GLOBAL_STRING = "/";
-
- @Override
- public String stringKey() {
- return GLOBAL_STRING;
- }
-
- @Override
- public Object getCacheKey() {
- return GLOBAL_STRING;
- }
-
- @Override
- public boolean equals(Object obj) {
- return obj == this || obj instanceof GlobalNamespace;
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(Namespace.GLOBAL);
- }
-
- @Override
- public String toString() {
- return "Global";
- }
-
- @Override
- public void appendTo(Appendable sb) throws IOException {
- sb.append(GLOBAL_STRING);
- }
- }
-
- /**
- * {@link StateNamespace} that is scoped to a specific window.
- */
- public static class WindowNamespace<W extends BoundedWindow> implements StateNamespace {
-
- private static final String WINDOW_FORMAT = "/%s/";
-
- private Coder<W> windowCoder;
- private W window;
-
- private WindowNamespace(Coder<W> windowCoder, W window) {
- this.windowCoder = windowCoder;
- this.window = window;
- }
-
- public W getWindow() {
- return window;
- }
-
- @Override
- public String stringKey() {
- try {
- return String.format(WINDOW_FORMAT, CoderUtils.encodeToBase64(windowCoder, window));
- } catch (CoderException e) {
- throw new RuntimeException("Unable to generate string key from window " + window, e);
- }
- }
-
- @Override
- public void appendTo(Appendable sb) throws IOException {
- sb.append('/').append(CoderUtils.encodeToBase64(windowCoder, window)).append('/');
- }
-
- /**
- * State in the same window will all be evicted together.
- */
- @Override
- public Object getCacheKey() {
- return window;
- }
-
- @Override
- public boolean equals(Object obj) {
- if (obj == this) {
- return true;
- }
-
- if (!(obj instanceof WindowNamespace)) {
- return false;
- }
-
- WindowNamespace<?> that = (WindowNamespace<?>) obj;
- return Objects.equals(this.window, that.window);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(Namespace.WINDOW, window);
- }
-
- @Override
- public String toString() {
- return "Window(" + window + ")";
- }
- }
-
- /**
- * {@link StateNamespace} that is scoped to a particular window and trigger index.
- */
- public static class WindowAndTriggerNamespace<W extends BoundedWindow>
- implements StateNamespace {
-
- private static final String WINDOW_AND_TRIGGER_FORMAT = "/%s/%s/";
-
- private static final int TRIGGER_RADIX = 36;
- private Coder<W> windowCoder;
- private W window;
- private int triggerIndex;
-
- private WindowAndTriggerNamespace(Coder<W> windowCoder, W window, int triggerIndex) {
- this.windowCoder = windowCoder;
- this.window = window;
- this.triggerIndex = triggerIndex;
- }
-
- public W getWindow() {
- return window;
- }
-
- public int getTriggerIndex() {
- return triggerIndex;
- }
-
- @Override
- public String stringKey() {
- try {
- return String.format(WINDOW_AND_TRIGGER_FORMAT,
- CoderUtils.encodeToBase64(windowCoder, window),
- // Use base 36 so that can address 36 triggers in a single byte and still be human
- // readable.
- Integer.toString(triggerIndex, TRIGGER_RADIX).toUpperCase());
- } catch (CoderException e) {
- throw new RuntimeException("Unable to generate string key from window " + window, e);
- }
- }
-
- @Override
- public void appendTo(Appendable sb) throws IOException {
- sb.append('/').append(CoderUtils.encodeToBase64(windowCoder, window));
- sb.append('/').append(Integer.toString(triggerIndex, TRIGGER_RADIX).toUpperCase());
- sb.append('/');
- }
-
- /**
- * State in the same window will all be evicted together.
- */
- @Override
- public Object getCacheKey() {
- return window;
- }
-
- @Override
- public boolean equals(Object obj) {
- if (obj == this) {
- return true;
- }
-
- if (!(obj instanceof WindowAndTriggerNamespace)) {
- return false;
- }
-
- WindowAndTriggerNamespace<?> that = (WindowAndTriggerNamespace<?>) obj;
- return this.triggerIndex == that.triggerIndex
- && Objects.equals(this.window, that.window);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(Namespace.WINDOW_AND_TRIGGER, window, triggerIndex);
- }
-
- @Override
- public String toString() {
- return "WindowAndTrigger(" + window + "," + triggerIndex + ")";
- }
- }
-
- private static final Splitter SLASH_SPLITTER = Splitter.on('/');
-
- /**
- * Convert a {@code stringKey} produced using {@link StateNamespace#stringKey}
- * on one of the namespaces produced by this class into the original
- * {@link StateNamespace}.
- */
- public static <W extends BoundedWindow> StateNamespace fromString(
- String stringKey, Coder<W> windowCoder) {
- if (!stringKey.startsWith("/") || !stringKey.endsWith("/")) {
- throw new RuntimeException("Invalid namespace string: '" + stringKey + "'");
- }
-
- if (GlobalNamespace.GLOBAL_STRING.equals(stringKey)) {
- return global();
- }
-
- List<String> parts = SLASH_SPLITTER.splitToList(stringKey);
- if (parts.size() != 3 && parts.size() != 4) {
- throw new RuntimeException("Invalid namespace string: '" + stringKey + "'");
- }
- // Ends should be empty (we start and end with /)
- if (!parts.get(0).isEmpty() || !parts.get(parts.size() - 1).isEmpty()) {
- throw new RuntimeException("Invalid namespace string: '" + stringKey + "'");
- }
-
- try {
- W window = CoderUtils.decodeFromBase64(windowCoder, parts.get(1));
- if (parts.size() > 3) {
- int index = Integer.parseInt(parts.get(2), WindowAndTriggerNamespace.TRIGGER_RADIX);
- return windowAndTrigger(windowCoder, window, index);
- } else {
- return window(windowCoder, window);
- }
- } catch (Exception e) {
- throw new RuntimeException("Invalid namespace string: '" + stringKey + "'", e);
- }
- }
-}