From 2122e849308016b0ec0c5ed50c337edb4c24d1d1 Mon Sep 17 00:00:00 2001 From: Kenneth Knowles Date: Tue, 2 May 2017 10:29:33 -0700 Subject: [PATCH 1/2] Move Java sdk.util.state to sdk.state --- .../translation/utils/ApexStateInternals.java | 20 ++++++++--------- .../utils/StateInternalsProxy.java | 4 ++-- .../utils/ApexStateInternalsTest.java | 12 +++++----- .../construction/PTransformMatchersTest.java | 6 ++--- .../beam/runners/core/DoFnAdapters.java | 2 +- .../runners/core/InMemoryStateInternals.java | 22 +++++++++---------- .../runners/core/MergingActiveWindowSet.java | 2 +- .../runners/core/MergingStateAccessor.java | 2 +- .../beam/runners/core/NonEmptyPanes.java | 4 ++-- ...oundedSplittableProcessElementInvoker.java | 2 +- .../beam/runners/core/PaneInfoTracker.java | 4 ++-- .../apache/beam/runners/core/ReduceFn.java | 2 +- .../runners/core/ReduceFnContextFactory.java | 6 ++--- .../beam/runners/core/SideInputHandler.java | 4 ++-- .../beam/runners/core/SimpleDoFnRunner.java | 4 ++-- .../beam/runners/core/SplittableParDo.java | 4 ++-- .../beam/runners/core/StateAccessor.java | 2 +- .../beam/runners/core/StateInternals.java | 4 ++-- .../beam/runners/core/StateMerging.java | 14 ++++++------ .../apache/beam/runners/core/StateTable.java | 4 ++-- .../apache/beam/runners/core/StateTag.java | 16 +++++++------- .../apache/beam/runners/core/StateTags.java | 20 ++++++++--------- .../beam/runners/core/StatefulDoFnRunner.java | 4 ++-- .../beam/runners/core/SystemReduceFn.java | 8 +++---- .../core/TestInMemoryStateInternals.java | 4 ++-- .../beam/runners/core/WatermarkHold.java | 4 ++-- ...fterDelayFromFirstElementStateMachine.java | 4 ++-- .../core/triggers/AfterPaneStateMachine.java | 2 +- .../TriggerStateMachineContextFactory.java | 2 +- .../triggers/TriggerStateMachineRunner.java | 2 +- .../core/InMemoryStateInternalsTest.java | 16 +++++++------- .../runners/core/StatefulDoFnRunnerTest.java | 6 ++--- .../CopyOnAccessInMemoryStateInternals.java | 18 +++++++-------- .../direct/StatefulParDoEvaluatorFactory.java | 2 +- ...opyOnAccessInMemoryStateInternalsTest.java | 14 ++++++------ .../runners/direct/EvaluationContextTest.java | 2 +- .../StatefulParDoEvaluatorFactoryTest.java | 6 ++--- .../wrappers/streaming/DoFnOperator.java | 2 +- .../state/FlinkBroadcastStateInternals.java | 20 ++++++++--------- .../state/FlinkKeyGroupStateInternals.java | 20 ++++++++--------- .../state/FlinkSplitStateInternals.java | 20 ++++++++--------- .../streaming/state/FlinkStateInternals.java | 20 ++++++++--------- .../flink/streaming/DoFnOperatorTest.java | 6 ++--- .../FlinkBroadcastStateInternalsTest.java | 10 ++++----- .../FlinkKeyGroupStateInternalsTest.java | 4 ++-- .../FlinkSplitStateInternalsTest.java | 4 ++-- .../streaming/FlinkStateInternalsTest.java | 12 +++++----- .../BatchStatefulParDoOverridesTest.java | 6 ++--- .../DataflowPipelineTranslatorTest.java | 6 ++--- .../spark/stateful/SparkStateInternals.java | 20 ++++++++--------- .../main/resources/beam/findbugs-filter.xml | 2 +- .../beam/sdk/{util => }/state/BagState.java | 2 +- .../sdk/{util => }/state/CombiningState.java | 2 +- .../sdk/{util => }/state/GroupingState.java | 2 +- .../beam/sdk/{util => }/state/MapState.java | 2 +- .../sdk/{util => }/state/ReadableState.java | 2 +- .../sdk/{util => }/state/ReadableStates.java | 2 +- .../beam/sdk/{util => }/state/SetState.java | 2 +- .../beam/sdk/{util => }/state/State.java | 2 +- .../sdk/{util => }/state/StateBinder.java | 2 +- .../sdk/{util => }/state/StateContext.java | 2 +- .../sdk/{util => }/state/StateContexts.java | 2 +- .../beam/sdk/{util => }/state/StateSpec.java | 2 +- .../beam/sdk/{util => }/state/StateSpecs.java | 2 +- .../beam/sdk/{util => }/state/ValueState.java | 2 +- .../{util => }/state/WatermarkHoldState.java | 2 +- .../sdk/{util => }/state/package-info.java | 2 +- .../apache/beam/sdk/testing/UsesMapState.java | 2 +- .../apache/beam/sdk/testing/UsesSetState.java | 2 +- .../org/apache/beam/sdk/transforms/DoFn.java | 4 ++-- .../beam/sdk/transforms/DoFnTester.java | 2 +- .../beam/sdk/transforms/GroupIntoBatches.java | 10 ++++----- .../org/apache/beam/sdk/transforms/ParDo.java | 2 +- .../sdk/transforms/reflect/DoFnInvoker.java | 2 +- .../sdk/transforms/reflect/DoFnSignature.java | 4 ++-- .../transforms/reflect/DoFnSignatures.java | 4 ++-- .../beam/sdk/util/CombineContextFactory.java | 2 +- .../apache/beam/sdk/util/CombineFnUtil.java | 2 +- .../apache/beam/sdk/transforms/ParDoTest.java | 14 ++++++------ .../transforms/reflect/DoFnInvokersTest.java | 6 ++--- .../reflect/DoFnSignaturesTest.java | 8 +++---- .../beam/sdk/util/CombineFnUtilTest.java | 2 +- 82 files changed, 252 insertions(+), 252 deletions(-) rename sdks/java/core/src/main/java/org/apache/beam/sdk/{util => }/state/BagState.java (96%) rename sdks/java/core/src/main/java/org/apache/beam/sdk/{util => }/state/CombiningState.java (98%) rename sdks/java/core/src/main/java/org/apache/beam/sdk/{util => }/state/GroupingState.java (97%) rename sdks/java/core/src/main/java/org/apache/beam/sdk/{util => }/state/MapState.java (98%) rename sdks/java/core/src/main/java/org/apache/beam/sdk/{util => }/state/ReadableState.java (97%) rename sdks/java/core/src/main/java/org/apache/beam/sdk/{util => }/state/ReadableStates.java (97%) rename sdks/java/core/src/main/java/org/apache/beam/sdk/{util => }/state/SetState.java (97%) rename sdks/java/core/src/main/java/org/apache/beam/sdk/{util => }/state/State.java (96%) rename sdks/java/core/src/main/java/org/apache/beam/sdk/{util => }/state/StateBinder.java (98%) rename sdks/java/core/src/main/java/org/apache/beam/sdk/{util => }/state/StateContext.java (97%) rename sdks/java/core/src/main/java/org/apache/beam/sdk/{util => }/state/StateContexts.java (97%) rename sdks/java/core/src/main/java/org/apache/beam/sdk/{util => }/state/StateSpec.java (98%) rename sdks/java/core/src/main/java/org/apache/beam/sdk/{util => }/state/StateSpecs.java (99%) rename sdks/java/core/src/main/java/org/apache/beam/sdk/{util => }/state/ValueState.java (96%) rename sdks/java/core/src/main/java/org/apache/beam/sdk/{util => }/state/WatermarkHoldState.java (97%) rename sdks/java/core/src/main/java/org/apache/beam/sdk/{util => }/state/package-info.java (95%) 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 430056718696..b581f7c5fbb1 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 @@ -47,16 +47,16 @@ import org.apache.beam.sdk.transforms.windowing.TimestampCombiner; import org.apache.beam.sdk.util.CoderUtils; import org.apache.beam.sdk.util.CombineFnUtil; -import org.apache.beam.sdk.util.state.BagState; -import org.apache.beam.sdk.util.state.CombiningState; -import org.apache.beam.sdk.util.state.MapState; -import org.apache.beam.sdk.util.state.ReadableState; -import org.apache.beam.sdk.util.state.SetState; -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.ValueState; -import org.apache.beam.sdk.util.state.WatermarkHoldState; +import org.apache.beam.sdk.state.BagState; +import org.apache.beam.sdk.state.CombiningState; +import org.apache.beam.sdk.state.MapState; +import org.apache.beam.sdk.state.ReadableState; +import org.apache.beam.sdk.state.SetState; +import org.apache.beam.sdk.state.State; +import org.apache.beam.sdk.state.StateContext; +import org.apache.beam.sdk.state.StateContexts; +import org.apache.beam.sdk.state.ValueState; +import org.apache.beam.sdk.state.WatermarkHoldState; import org.joda.time.Instant; /** diff --git a/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/utils/StateInternalsProxy.java b/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/utils/StateInternalsProxy.java index 746be2f8f572..ccf7e43e536c 100644 --- a/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/utils/StateInternalsProxy.java +++ b/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/utils/StateInternalsProxy.java @@ -24,8 +24,8 @@ import org.apache.beam.runners.core.StateInternalsFactory; import org.apache.beam.runners.core.StateNamespace; import org.apache.beam.runners.core.StateTag; -import org.apache.beam.sdk.util.state.State; -import org.apache.beam.sdk.util.state.StateContext; +import org.apache.beam.sdk.state.State; +import org.apache.beam.sdk.state.StateContext; /** * State internals for reusable processing context. 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 8b48a74148b8..8b38726b4081 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 @@ -37,12 +37,12 @@ import org.apache.beam.sdk.transforms.windowing.BoundedWindow; import org.apache.beam.sdk.transforms.windowing.IntervalWindow; import org.apache.beam.sdk.transforms.windowing.TimestampCombiner; -import org.apache.beam.sdk.util.state.BagState; -import org.apache.beam.sdk.util.state.CombiningState; -import org.apache.beam.sdk.util.state.GroupingState; -import org.apache.beam.sdk.util.state.ReadableState; -import org.apache.beam.sdk.util.state.ValueState; -import org.apache.beam.sdk.util.state.WatermarkHoldState; +import org.apache.beam.sdk.state.BagState; +import org.apache.beam.sdk.state.CombiningState; +import org.apache.beam.sdk.state.GroupingState; +import org.apache.beam.sdk.state.ReadableState; +import org.apache.beam.sdk.state.ValueState; +import org.apache.beam.sdk.state.WatermarkHoldState; import org.hamcrest.Matchers; import org.joda.time.Instant; import org.junit.Before; diff --git a/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/PTransformMatchersTest.java b/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/PTransformMatchersTest.java index 33ba80ca2ecf..a5c1e6b30274 100644 --- a/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/PTransformMatchersTest.java +++ b/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/PTransformMatchersTest.java @@ -55,9 +55,9 @@ import org.apache.beam.sdk.util.TimerSpecs; import org.apache.beam.sdk.util.WindowedValue; import org.apache.beam.sdk.util.WindowingStrategy; -import org.apache.beam.sdk.util.state.StateSpec; -import org.apache.beam.sdk.util.state.StateSpecs; -import org.apache.beam.sdk.util.state.ValueState; +import org.apache.beam.sdk.state.StateSpec; +import org.apache.beam.sdk.state.StateSpecs; +import org.apache.beam.sdk.state.ValueState; import org.apache.beam.sdk.values.KV; import org.apache.beam.sdk.values.PCollection; import org.apache.beam.sdk.values.PCollection.IsBounded; diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/DoFnAdapters.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/DoFnAdapters.java index b2069e19d010..e9c7e942c53f 100644 --- a/runners/core-java/src/main/java/org/apache/beam/runners/core/DoFnAdapters.java +++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/DoFnAdapters.java @@ -32,7 +32,7 @@ import org.apache.beam.sdk.transforms.windowing.BoundedWindow; import org.apache.beam.sdk.transforms.windowing.PaneInfo; import org.apache.beam.sdk.util.Timer; -import org.apache.beam.sdk.util.state.State; +import org.apache.beam.sdk.state.State; import org.apache.beam.sdk.values.PCollectionView; import org.apache.beam.sdk.values.TupleTag; import org.joda.time.Duration; 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 199ce41a867d..e85bfb7eda73 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 @@ -35,17 +35,17 @@ import org.apache.beam.sdk.transforms.windowing.BoundedWindow; import org.apache.beam.sdk.transforms.windowing.TimestampCombiner; import org.apache.beam.sdk.util.CombineFnUtil; -import org.apache.beam.sdk.util.state.BagState; -import org.apache.beam.sdk.util.state.CombiningState; -import org.apache.beam.sdk.util.state.MapState; -import org.apache.beam.sdk.util.state.ReadableState; -import org.apache.beam.sdk.util.state.ReadableStates; -import org.apache.beam.sdk.util.state.SetState; -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.ValueState; -import org.apache.beam.sdk.util.state.WatermarkHoldState; +import org.apache.beam.sdk.state.BagState; +import org.apache.beam.sdk.state.CombiningState; +import org.apache.beam.sdk.state.MapState; +import org.apache.beam.sdk.state.ReadableState; +import org.apache.beam.sdk.state.ReadableStates; +import org.apache.beam.sdk.state.SetState; +import org.apache.beam.sdk.state.State; +import org.apache.beam.sdk.state.StateContext; +import org.apache.beam.sdk.state.StateContexts; +import org.apache.beam.sdk.state.ValueState; +import org.apache.beam.sdk.state.WatermarkHoldState; import org.joda.time.Instant; /** 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 2faedbb0f630..ebe3588567d1 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,7 @@ 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.ValueState; +import org.apache.beam.sdk.state.ValueState; /** * An {@link ActiveWindowSet} for merging {@link WindowFn} implementations. diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/MergingStateAccessor.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/MergingStateAccessor.java index 5ffb9a22218a..8514e49b2aca 100644 --- a/runners/core-java/src/main/java/org/apache/beam/runners/core/MergingStateAccessor.java +++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/MergingStateAccessor.java @@ -21,7 +21,7 @@ import org.apache.beam.sdk.annotations.Experimental; import org.apache.beam.sdk.annotations.Experimental.Kind; import org.apache.beam.sdk.transforms.windowing.BoundedWindow; -import org.apache.beam.sdk.util.state.State; +import org.apache.beam.sdk.state.State; /** * Interface for accessing persistent state while windows are merging. diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/NonEmptyPanes.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/NonEmptyPanes.java index 06dcc9cdcfe1..887857d01b8d 100644 --- a/runners/core-java/src/main/java/org/apache/beam/runners/core/NonEmptyPanes.java +++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/NonEmptyPanes.java @@ -22,8 +22,8 @@ import org.apache.beam.sdk.transforms.windowing.BoundedWindow; import org.apache.beam.sdk.util.WindowingStrategy; import org.apache.beam.sdk.util.WindowingStrategy.AccumulationMode; -import org.apache.beam.sdk.util.state.CombiningState; -import org.apache.beam.sdk.util.state.ReadableState; +import org.apache.beam.sdk.state.CombiningState; +import org.apache.beam.sdk.state.ReadableState; /** * Tracks which windows have non-empty panes. Specifically, which windows have new elements since diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/OutputAndTimeBoundedSplittableProcessElementInvoker.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/OutputAndTimeBoundedSplittableProcessElementInvoker.java index 35d6eb7e2bd6..65e4ca6f85fc 100644 --- a/runners/core-java/src/main/java/org/apache/beam/runners/core/OutputAndTimeBoundedSplittableProcessElementInvoker.java +++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/OutputAndTimeBoundedSplittableProcessElementInvoker.java @@ -35,7 +35,7 @@ import org.apache.beam.sdk.util.SideInputReader; import org.apache.beam.sdk.util.Timer; import org.apache.beam.sdk.util.WindowedValue; -import org.apache.beam.sdk.util.state.State; +import org.apache.beam.sdk.state.State; import org.apache.beam.sdk.values.PCollectionView; import org.apache.beam.sdk.values.TupleTag; import org.joda.time.Duration; diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/PaneInfoTracker.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/PaneInfoTracker.java index 66b39608a4af..c9330b9226c7 100644 --- a/runners/core-java/src/main/java/org/apache/beam/runners/core/PaneInfoTracker.java +++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/PaneInfoTracker.java @@ -25,8 +25,8 @@ import org.apache.beam.sdk.transforms.windowing.PaneInfo.PaneInfoCoder; import org.apache.beam.sdk.transforms.windowing.PaneInfo.Timing; import org.apache.beam.sdk.util.WindowTracing; -import org.apache.beam.sdk.util.state.ReadableState; -import org.apache.beam.sdk.util.state.ValueState; +import org.apache.beam.sdk.state.ReadableState; +import org.apache.beam.sdk.state.ValueState; import org.joda.time.Instant; /** diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFn.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFn.java index cd62d859aa54..a4e456463466 100644 --- a/runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFn.java +++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFn.java @@ -22,7 +22,7 @@ import org.apache.beam.sdk.transforms.windowing.PaneInfo; import org.apache.beam.sdk.util.Timers; import org.apache.beam.sdk.util.WindowingStrategy; -import org.apache.beam.sdk.util.state.ReadableState; +import org.apache.beam.sdk.state.ReadableState; import org.joda.time.Instant; /** 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 3031ebf6d8f7..d224091b2f44 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 @@ -34,9 +34,9 @@ import org.apache.beam.sdk.util.TimeDomain; import org.apache.beam.sdk.util.Timers; import org.apache.beam.sdk.util.WindowingStrategy; -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.state.State; +import org.apache.beam.sdk.state.StateContext; +import org.apache.beam.sdk.state.StateContexts; import org.apache.beam.sdk.values.PCollectionView; import org.joda.time.Instant; 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 5c67148ea0e6..588fedd3764a 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 @@ -31,8 +31,8 @@ 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.CombiningState; -import org.apache.beam.sdk.util.state.ValueState; +import org.apache.beam.sdk.state.CombiningState; +import org.apache.beam.sdk.state.ValueState; import org.apache.beam.sdk.values.PCollectionView; /** 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 edce9a2dd7fd..560aac382e04 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 @@ -54,8 +54,8 @@ 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.StateSpec; +import org.apache.beam.sdk.state.State; +import org.apache.beam.sdk.state.StateSpec; import org.apache.beam.sdk.values.PCollectionView; import org.apache.beam.sdk.values.TupleTag; import org.joda.time.Duration; 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 94f5f3504ac9..6994cc638ff7 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 @@ -47,8 +47,8 @@ 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.ValueState; -import org.apache.beam.sdk.util.state.WatermarkHoldState; +import org.apache.beam.sdk.state.ValueState; +import org.apache.beam.sdk.state.WatermarkHoldState; import org.apache.beam.sdk.values.KV; import org.apache.beam.sdk.values.PCollection; import org.apache.beam.sdk.values.PCollectionTuple; diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/StateAccessor.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/StateAccessor.java index eda896bc683f..6c4052023586 100644 --- a/runners/core-java/src/main/java/org/apache/beam/runners/core/StateAccessor.java +++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/StateAccessor.java @@ -19,7 +19,7 @@ import org.apache.beam.sdk.annotations.Experimental; import org.apache.beam.sdk.annotations.Experimental.Kind; -import org.apache.beam.sdk.util.state.State; +import org.apache.beam.sdk.state.State; /** * Interface for accessing a {@link StateTag} in the current context. 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 c2e94120c707..b288ef1ab001 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 @@ -20,8 +20,8 @@ import org.apache.beam.sdk.annotations.Experimental; 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.state.State; +import org.apache.beam.sdk.state.StateContext; /** * {@code StateInternals} describes the functionality a runner needs to provide for the diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/StateMerging.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/StateMerging.java index f6b91032e5d7..f19185752500 100644 --- a/runners/core-java/src/main/java/org/apache/beam/runners/core/StateMerging.java +++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/StateMerging.java @@ -24,13 +24,13 @@ import java.util.List; import java.util.Map; import org.apache.beam.sdk.transforms.windowing.BoundedWindow; -import org.apache.beam.sdk.util.state.BagState; -import org.apache.beam.sdk.util.state.CombiningState; -import org.apache.beam.sdk.util.state.GroupingState; -import org.apache.beam.sdk.util.state.ReadableState; -import org.apache.beam.sdk.util.state.SetState; -import org.apache.beam.sdk.util.state.State; -import org.apache.beam.sdk.util.state.WatermarkHoldState; +import org.apache.beam.sdk.state.BagState; +import org.apache.beam.sdk.state.CombiningState; +import org.apache.beam.sdk.state.GroupingState; +import org.apache.beam.sdk.state.ReadableState; +import org.apache.beam.sdk.state.SetState; +import org.apache.beam.sdk.state.State; +import org.apache.beam.sdk.state.WatermarkHoldState; import org.joda.time.Instant; /** 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 1bf4ff511392..d996729a476a 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 @@ -22,8 +22,8 @@ import java.util.Map; import java.util.Set; import org.apache.beam.runners.core.StateTag.StateBinder; -import org.apache.beam.sdk.util.state.State; -import org.apache.beam.sdk.util.state.StateContext; +import org.apache.beam.sdk.state.State; +import org.apache.beam.sdk.state.StateContext; /** * Table mapping {@code StateNamespace} and {@code StateTag} to a {@code State} instance. diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/StateTag.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/StateTag.java index 38e9dea6f712..226c7a761086 100644 --- a/runners/core-java/src/main/java/org/apache/beam/runners/core/StateTag.java +++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/StateTag.java @@ -26,14 +26,14 @@ import org.apache.beam.sdk.transforms.CombineWithContext.CombineFnWithContext; import org.apache.beam.sdk.transforms.GroupByKey; import org.apache.beam.sdk.transforms.windowing.TimestampCombiner; -import org.apache.beam.sdk.util.state.BagState; -import org.apache.beam.sdk.util.state.CombiningState; -import org.apache.beam.sdk.util.state.MapState; -import org.apache.beam.sdk.util.state.SetState; -import org.apache.beam.sdk.util.state.State; -import org.apache.beam.sdk.util.state.StateSpec; -import org.apache.beam.sdk.util.state.ValueState; -import org.apache.beam.sdk.util.state.WatermarkHoldState; +import org.apache.beam.sdk.state.BagState; +import org.apache.beam.sdk.state.CombiningState; +import org.apache.beam.sdk.state.MapState; +import org.apache.beam.sdk.state.SetState; +import org.apache.beam.sdk.state.State; +import org.apache.beam.sdk.state.StateSpec; +import org.apache.beam.sdk.state.ValueState; +import org.apache.beam.sdk.state.WatermarkHoldState; /** * An address and specification for a persistent state cell. This includes a unique identifier for diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/StateTags.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/StateTags.java index ca8b2387f52e..136b3b4a5d0c 100644 --- a/runners/core-java/src/main/java/org/apache/beam/runners/core/StateTags.java +++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/StateTags.java @@ -29,16 +29,16 @@ import org.apache.beam.sdk.transforms.CombineWithContext.CombineFnWithContext; import org.apache.beam.sdk.transforms.windowing.BoundedWindow; import org.apache.beam.sdk.transforms.windowing.TimestampCombiner; -import org.apache.beam.sdk.util.state.BagState; -import org.apache.beam.sdk.util.state.CombiningState; -import org.apache.beam.sdk.util.state.MapState; -import org.apache.beam.sdk.util.state.SetState; -import org.apache.beam.sdk.util.state.State; -import org.apache.beam.sdk.util.state.StateBinder; -import org.apache.beam.sdk.util.state.StateSpec; -import org.apache.beam.sdk.util.state.StateSpecs; -import org.apache.beam.sdk.util.state.ValueState; -import org.apache.beam.sdk.util.state.WatermarkHoldState; +import org.apache.beam.sdk.state.BagState; +import org.apache.beam.sdk.state.CombiningState; +import org.apache.beam.sdk.state.MapState; +import org.apache.beam.sdk.state.SetState; +import org.apache.beam.sdk.state.State; +import org.apache.beam.sdk.state.StateBinder; +import org.apache.beam.sdk.state.StateSpec; +import org.apache.beam.sdk.state.StateSpecs; +import org.apache.beam.sdk.state.ValueState; +import org.apache.beam.sdk.state.WatermarkHoldState; /** * Static utility methods for creating {@link StateTag} instances. diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/StatefulDoFnRunner.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/StatefulDoFnRunner.java index e3717a89daeb..44745fe8b4a6 100644 --- a/runners/core-java/src/main/java/org/apache/beam/runners/core/StatefulDoFnRunner.java +++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/StatefulDoFnRunner.java @@ -31,8 +31,8 @@ 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.state.State; -import org.apache.beam.sdk.util.state.StateSpec; +import org.apache.beam.sdk.state.State; +import org.apache.beam.sdk.state.StateSpec; import org.joda.time.Instant; /** diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/SystemReduceFn.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/SystemReduceFn.java index f18460ae4044..3c6243eccbcc 100644 --- a/runners/core-java/src/main/java/org/apache/beam/runners/core/SystemReduceFn.java +++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/SystemReduceFn.java @@ -24,10 +24,10 @@ import org.apache.beam.sdk.transforms.GroupByKey; import org.apache.beam.sdk.transforms.windowing.BoundedWindow; import org.apache.beam.sdk.util.AppliedCombineFn; -import org.apache.beam.sdk.util.state.BagState; -import org.apache.beam.sdk.util.state.CombiningState; -import org.apache.beam.sdk.util.state.GroupingState; -import org.apache.beam.sdk.util.state.ReadableState; +import org.apache.beam.sdk.state.BagState; +import org.apache.beam.sdk.state.CombiningState; +import org.apache.beam.sdk.state.GroupingState; +import org.apache.beam.sdk.state.ReadableState; /** * {@link ReduceFn} implementing the default reduction behaviors of {@link GroupByKey}. 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 18b50dba0b39..2052c039f80a 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 @@ -20,8 +20,8 @@ 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.WatermarkHoldState; +import org.apache.beam.sdk.state.State; +import org.apache.beam.sdk.state.WatermarkHoldState; import org.joda.time.Instant; /** diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/WatermarkHold.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/WatermarkHold.java index e6e4ffb3bafe..13f9f27ee69b 100644 --- a/runners/core-java/src/main/java/org/apache/beam/runners/core/WatermarkHold.java +++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/WatermarkHold.java @@ -28,8 +28,8 @@ import org.apache.beam.sdk.transforms.windowing.Window.ClosingBehavior; import org.apache.beam.sdk.util.WindowTracing; import org.apache.beam.sdk.util.WindowingStrategy; -import org.apache.beam.sdk.util.state.ReadableState; -import org.apache.beam.sdk.util.state.WatermarkHoldState; +import org.apache.beam.sdk.state.ReadableState; +import org.apache.beam.sdk.state.WatermarkHoldState; import org.joda.time.Duration; import org.joda.time.Instant; diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterDelayFromFirstElementStateMachine.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterDelayFromFirstElementStateMachine.java index ed2c26f68a54..e7f8756d722e 100644 --- a/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterDelayFromFirstElementStateMachine.java +++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterDelayFromFirstElementStateMachine.java @@ -34,8 +34,8 @@ import org.apache.beam.sdk.transforms.Min; import org.apache.beam.sdk.transforms.SerializableFunction; import org.apache.beam.sdk.util.TimeDomain; -import org.apache.beam.sdk.util.state.CombiningState; -import org.apache.beam.sdk.util.state.GroupingState; +import org.apache.beam.sdk.state.CombiningState; +import org.apache.beam.sdk.state.GroupingState; import org.joda.time.Duration; import org.joda.time.Instant; import org.joda.time.format.PeriodFormat; diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterPaneStateMachine.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterPaneStateMachine.java index 52fb5ff495e3..6090bd40c2e4 100644 --- a/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterPaneStateMachine.java +++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterPaneStateMachine.java @@ -27,7 +27,7 @@ import org.apache.beam.sdk.annotations.Experimental; import org.apache.beam.sdk.coders.VarLongCoder; import org.apache.beam.sdk.transforms.Sum; -import org.apache.beam.sdk.util.state.CombiningState; +import org.apache.beam.sdk.state.CombiningState; /** * {@link TriggerStateMachine}s that fire based on properties of the elements in the current pane. 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 a0569374b396..d63ded2a38aa 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 @@ -38,7 +38,7 @@ import org.apache.beam.sdk.transforms.windowing.WindowFn; import org.apache.beam.sdk.util.TimeDomain; import org.apache.beam.sdk.util.Timers; -import org.apache.beam.sdk.util.state.State; +import org.apache.beam.sdk.state.State; import org.joda.time.Instant; /** diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/TriggerStateMachineRunner.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/TriggerStateMachineRunner.java index fc2f69681480..1c678aae4043 100644 --- a/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/TriggerStateMachineRunner.java +++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/TriggerStateMachineRunner.java @@ -31,7 +31,7 @@ import org.apache.beam.sdk.coders.BitSetCoder; import org.apache.beam.sdk.transforms.windowing.BoundedWindow; import org.apache.beam.sdk.util.Timers; -import org.apache.beam.sdk.util.state.ValueState; +import org.apache.beam.sdk.state.ValueState; import org.joda.time.Instant; /** 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 16f7f26dc965..d18511fac379 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 @@ -36,14 +36,14 @@ import org.apache.beam.sdk.transforms.windowing.BoundedWindow; import org.apache.beam.sdk.transforms.windowing.IntervalWindow; import org.apache.beam.sdk.transforms.windowing.TimestampCombiner; -import org.apache.beam.sdk.util.state.BagState; -import org.apache.beam.sdk.util.state.CombiningState; -import org.apache.beam.sdk.util.state.GroupingState; -import org.apache.beam.sdk.util.state.MapState; -import org.apache.beam.sdk.util.state.ReadableState; -import org.apache.beam.sdk.util.state.SetState; -import org.apache.beam.sdk.util.state.ValueState; -import org.apache.beam.sdk.util.state.WatermarkHoldState; +import org.apache.beam.sdk.state.BagState; +import org.apache.beam.sdk.state.CombiningState; +import org.apache.beam.sdk.state.GroupingState; +import org.apache.beam.sdk.state.MapState; +import org.apache.beam.sdk.state.ReadableState; +import org.apache.beam.sdk.state.SetState; +import org.apache.beam.sdk.state.ValueState; +import org.apache.beam.sdk.state.WatermarkHoldState; import org.hamcrest.Matchers; import org.joda.time.Instant; import org.junit.Test; diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/StatefulDoFnRunnerTest.java b/runners/core-java/src/test/java/org/apache/beam/runners/core/StatefulDoFnRunnerTest.java index f80643aea158..354ac8c5e4f5 100644 --- a/runners/core-java/src/test/java/org/apache/beam/runners/core/StatefulDoFnRunnerTest.java +++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/StatefulDoFnRunnerTest.java @@ -38,9 +38,9 @@ import org.apache.beam.sdk.util.NullSideInputReader; import org.apache.beam.sdk.util.WindowedValue; import org.apache.beam.sdk.util.WindowingStrategy; -import org.apache.beam.sdk.util.state.StateSpec; -import org.apache.beam.sdk.util.state.StateSpecs; -import org.apache.beam.sdk.util.state.ValueState; +import org.apache.beam.sdk.state.StateSpec; +import org.apache.beam.sdk.state.StateSpecs; +import org.apache.beam.sdk.state.ValueState; import org.apache.beam.sdk.values.KV; import org.apache.beam.sdk.values.TupleTag; import org.joda.time.Duration; 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 ef3a05339760..9bb2f22ec97b 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 @@ -44,15 +44,15 @@ import org.apache.beam.sdk.transforms.windowing.BoundedWindow; import org.apache.beam.sdk.transforms.windowing.TimestampCombiner; import org.apache.beam.sdk.util.CombineFnUtil; -import org.apache.beam.sdk.util.state.BagState; -import org.apache.beam.sdk.util.state.CombiningState; -import org.apache.beam.sdk.util.state.MapState; -import org.apache.beam.sdk.util.state.SetState; -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.ValueState; -import org.apache.beam.sdk.util.state.WatermarkHoldState; +import org.apache.beam.sdk.state.BagState; +import org.apache.beam.sdk.state.CombiningState; +import org.apache.beam.sdk.state.MapState; +import org.apache.beam.sdk.state.SetState; +import org.apache.beam.sdk.state.State; +import org.apache.beam.sdk.state.StateContext; +import org.apache.beam.sdk.state.StateContexts; +import org.apache.beam.sdk.state.ValueState; +import org.apache.beam.sdk.state.WatermarkHoldState; import org.joda.time.Instant; /** 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 93ab07797092..5c9404d192ea 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 @@ -49,7 +49,7 @@ 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.StateSpec; +import org.apache.beam.sdk.state.StateSpec; import org.apache.beam.sdk.values.KV; import org.apache.beam.sdk.values.PCollection; import org.apache.beam.sdk.values.PCollectionTuple; 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 3e29a69bdf82..e679a1059dae 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 @@ -43,13 +43,13 @@ import org.apache.beam.sdk.transforms.Sum; import org.apache.beam.sdk.transforms.windowing.BoundedWindow; import org.apache.beam.sdk.transforms.windowing.TimestampCombiner; -import org.apache.beam.sdk.util.state.BagState; -import org.apache.beam.sdk.util.state.CombiningState; -import org.apache.beam.sdk.util.state.GroupingState; -import org.apache.beam.sdk.util.state.MapState; -import org.apache.beam.sdk.util.state.SetState; -import org.apache.beam.sdk.util.state.ValueState; -import org.apache.beam.sdk.util.state.WatermarkHoldState; +import org.apache.beam.sdk.state.BagState; +import org.apache.beam.sdk.state.CombiningState; +import org.apache.beam.sdk.state.GroupingState; +import org.apache.beam.sdk.state.MapState; +import org.apache.beam.sdk.state.SetState; +import org.apache.beam.sdk.state.ValueState; +import org.apache.beam.sdk.state.WatermarkHoldState; import org.joda.time.Instant; import org.junit.Rule; import org.junit.Test; 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 bfbcd7953101..ac1120501767 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 @@ -59,7 +59,7 @@ 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.state.BagState; import org.apache.beam.sdk.values.KV; import org.apache.beam.sdk.values.PCollection; import org.apache.beam.sdk.values.PCollection.IsBounded; 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 fc634066239a..0cd0b1286e9d 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 @@ -60,9 +60,9 @@ 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.StateSpec; -import org.apache.beam.sdk.util.state.StateSpecs; -import org.apache.beam.sdk.util.state.ValueState; +import org.apache.beam.sdk.state.StateSpec; +import org.apache.beam.sdk.state.StateSpecs; +import org.apache.beam.sdk.state.ValueState; import org.apache.beam.sdk.values.KV; import org.apache.beam.sdk.values.PCollection; import org.apache.beam.sdk.values.PCollectionTuple; diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.java index d8fd79a2be62..1390567c12ce 100644 --- a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.java +++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.java @@ -67,7 +67,7 @@ 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.state.BagState; import org.apache.beam.sdk.values.PCollectionView; import org.apache.beam.sdk.values.TupleTag; import org.apache.flink.core.memory.DataInputViewStreamWrapper; diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/state/FlinkBroadcastStateInternals.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/state/FlinkBroadcastStateInternals.java index cfe3f9bf66eb..d9e46fef05dc 100644 --- a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/state/FlinkBroadcastStateInternals.java +++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/state/FlinkBroadcastStateInternals.java @@ -36,16 +36,16 @@ import org.apache.beam.sdk.transforms.CombineWithContext; import org.apache.beam.sdk.transforms.windowing.TimestampCombiner; import org.apache.beam.sdk.util.CombineContextFactory; -import org.apache.beam.sdk.util.state.BagState; -import org.apache.beam.sdk.util.state.CombiningState; -import org.apache.beam.sdk.util.state.MapState; -import org.apache.beam.sdk.util.state.ReadableState; -import org.apache.beam.sdk.util.state.SetState; -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.ValueState; -import org.apache.beam.sdk.util.state.WatermarkHoldState; +import org.apache.beam.sdk.state.BagState; +import org.apache.beam.sdk.state.CombiningState; +import org.apache.beam.sdk.state.MapState; +import org.apache.beam.sdk.state.ReadableState; +import org.apache.beam.sdk.state.SetState; +import org.apache.beam.sdk.state.State; +import org.apache.beam.sdk.state.StateContext; +import org.apache.beam.sdk.state.StateContexts; +import org.apache.beam.sdk.state.ValueState; +import org.apache.beam.sdk.state.WatermarkHoldState; import org.apache.flink.api.common.ExecutionConfig; import org.apache.flink.api.common.state.ListState; import org.apache.flink.api.common.state.ListStateDescriptor; diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/state/FlinkKeyGroupStateInternals.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/state/FlinkKeyGroupStateInternals.java index c9b7797ef1a8..cbd0861be93a 100644 --- a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/state/FlinkKeyGroupStateInternals.java +++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/state/FlinkKeyGroupStateInternals.java @@ -39,16 +39,16 @@ import org.apache.beam.sdk.transforms.CombineWithContext; import org.apache.beam.sdk.transforms.windowing.TimestampCombiner; import org.apache.beam.sdk.util.CoderUtils; -import org.apache.beam.sdk.util.state.BagState; -import org.apache.beam.sdk.util.state.CombiningState; -import org.apache.beam.sdk.util.state.MapState; -import org.apache.beam.sdk.util.state.ReadableState; -import org.apache.beam.sdk.util.state.SetState; -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.ValueState; -import org.apache.beam.sdk.util.state.WatermarkHoldState; +import org.apache.beam.sdk.state.BagState; +import org.apache.beam.sdk.state.CombiningState; +import org.apache.beam.sdk.state.MapState; +import org.apache.beam.sdk.state.ReadableState; +import org.apache.beam.sdk.state.SetState; +import org.apache.beam.sdk.state.State; +import org.apache.beam.sdk.state.StateContext; +import org.apache.beam.sdk.state.StateContexts; +import org.apache.beam.sdk.state.ValueState; +import org.apache.beam.sdk.state.WatermarkHoldState; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.runtime.state.KeyGroupsList; import org.apache.flink.runtime.state.KeyedStateBackend; diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/state/FlinkSplitStateInternals.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/state/FlinkSplitStateInternals.java index 3d38f88c193a..e2461ed719fd 100644 --- a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/state/FlinkSplitStateInternals.java +++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/state/FlinkSplitStateInternals.java @@ -27,16 +27,16 @@ import org.apache.beam.sdk.transforms.Combine; import org.apache.beam.sdk.transforms.CombineWithContext; import org.apache.beam.sdk.transforms.windowing.TimestampCombiner; -import org.apache.beam.sdk.util.state.BagState; -import org.apache.beam.sdk.util.state.CombiningState; -import org.apache.beam.sdk.util.state.MapState; -import org.apache.beam.sdk.util.state.ReadableState; -import org.apache.beam.sdk.util.state.SetState; -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.ValueState; -import org.apache.beam.sdk.util.state.WatermarkHoldState; +import org.apache.beam.sdk.state.BagState; +import org.apache.beam.sdk.state.CombiningState; +import org.apache.beam.sdk.state.MapState; +import org.apache.beam.sdk.state.ReadableState; +import org.apache.beam.sdk.state.SetState; +import org.apache.beam.sdk.state.State; +import org.apache.beam.sdk.state.StateContext; +import org.apache.beam.sdk.state.StateContexts; +import org.apache.beam.sdk.state.ValueState; +import org.apache.beam.sdk.state.WatermarkHoldState; import org.apache.flink.api.common.ExecutionConfig; import org.apache.flink.api.common.state.ListStateDescriptor; import org.apache.flink.runtime.state.OperatorStateBackend; diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/state/FlinkStateInternals.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/state/FlinkStateInternals.java index c033be672d9a..fd1642e4786e 100644 --- a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/state/FlinkStateInternals.java +++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/state/FlinkStateInternals.java @@ -35,16 +35,16 @@ import org.apache.beam.sdk.transforms.windowing.TimestampCombiner; import org.apache.beam.sdk.util.CoderUtils; import org.apache.beam.sdk.util.CombineContextFactory; -import org.apache.beam.sdk.util.state.BagState; -import org.apache.beam.sdk.util.state.CombiningState; -import org.apache.beam.sdk.util.state.MapState; -import org.apache.beam.sdk.util.state.ReadableState; -import org.apache.beam.sdk.util.state.SetState; -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.ValueState; -import org.apache.beam.sdk.util.state.WatermarkHoldState; +import org.apache.beam.sdk.state.BagState; +import org.apache.beam.sdk.state.CombiningState; +import org.apache.beam.sdk.state.MapState; +import org.apache.beam.sdk.state.ReadableState; +import org.apache.beam.sdk.state.SetState; +import org.apache.beam.sdk.state.State; +import org.apache.beam.sdk.state.StateContext; +import org.apache.beam.sdk.state.StateContexts; +import org.apache.beam.sdk.state.ValueState; +import org.apache.beam.sdk.state.WatermarkHoldState; import org.apache.flink.api.common.state.ListStateDescriptor; import org.apache.flink.api.common.state.ValueStateDescriptor; import org.apache.flink.api.common.typeutils.base.StringSerializer; diff --git a/runners/flink/src/test/java/org/apache/beam/runners/flink/streaming/DoFnOperatorTest.java b/runners/flink/src/test/java/org/apache/beam/runners/flink/streaming/DoFnOperatorTest.java index bda30e4b8be7..e7775078b536 100644 --- a/runners/flink/src/test/java/org/apache/beam/runners/flink/streaming/DoFnOperatorTest.java +++ b/runners/flink/src/test/java/org/apache/beam/runners/flink/streaming/DoFnOperatorTest.java @@ -52,9 +52,9 @@ import org.apache.beam.sdk.util.TimerSpecs; import org.apache.beam.sdk.util.WindowedValue; import org.apache.beam.sdk.util.WindowingStrategy; -import org.apache.beam.sdk.util.state.StateSpec; -import org.apache.beam.sdk.util.state.StateSpecs; -import org.apache.beam.sdk.util.state.ValueState; +import org.apache.beam.sdk.state.StateSpec; +import org.apache.beam.sdk.state.StateSpecs; +import org.apache.beam.sdk.state.ValueState; import org.apache.beam.sdk.values.KV; import org.apache.beam.sdk.values.PCollectionView; import org.apache.beam.sdk.values.TupleTag; diff --git a/runners/flink/src/test/java/org/apache/beam/runners/flink/streaming/FlinkBroadcastStateInternalsTest.java b/runners/flink/src/test/java/org/apache/beam/runners/flink/streaming/FlinkBroadcastStateInternalsTest.java index eb2c05f68703..f07a891f54ee 100644 --- a/runners/flink/src/test/java/org/apache/beam/runners/flink/streaming/FlinkBroadcastStateInternalsTest.java +++ b/runners/flink/src/test/java/org/apache/beam/runners/flink/streaming/FlinkBroadcastStateInternalsTest.java @@ -32,11 +32,11 @@ import org.apache.beam.sdk.coders.StringUtf8Coder; import org.apache.beam.sdk.coders.VarIntCoder; import org.apache.beam.sdk.transforms.Sum; -import org.apache.beam.sdk.util.state.BagState; -import org.apache.beam.sdk.util.state.CombiningState; -import org.apache.beam.sdk.util.state.GroupingState; -import org.apache.beam.sdk.util.state.ReadableState; -import org.apache.beam.sdk.util.state.ValueState; +import org.apache.beam.sdk.state.BagState; +import org.apache.beam.sdk.state.CombiningState; +import org.apache.beam.sdk.state.GroupingState; +import org.apache.beam.sdk.state.ReadableState; +import org.apache.beam.sdk.state.ValueState; import org.apache.flink.runtime.operators.testutils.DummyEnvironment; import org.apache.flink.runtime.state.OperatorStateBackend; import org.apache.flink.runtime.state.memory.MemoryStateBackend; diff --git a/runners/flink/src/test/java/org/apache/beam/runners/flink/streaming/FlinkKeyGroupStateInternalsTest.java b/runners/flink/src/test/java/org/apache/beam/runners/flink/streaming/FlinkKeyGroupStateInternalsTest.java index 0e0267bf9cac..9df20ede68f8 100644 --- a/runners/flink/src/test/java/org/apache/beam/runners/flink/streaming/FlinkKeyGroupStateInternalsTest.java +++ b/runners/flink/src/test/java/org/apache/beam/runners/flink/streaming/FlinkKeyGroupStateInternalsTest.java @@ -35,8 +35,8 @@ import org.apache.beam.runners.flink.translation.wrappers.streaming.state.FlinkKeyGroupStateInternals; import org.apache.beam.sdk.coders.StringUtf8Coder; import org.apache.beam.sdk.util.CoderUtils; -import org.apache.beam.sdk.util.state.BagState; -import org.apache.beam.sdk.util.state.ReadableState; +import org.apache.beam.sdk.state.BagState; +import org.apache.beam.sdk.state.ReadableState; import org.apache.flink.api.common.ExecutionConfig; import org.apache.flink.api.common.JobID; import org.apache.flink.api.java.typeutils.GenericTypeInfo; diff --git a/runners/flink/src/test/java/org/apache/beam/runners/flink/streaming/FlinkSplitStateInternalsTest.java b/runners/flink/src/test/java/org/apache/beam/runners/flink/streaming/FlinkSplitStateInternalsTest.java index 8033a9daf47e..17cd3f5d2c84 100644 --- a/runners/flink/src/test/java/org/apache/beam/runners/flink/streaming/FlinkSplitStateInternalsTest.java +++ b/runners/flink/src/test/java/org/apache/beam/runners/flink/streaming/FlinkSplitStateInternalsTest.java @@ -27,8 +27,8 @@ import org.apache.beam.runners.core.StateTags; import org.apache.beam.runners.flink.translation.wrappers.streaming.state.FlinkSplitStateInternals; import org.apache.beam.sdk.coders.StringUtf8Coder; -import org.apache.beam.sdk.util.state.BagState; -import org.apache.beam.sdk.util.state.ReadableState; +import org.apache.beam.sdk.state.BagState; +import org.apache.beam.sdk.state.ReadableState; import org.apache.flink.runtime.operators.testutils.DummyEnvironment; import org.apache.flink.runtime.state.OperatorStateBackend; import org.apache.flink.runtime.state.memory.MemoryStateBackend; diff --git a/runners/flink/src/test/java/org/apache/beam/runners/flink/streaming/FlinkStateInternalsTest.java b/runners/flink/src/test/java/org/apache/beam/runners/flink/streaming/FlinkStateInternalsTest.java index cd00d9edf234..e7f2f4073baa 100644 --- a/runners/flink/src/test/java/org/apache/beam/runners/flink/streaming/FlinkStateInternalsTest.java +++ b/runners/flink/src/test/java/org/apache/beam/runners/flink/streaming/FlinkStateInternalsTest.java @@ -37,12 +37,12 @@ import org.apache.beam.sdk.transforms.windowing.IntervalWindow; import org.apache.beam.sdk.transforms.windowing.TimestampCombiner; import org.apache.beam.sdk.util.CoderUtils; -import org.apache.beam.sdk.util.state.BagState; -import org.apache.beam.sdk.util.state.CombiningState; -import org.apache.beam.sdk.util.state.GroupingState; -import org.apache.beam.sdk.util.state.ReadableState; -import org.apache.beam.sdk.util.state.ValueState; -import org.apache.beam.sdk.util.state.WatermarkHoldState; +import org.apache.beam.sdk.state.BagState; +import org.apache.beam.sdk.state.CombiningState; +import org.apache.beam.sdk.state.GroupingState; +import org.apache.beam.sdk.state.ReadableState; +import org.apache.beam.sdk.state.ValueState; +import org.apache.beam.sdk.state.WatermarkHoldState; import org.apache.flink.api.common.ExecutionConfig; import org.apache.flink.api.common.JobID; import org.apache.flink.api.java.typeutils.GenericTypeInfo; diff --git a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/BatchStatefulParDoOverridesTest.java b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/BatchStatefulParDoOverridesTest.java index 38129ab08216..102c0228de61 100644 --- a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/BatchStatefulParDoOverridesTest.java +++ b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/BatchStatefulParDoOverridesTest.java @@ -46,9 +46,9 @@ import org.apache.beam.sdk.transforms.ParDo; import org.apache.beam.sdk.util.GcsUtil; import org.apache.beam.sdk.util.gcsfs.GcsPath; -import org.apache.beam.sdk.util.state.StateSpec; -import org.apache.beam.sdk.util.state.StateSpecs; -import org.apache.beam.sdk.util.state.ValueState; +import org.apache.beam.sdk.state.StateSpec; +import org.apache.beam.sdk.state.StateSpecs; +import org.apache.beam.sdk.state.ValueState; import org.apache.beam.sdk.values.KV; import org.apache.beam.sdk.values.TupleTag; import org.apache.beam.sdk.values.TupleTagList; diff --git a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslatorTest.java b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslatorTest.java index 63e1166ddced..f63cb1ef0ed6 100644 --- a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslatorTest.java +++ b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslatorTest.java @@ -84,9 +84,9 @@ import org.apache.beam.sdk.util.Structs; import org.apache.beam.sdk.util.WindowingStrategy; import org.apache.beam.sdk.util.gcsfs.GcsPath; -import org.apache.beam.sdk.util.state.StateSpec; -import org.apache.beam.sdk.util.state.StateSpecs; -import org.apache.beam.sdk.util.state.ValueState; +import org.apache.beam.sdk.state.StateSpec; +import org.apache.beam.sdk.state.StateSpecs; +import org.apache.beam.sdk.state.ValueState; import org.apache.beam.sdk.values.KV; import org.apache.beam.sdk.values.PCollection; import org.apache.beam.sdk.values.PCollectionTuple; diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/stateful/SparkStateInternals.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/stateful/SparkStateInternals.java index afaba3a571f0..7d7f422bf232 100644 --- a/runners/spark/src/main/java/org/apache/beam/runners/spark/stateful/SparkStateInternals.java +++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/stateful/SparkStateInternals.java @@ -34,16 +34,16 @@ import org.apache.beam.sdk.transforms.CombineWithContext.CombineFnWithContext; import org.apache.beam.sdk.transforms.windowing.TimestampCombiner; import org.apache.beam.sdk.util.CombineFnUtil; -import org.apache.beam.sdk.util.state.BagState; -import org.apache.beam.sdk.util.state.CombiningState; -import org.apache.beam.sdk.util.state.MapState; -import org.apache.beam.sdk.util.state.ReadableState; -import org.apache.beam.sdk.util.state.SetState; -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.ValueState; -import org.apache.beam.sdk.util.state.WatermarkHoldState; +import org.apache.beam.sdk.state.BagState; +import org.apache.beam.sdk.state.CombiningState; +import org.apache.beam.sdk.state.MapState; +import org.apache.beam.sdk.state.ReadableState; +import org.apache.beam.sdk.state.SetState; +import org.apache.beam.sdk.state.State; +import org.apache.beam.sdk.state.StateContext; +import org.apache.beam.sdk.state.StateContexts; +import org.apache.beam.sdk.state.ValueState; +import org.apache.beam.sdk.state.WatermarkHoldState; import org.joda.time.Instant; diff --git a/sdks/java/build-tools/src/main/resources/beam/findbugs-filter.xml b/sdks/java/build-tools/src/main/resources/beam/findbugs-filter.xml index d1d8b4dae92a..d8e4279eb23f 100644 --- a/sdks/java/build-tools/src/main/resources/beam/findbugs-filter.xml +++ b/sdks/java/build-tools/src/main/resources/beam/findbugs-filter.xml @@ -373,7 +373,7 @@ - + diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/BagState.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/BagState.java similarity index 96% rename from sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/BagState.java rename to sdks/java/core/src/main/java/org/apache/beam/sdk/state/BagState.java index e0eebe598dff..189d1517a302 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/BagState.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/BagState.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.beam.sdk.util.state; +package org.apache.beam.sdk.state; /** * State containing a bag values. Items can be added to the bag and the contents read out. diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/CombiningState.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/CombiningState.java similarity index 98% rename from sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/CombiningState.java rename to sdks/java/core/src/main/java/org/apache/beam/sdk/state/CombiningState.java index 80e4dc9478f3..6080127d3800 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/CombiningState.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/CombiningState.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.beam.sdk.util.state; +package org.apache.beam.sdk.state; import org.apache.beam.sdk.transforms.Combine.CombineFn; diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/GroupingState.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/GroupingState.java similarity index 97% rename from sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/GroupingState.java rename to sdks/java/core/src/main/java/org/apache/beam/sdk/state/GroupingState.java index bd7a8d9cf9d6..3a12e79ac4a3 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/GroupingState.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/GroupingState.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.beam.sdk.util.state; +package org.apache.beam.sdk.state; import org.apache.beam.sdk.transforms.Combine.CombineFn; diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/MapState.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/MapState.java similarity index 98% rename from sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/MapState.java rename to sdks/java/core/src/main/java/org/apache/beam/sdk/state/MapState.java index fb7e807804d0..9f0eee9a59cf 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/MapState.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/MapState.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.beam.sdk.util.state; +package org.apache.beam.sdk.state; import java.util.Map; diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/ReadableState.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/ReadableState.java similarity index 97% rename from sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/ReadableState.java rename to sdks/java/core/src/main/java/org/apache/beam/sdk/state/ReadableState.java index c3e9936b9305..b29ab266a1ca 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/ReadableState.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/ReadableState.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.beam.sdk.util.state; +package org.apache.beam.sdk.state; import org.apache.beam.sdk.annotations.Experimental; import org.apache.beam.sdk.annotations.Experimental.Kind; diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/ReadableStates.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/ReadableStates.java similarity index 97% rename from sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/ReadableStates.java rename to sdks/java/core/src/main/java/org/apache/beam/sdk/state/ReadableStates.java index 819eda694a0c..d8df04e8e255 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/ReadableStates.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/ReadableStates.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.beam.sdk.util.state; +package org.apache.beam.sdk.state; import org.apache.beam.sdk.annotations.Experimental; import org.apache.beam.sdk.annotations.Experimental.Kind; diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/SetState.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/SetState.java similarity index 97% rename from sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/SetState.java rename to sdks/java/core/src/main/java/org/apache/beam/sdk/state/SetState.java index 56ea510411be..14aa6406f4fc 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/SetState.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/SetState.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.beam.sdk.util.state; +package org.apache.beam.sdk.state; /** * State containing no duplicate elements. diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/State.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/State.java similarity index 96% rename from sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/State.java rename to sdks/java/core/src/main/java/org/apache/beam/sdk/state/State.java index 3a49f017bec7..6b10c9100492 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/State.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/State.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.beam.sdk.util.state; +package org.apache.beam.sdk.state; /** * Base interface for all state locations. diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/StateBinder.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateBinder.java similarity index 98% rename from sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/StateBinder.java rename to sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateBinder.java index 48fa7428dd28..ee4aa78983b6 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/StateBinder.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateBinder.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.beam.sdk.util.state; +package org.apache.beam.sdk.state; import org.apache.beam.sdk.coders.Coder; import org.apache.beam.sdk.transforms.Combine; diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/StateContext.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateContext.java similarity index 97% rename from sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/StateContext.java rename to sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateContext.java index 887a5f11db1a..110a51569065 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/StateContext.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateContext.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.beam.sdk.util.state; +package org.apache.beam.sdk.state; import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.transforms.windowing.BoundedWindow; diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/StateContexts.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateContexts.java similarity index 97% rename from sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/StateContexts.java rename to sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateContexts.java index 2ce959430569..63afe4f2aada 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/StateContexts.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateContexts.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.beam.sdk.util.state; +package org.apache.beam.sdk.state; import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.transforms.windowing.BoundedWindow; diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/StateSpec.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateSpec.java similarity index 98% rename from sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/StateSpec.java rename to sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateSpec.java index 8eda218193c8..3b0b8407ca97 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/StateSpec.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateSpec.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.beam.sdk.util.state; +package org.apache.beam.sdk.state; import java.io.Serializable; import org.apache.beam.sdk.annotations.Experimental; diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/StateSpecs.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateSpecs.java similarity index 99% rename from sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/StateSpecs.java rename to sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateSpecs.java index 49d57228eb39..09cc4e733bf4 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/StateSpecs.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateSpecs.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.beam.sdk.util.state; +package org.apache.beam.sdk.state; import static com.google.common.base.Preconditions.checkArgument; diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/ValueState.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/ValueState.java similarity index 96% rename from sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/ValueState.java rename to sdks/java/core/src/main/java/org/apache/beam/sdk/state/ValueState.java index b4322039a6c6..ca97db2fae4b 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/ValueState.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/ValueState.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.beam.sdk.util.state; +package org.apache.beam.sdk.state; import org.apache.beam.sdk.annotations.Experimental; import org.apache.beam.sdk.annotations.Experimental.Kind; diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/WatermarkHoldState.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/WatermarkHoldState.java similarity index 97% rename from sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/WatermarkHoldState.java rename to sdks/java/core/src/main/java/org/apache/beam/sdk/state/WatermarkHoldState.java index ae9b700f1fa2..9f6c20383747 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/WatermarkHoldState.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/WatermarkHoldState.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.beam.sdk.util.state; +package org.apache.beam.sdk.state; import org.apache.beam.sdk.annotations.Experimental; import org.apache.beam.sdk.annotations.Experimental.Kind; diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/package-info.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/package-info.java similarity index 95% rename from sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/package-info.java rename to sdks/java/core/src/main/java/org/apache/beam/sdk/state/package-info.java index b9bec1600d01..de5eeeb22468 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/package-info.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/package-info.java @@ -19,4 +19,4 @@ /** * Defines internal utilities for interacting with pipeline state. */ -package org.apache.beam.sdk.util.state; +package org.apache.beam.sdk.state; diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/UsesMapState.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/UsesMapState.java index 9bced418d225..585d8b7e481a 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/UsesMapState.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/UsesMapState.java @@ -17,7 +17,7 @@ */ package org.apache.beam.sdk.testing; -import org.apache.beam.sdk.util.state.MapState; +import org.apache.beam.sdk.state.MapState; /** * Category tag for validation tests which utilize {@link MapState}. diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/UsesSetState.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/UsesSetState.java index 6fd74bdf4f13..7d82d227a3c9 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/UsesSetState.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/UsesSetState.java @@ -17,7 +17,7 @@ */ package org.apache.beam.sdk.testing; -import org.apache.beam.sdk.util.state.SetState; +import org.apache.beam.sdk.state.SetState; /** * Category tag for validation tests which utilize {@link SetState}. diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFn.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFn.java index 036847672e25..7c9d5c925e83 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFn.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFn.java @@ -37,8 +37,8 @@ import org.apache.beam.sdk.util.TimeDomain; import org.apache.beam.sdk.util.Timer; import org.apache.beam.sdk.util.TimerSpec; -import org.apache.beam.sdk.util.state.State; -import org.apache.beam.sdk.util.state.StateSpec; +import org.apache.beam.sdk.state.State; +import org.apache.beam.sdk.state.StateSpec; import org.apache.beam.sdk.values.PCollection; import org.apache.beam.sdk.values.PCollectionView; import org.apache.beam.sdk.values.TupleTag; diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnTester.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnTester.java index 70fb0ae9cb0b..6138c0060db7 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnTester.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnTester.java @@ -314,7 +314,7 @@ public RestrictionTracker restrictionTracker() { } @Override - public org.apache.beam.sdk.util.state.State state(String stateId) { + public org.apache.beam.sdk.state.State state(String stateId) { throw new UnsupportedOperationException("DoFnTester doesn't support state yet"); } diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupIntoBatches.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupIntoBatches.java index 02f3a8598f88..9c0918e9e8b2 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupIntoBatches.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupIntoBatches.java @@ -29,11 +29,11 @@ import org.apache.beam.sdk.util.Timer; import org.apache.beam.sdk.util.TimerSpec; import org.apache.beam.sdk.util.TimerSpecs; -import org.apache.beam.sdk.util.state.BagState; -import org.apache.beam.sdk.util.state.CombiningState; -import org.apache.beam.sdk.util.state.StateSpec; -import org.apache.beam.sdk.util.state.StateSpecs; -import org.apache.beam.sdk.util.state.ValueState; +import org.apache.beam.sdk.state.BagState; +import org.apache.beam.sdk.state.CombiningState; +import org.apache.beam.sdk.state.StateSpec; +import org.apache.beam.sdk.state.StateSpecs; +import org.apache.beam.sdk.state.ValueState; import org.apache.beam.sdk.values.KV; import org.apache.beam.sdk.values.PCollection; import org.joda.time.Duration; diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java index 6137a7b6663b..9840c2b7f8c5 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java @@ -45,7 +45,7 @@ import org.apache.beam.sdk.transforms.windowing.WindowFn; import org.apache.beam.sdk.util.NameUtils; import org.apache.beam.sdk.util.SerializableUtils; -import org.apache.beam.sdk.util.state.StateSpec; +import org.apache.beam.sdk.state.StateSpec; import org.apache.beam.sdk.values.PCollection; import org.apache.beam.sdk.values.PCollectionTuple; import org.apache.beam.sdk.values.PCollectionView; diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnInvoker.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnInvoker.java index 0fbcc84cfd9a..3171c6b18acb 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnInvoker.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnInvoker.java @@ -28,7 +28,7 @@ import org.apache.beam.sdk.transforms.splittabledofn.RestrictionTracker; import org.apache.beam.sdk.transforms.windowing.BoundedWindow; import org.apache.beam.sdk.util.Timer; -import org.apache.beam.sdk.util.state.State; +import org.apache.beam.sdk.state.State; /** * Interface for invoking the {@code DoFn} processing methods. diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignature.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignature.java index 1be741f2db39..d205a7ade7ca 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignature.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignature.java @@ -38,8 +38,8 @@ import org.apache.beam.sdk.transforms.windowing.BoundedWindow; import org.apache.beam.sdk.util.Timer; import org.apache.beam.sdk.util.TimerSpec; -import org.apache.beam.sdk.util.state.State; -import org.apache.beam.sdk.util.state.StateSpec; +import org.apache.beam.sdk.state.State; +import org.apache.beam.sdk.state.StateSpec; import org.apache.beam.sdk.values.PCollection; import org.apache.beam.sdk.values.TypeDescriptor; diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignatures.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignatures.java index 666c7f69b3b7..49631f007061 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignatures.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignatures.java @@ -58,8 +58,8 @@ import org.apache.beam.sdk.util.Timer; import org.apache.beam.sdk.util.TimerSpec; import org.apache.beam.sdk.util.common.ReflectHelpers; -import org.apache.beam.sdk.util.state.State; -import org.apache.beam.sdk.util.state.StateSpec; +import org.apache.beam.sdk.state.State; +import org.apache.beam.sdk.state.StateSpec; import org.apache.beam.sdk.values.PCollection; import org.apache.beam.sdk.values.TypeDescriptor; import org.apache.beam.sdk.values.TypeParameter; diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/CombineContextFactory.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/CombineContextFactory.java index 31d1f645fd7f..7d9342d320f9 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/CombineContextFactory.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/CombineContextFactory.java @@ -20,7 +20,7 @@ import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.transforms.CombineWithContext.Context; import org.apache.beam.sdk.transforms.windowing.BoundedWindow; -import org.apache.beam.sdk.util.state.StateContext; +import org.apache.beam.sdk.state.StateContext; import org.apache.beam.sdk.values.PCollectionView; /** diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/CombineFnUtil.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/CombineFnUtil.java index a3941803c2f0..6e3db6f17ca5 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/CombineFnUtil.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/CombineFnUtil.java @@ -28,7 +28,7 @@ import org.apache.beam.sdk.transforms.CombineWithContext.CombineFnWithContext; import org.apache.beam.sdk.transforms.CombineWithContext.Context; import org.apache.beam.sdk.transforms.display.DisplayData; -import org.apache.beam.sdk.util.state.StateContext; +import org.apache.beam.sdk.state.StateContext; /** * Static utility methods that create combine function instances. diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java index 26904aa09d15..47d571fdefe8 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java @@ -92,13 +92,13 @@ import org.apache.beam.sdk.util.TimerSpec; import org.apache.beam.sdk.util.TimerSpecs; import org.apache.beam.sdk.util.common.ElementByteSizeObserver; -import org.apache.beam.sdk.util.state.BagState; -import org.apache.beam.sdk.util.state.CombiningState; -import org.apache.beam.sdk.util.state.MapState; -import org.apache.beam.sdk.util.state.SetState; -import org.apache.beam.sdk.util.state.StateSpec; -import org.apache.beam.sdk.util.state.StateSpecs; -import org.apache.beam.sdk.util.state.ValueState; +import org.apache.beam.sdk.state.BagState; +import org.apache.beam.sdk.state.CombiningState; +import org.apache.beam.sdk.state.MapState; +import org.apache.beam.sdk.state.SetState; +import org.apache.beam.sdk.state.StateSpec; +import org.apache.beam.sdk.state.StateSpecs; +import org.apache.beam.sdk.state.ValueState; import org.apache.beam.sdk.values.KV; import org.apache.beam.sdk.values.PCollection; import org.apache.beam.sdk.values.PCollectionTuple; diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnInvokersTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnInvokersTest.java index c16eea2ba5c9..c564907b3caf 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnInvokersTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnInvokersTest.java @@ -50,9 +50,9 @@ import org.apache.beam.sdk.util.TimerSpec; import org.apache.beam.sdk.util.TimerSpecs; import org.apache.beam.sdk.util.UserCodeException; -import org.apache.beam.sdk.util.state.StateSpec; -import org.apache.beam.sdk.util.state.StateSpecs; -import org.apache.beam.sdk.util.state.ValueState; +import org.apache.beam.sdk.state.StateSpec; +import org.apache.beam.sdk.state.StateSpecs; +import org.apache.beam.sdk.state.ValueState; import org.joda.time.Instant; import org.junit.Before; import org.junit.Rule; diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnSignaturesTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnSignaturesTest.java index d6cc4f6a51f1..4688105447be 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnSignaturesTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnSignaturesTest.java @@ -41,10 +41,10 @@ import org.apache.beam.sdk.util.Timer; import org.apache.beam.sdk.util.TimerSpec; import org.apache.beam.sdk.util.TimerSpecs; -import org.apache.beam.sdk.util.state.StateSpec; -import org.apache.beam.sdk.util.state.StateSpecs; -import org.apache.beam.sdk.util.state.ValueState; -import org.apache.beam.sdk.util.state.WatermarkHoldState; +import org.apache.beam.sdk.state.StateSpec; +import org.apache.beam.sdk.state.StateSpecs; +import org.apache.beam.sdk.state.ValueState; +import org.apache.beam.sdk.state.WatermarkHoldState; import org.apache.beam.sdk.values.KV; import org.apache.beam.sdk.values.TypeDescriptor; import org.hamcrest.Matcher; diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/CombineFnUtilTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/CombineFnUtilTest.java index 798e8dca4663..8490d50bcf27 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/CombineFnUtilTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/CombineFnUtilTest.java @@ -30,7 +30,7 @@ import org.apache.beam.sdk.transforms.CombineWithContext.CombineFnWithContext; import org.apache.beam.sdk.transforms.CombineWithContext.Context; import org.apache.beam.sdk.transforms.Sum; -import org.apache.beam.sdk.util.state.StateContexts; +import org.apache.beam.sdk.state.StateContexts; import org.junit.Before; import org.junit.Rule; import org.junit.Test; From 6e521178015944c44f6a7eb4cc1edd7eaf92486a Mon Sep 17 00:00:00 2001 From: Kenneth Knowles Date: Tue, 2 May 2017 10:41:01 -0700 Subject: [PATCH 2/2] Add @Internal and @Experimental to state package --- .../org/apache/beam/sdk/state/BagState.java | 4 + .../apache/beam/sdk/state/CombiningState.java | 3 + .../apache/beam/sdk/state/GroupingState.java | 3 + .../org/apache/beam/sdk/state/MapState.java | 3 + .../apache/beam/sdk/state/ReadableStates.java | 7 +- .../org/apache/beam/sdk/state/SetState.java | 4 + .../java/org/apache/beam/sdk/state/State.java | 4 + .../apache/beam/sdk/state/StateBinder.java | 7 +- .../apache/beam/sdk/state/StateContext.java | 5 +- .../apache/beam/sdk/state/StateContexts.java | 5 +- .../org/apache/beam/sdk/state/StateSpec.java | 16 +++- .../org/apache/beam/sdk/state/StateSpecs.java | 88 +++++++++++-------- .../beam/sdk/state/WatermarkHoldState.java | 11 ++- .../apache/beam/sdk/state/package-info.java | 2 +- 14 files changed, 108 insertions(+), 54 deletions(-) diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/state/BagState.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/BagState.java index 189d1517a302..10fba05d1fbb 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/state/BagState.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/BagState.java @@ -17,11 +17,15 @@ */ package org.apache.beam.sdk.state; +import org.apache.beam.sdk.annotations.Experimental; +import org.apache.beam.sdk.annotations.Experimental.Kind; + /** * State containing a bag values. Items can be added to the bag and the contents read out. * * @param The type of elements in the bag. */ +@Experimental(Kind.STATE) public interface BagState extends GroupingState> { @Override BagState readLater(); diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/state/CombiningState.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/CombiningState.java index 6080127d3800..ddda25519e3f 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/state/CombiningState.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/CombiningState.java @@ -17,6 +17,8 @@ */ package org.apache.beam.sdk.state; +import org.apache.beam.sdk.annotations.Experimental; +import org.apache.beam.sdk.annotations.Experimental.Kind; import org.apache.beam.sdk.transforms.Combine.CombineFn; /** @@ -27,6 +29,7 @@ * @param the type of accumulator * @param the type of value extracted from the state */ +@Experimental(Kind.STATE) public interface CombiningState extends GroupingState { diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/state/GroupingState.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/GroupingState.java index 3a12e79ac4a3..d99ff25515cf 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/state/GroupingState.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/GroupingState.java @@ -17,6 +17,8 @@ */ package org.apache.beam.sdk.state; +import org.apache.beam.sdk.annotations.Experimental; +import org.apache.beam.sdk.annotations.Experimental.Kind; import org.apache.beam.sdk.transforms.Combine.CombineFn; /** @@ -26,6 +28,7 @@ * @param the type of values added to the state * @param the type of value extracted from the state */ +@Experimental(Kind.STATE) public interface GroupingState extends ReadableState, State { /** * Add a value to the buffer. diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/state/MapState.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/MapState.java index 9f0eee9a59cf..649c3c7653d8 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/state/MapState.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/MapState.java @@ -18,6 +18,8 @@ package org.apache.beam.sdk.state; import java.util.Map; +import org.apache.beam.sdk.annotations.Experimental; +import org.apache.beam.sdk.annotations.Experimental.Kind; /** * An object that maps keys to values. @@ -27,6 +29,7 @@ * @param the type of keys maintained by this map * @param the type of mapped values */ +@Experimental(Kind.STATE) public interface MapState extends State { /** diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/state/ReadableStates.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/ReadableStates.java index d8df04e8e255..6977a9718bb3 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/state/ReadableStates.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/ReadableStates.java @@ -17,13 +17,12 @@ */ package org.apache.beam.sdk.state; -import org.apache.beam.sdk.annotations.Experimental; -import org.apache.beam.sdk.annotations.Experimental.Kind; +import org.apache.beam.sdk.annotations.Internal; /** - * Utilities for constructing and manipulating {@link ReadableState} instances. + * For internal use only; no backwards-compatibility guarantees. */ -@Experimental(Kind.STATE) +@Internal public class ReadableStates { /** diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/state/SetState.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/SetState.java index 14aa6406f4fc..cb9a0e6dd780 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/state/SetState.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/SetState.java @@ -17,12 +17,16 @@ */ package org.apache.beam.sdk.state; +import org.apache.beam.sdk.annotations.Experimental; +import org.apache.beam.sdk.annotations.Experimental.Kind; + /** * State containing no duplicate elements. * Items can be added to the set and the contents read out. * * @param The type of elements in the set. */ +@Experimental(Kind.STATE) public interface SetState extends GroupingState> { /** * Returns true if this set contains the specified element. diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/state/State.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/State.java index 6b10c9100492..0c0ca32f2ecc 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/state/State.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/State.java @@ -17,12 +17,16 @@ */ package org.apache.beam.sdk.state; +import org.apache.beam.sdk.annotations.Experimental; +import org.apache.beam.sdk.annotations.Experimental.Kind; + /** * Base interface for all state locations. * *

Specific types of state add appropriate accessors for reading and writing values, see * {@link ValueState}, {@link BagState}, and {@link GroupingState}. */ +@Experimental(Kind.STATE) public interface State { /** diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateBinder.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateBinder.java index ee4aa78983b6..af834c3cdc91 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateBinder.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateBinder.java @@ -17,14 +17,19 @@ */ package org.apache.beam.sdk.state; +import org.apache.beam.sdk.annotations.Internal; import org.apache.beam.sdk.coders.Coder; import org.apache.beam.sdk.transforms.Combine; import org.apache.beam.sdk.transforms.CombineWithContext; import org.apache.beam.sdk.transforms.windowing.TimestampCombiner; /** - * Visitor for binding a {@link StateSpec} and to the associated {@link State}. + * For internal use only; no backwards-compatibility guarantees. + * For internal use only; no backwards-compatibility guarantees. + * + *

Visitor for binding a {@link StateSpec} and to the associated {@link State}. */ +@Internal public interface StateBinder { ValueState bindValue( String id, StateSpec> spec, Coder coder); diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateContext.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateContext.java index 110a51569065..52177cc0a41c 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateContext.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateContext.java @@ -17,13 +17,16 @@ */ package org.apache.beam.sdk.state; +import org.apache.beam.sdk.annotations.Internal; import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.transforms.windowing.BoundedWindow; import org.apache.beam.sdk.values.PCollectionView; /** - * Information accessible the state API. + * For internal use only; no backwards-compatibility guarantees. + * For internal use only; no backwards-compatibility guarantees. */ +@Internal public interface StateContext { /** * Returns the {@code PipelineOptions} specified with the diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateContexts.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateContexts.java index 63afe4f2aada..2e21a27c1a6a 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateContexts.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateContexts.java @@ -17,13 +17,16 @@ */ package org.apache.beam.sdk.state; +import org.apache.beam.sdk.annotations.Internal; import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.transforms.windowing.BoundedWindow; import org.apache.beam.sdk.values.PCollectionView; /** - * Factory that produces {@link StateContext} based on different inputs. + * For internal use only; no backwards-compatibility guarantees. + * For internal use only; no backwards-compatibility guarantees. */ +@Internal public class StateContexts { private static final StateContext NULL_CONTEXT = new StateContext() { diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateSpec.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateSpec.java index 3b0b8407ca97..e51235ed06da 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateSpec.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateSpec.java @@ -20,6 +20,7 @@ import java.io.Serializable; import org.apache.beam.sdk.annotations.Experimental; import org.apache.beam.sdk.annotations.Experimental.Kind; +import org.apache.beam.sdk.annotations.Internal; import org.apache.beam.sdk.coders.Coder; /** @@ -32,22 +33,31 @@ public interface StateSpec extends Serializable { /** - * Use the {@code binder} to create an instance of {@code StateT} appropriate for this address. + * For internal use only; no backwards-compatibility guarantees. + * + *

Use the {@code binder} to create an instance of {@code StateT} appropriate for this address. */ + @Internal StateT bind(String id, StateBinder binder); /** - * Given {code coders} are inferred from type arguments defined for this class. Coders which are + * For internal use only; no backwards-compatibility guarantees. + * + *

Given {code coders} are inferred from type arguments defined for this class. Coders which are * already set should take precedence over offered coders. * * @param coders Array of coders indexed by the type arguments order. Entries might be null if the * coder could not be inferred. */ + @Internal void offerCoders(Coder[] coders); /** - * Validates that this {@link StateSpec} has been specified correctly and finalizes it. + * For internal use only; no backwards-compatibility guarantees. + * + *

Validates that this {@link StateSpec} has been specified correctly and finalizes it. * Automatically invoked when the pipeline is built. */ + @Internal void finishSpecifying(); } diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateSpecs.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateSpecs.java index 09cc4e733bf4..8a3c87e4c6a4 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateSpecs.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateSpecs.java @@ -23,6 +23,7 @@ import javax.annotation.Nullable; import org.apache.beam.sdk.annotations.Experimental; import org.apache.beam.sdk.annotations.Experimental.Kind; +import org.apache.beam.sdk.annotations.Internal; import org.apache.beam.sdk.coders.CannotProvideCoderException; import org.apache.beam.sdk.coders.Coder; import org.apache.beam.sdk.coders.CoderRegistry; @@ -94,42 +95,6 @@ StateSpec> combining( return combiningInternal(accumCoder, combineFn); } - /** - * Create a state spec for values that use a {@link CombineFn} to automatically merge multiple - * {@code InputT}s into a single {@code OutputT}. - * - *

This determines the {@code Coder} from the given {@code Coder}, and should - * only be used to initialize static values. - */ - public static - StateSpec> - combiningFromInputInternal( - Coder inputCoder, CombineFn combineFn) { - try { - Coder accumCoder = combineFn.getAccumulatorCoder(STANDARD_REGISTRY, inputCoder); - return combiningInternal(accumCoder, combineFn); - } catch (CannotProvideCoderException e) { - throw new IllegalArgumentException( - "Unable to determine accumulator coder for " - + combineFn.getClass().getSimpleName() - + " from " - + inputCoder, - e); - } - } - - private static - StateSpec> combiningInternal( - Coder accumCoder, CombineFn combineFn) { - return new CombiningStateSpec(accumCoder, combineFn); - } - - private static - StateSpec> combiningInternal( - Coder accumCoder, CombineFnWithContext combineFn) { - return new CombiningWithContextStateSpec(accumCoder, combineFn); - } - /** * Create a state spec that is optimized for adding values frequently, and occasionally retrieving * all the values that have been added. @@ -172,13 +137,62 @@ public static StateSpec> map(Coder keyCoder, Coder v return new MapStateSpec<>(keyCoder, valueCoder); } - /** Create a state spec for holding the watermark. */ + /** + * For internal use only; no backwards-compatibility guarantees. + * + *

Create a state spec for values that use a {@link CombineFn} to automatically merge multiple + * {@code InputT}s into a single {@code OutputT}. + * + *

This determines the {@code Coder} from the given {@code Coder}, and should + * only be used to initialize static values. + */ + @Internal + public static + StateSpec> combiningFromInputInternal( + Coder inputCoder, CombineFn combineFn) { + try { + Coder accumCoder = combineFn.getAccumulatorCoder(STANDARD_REGISTRY, inputCoder); + return combiningInternal(accumCoder, combineFn); + } catch (CannotProvideCoderException e) { + throw new IllegalArgumentException( + "Unable to determine accumulator coder for " + + combineFn.getClass().getSimpleName() + + " from " + + inputCoder, + e); + } + } + + private static + StateSpec> combiningInternal( + Coder accumCoder, CombineFn combineFn) { + return new CombiningStateSpec(accumCoder, combineFn); + } + + private static + StateSpec> combiningInternal( + Coder accumCoder, CombineFnWithContext combineFn) { + return new CombiningWithContextStateSpec(accumCoder, combineFn); + } + + /** + * For internal use only; no backwards-compatibility guarantees. + * + *

Create a state spec for a watermark hold. + */ + @Internal public static StateSpec watermarkStateInternal( TimestampCombiner timestampCombiner) { return new WatermarkStateSpecInternal(timestampCombiner); } + /** + * For internal use only; no backwards-compatibility guarantees. + * + *

Convert a combining state spec to a bag of accumulators. + */ + @Internal public static StateSpec> convertToBagSpecInternal( StateSpec> combiningSpec) { diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/state/WatermarkHoldState.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/WatermarkHoldState.java index 9f6c20383747..38e2cbc27b64 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/state/WatermarkHoldState.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/WatermarkHoldState.java @@ -17,18 +17,17 @@ */ package org.apache.beam.sdk.state; -import org.apache.beam.sdk.annotations.Experimental; -import org.apache.beam.sdk.annotations.Experimental.Kind; +import org.apache.beam.sdk.annotations.Internal; import org.apache.beam.sdk.transforms.windowing.TimestampCombiner; import org.joda.time.Instant; /** - * A {@link State} accepting and aggregating output timestamps, which determines the time to which - * the output watermark must be held. + * For internal use only; no backwards-compatibility guarantees. * - *

For internal use only. This API may change at any time. + *

A {@link State} accepting and aggregating output timestamps, which determines the time to + * which the output watermark must be held. */ -@Experimental(Kind.STATE) +@Internal public interface WatermarkHoldState extends GroupingState { /** * Return the {@link TimestampCombiner} which will be used to determine a watermark hold time diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/state/package-info.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/package-info.java index de5eeeb22468..d8b8e9266c2e 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/state/package-info.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/package-info.java @@ -17,6 +17,6 @@ */ /** - * Defines internal utilities for interacting with pipeline state. + * Classes and interfaces for interacting with state. */ package org.apache.beam.sdk.state;