http://git-wip-us.apache.org/repos/asf/beam/blob/ca41af8f/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/ByteBuddyDoFnInvokerFactory.java ---------------------------------------------------------------------- diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/ByteBuddyDoFnInvokerFactory.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/ByteBuddyDoFnInvokerFactory.java index 2b6d5be..8ae2c65 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/ByteBuddyDoFnInvokerFactory.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/ByteBuddyDoFnInvokerFactory.java @@ -63,6 +63,7 @@ import net.bytebuddy.matcher.ElementMatchers; import org.apache.beam.sdk.coders.CannotProvideCoderException; import org.apache.beam.sdk.coders.Coder; import org.apache.beam.sdk.coders.CoderRegistry; +import org.apache.beam.sdk.state.Timer; import org.apache.beam.sdk.transforms.DoFn; import org.apache.beam.sdk.transforms.DoFn.ProcessElement; import org.apache.beam.sdk.transforms.reflect.DoFnSignature.OnTimerMethod; @@ -77,7 +78,6 @@ import org.apache.beam.sdk.transforms.reflect.DoFnSignature.Parameter.TimerParam import org.apache.beam.sdk.transforms.reflect.DoFnSignature.Parameter.WindowParameter; import org.apache.beam.sdk.transforms.splittabledofn.HasDefaultTracker; import org.apache.beam.sdk.transforms.splittabledofn.RestrictionTracker; -import org.apache.beam.sdk.util.Timer; import org.apache.beam.sdk.util.UserCodeException; import org.apache.beam.sdk.values.TypeDescriptor;
http://git-wip-us.apache.org/repos/asf/beam/blob/ca41af8f/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnInvoker.java ---------------------------------------------------------------------- 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 d5a1a94..6fd4052 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 @@ -20,6 +20,7 @@ package org.apache.beam.sdk.transforms.reflect; import org.apache.beam.sdk.coders.Coder; import org.apache.beam.sdk.coders.CoderRegistry; import org.apache.beam.sdk.state.State; +import org.apache.beam.sdk.state.Timer; import org.apache.beam.sdk.transforms.DoFn; import org.apache.beam.sdk.transforms.DoFn.FinishBundle; import org.apache.beam.sdk.transforms.DoFn.ProcessElement; @@ -28,7 +29,6 @@ import org.apache.beam.sdk.transforms.DoFn.StateId; import org.apache.beam.sdk.transforms.DoFn.TimerId; import org.apache.beam.sdk.transforms.splittabledofn.RestrictionTracker; import org.apache.beam.sdk.transforms.windowing.BoundedWindow; -import org.apache.beam.sdk.util.Timer; /** * Interface for invoking the {@code DoFn} processing methods. http://git-wip-us.apache.org/repos/asf/beam/blob/ca41af8f/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignature.java ---------------------------------------------------------------------- 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 72ad4b0..0b4bf90 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 @@ -29,6 +29,8 @@ import javax.annotation.Nullable; import org.apache.beam.sdk.coders.Coder; import org.apache.beam.sdk.state.State; import org.apache.beam.sdk.state.StateSpec; +import org.apache.beam.sdk.state.Timer; +import org.apache.beam.sdk.state.TimerSpec; import org.apache.beam.sdk.transforms.DoFn; import org.apache.beam.sdk.transforms.DoFn.StateId; import org.apache.beam.sdk.transforms.DoFn.TimerId; @@ -38,8 +40,6 @@ import org.apache.beam.sdk.transforms.reflect.DoFnSignature.Parameter.TimerParam import org.apache.beam.sdk.transforms.reflect.DoFnSignature.Parameter.WindowParameter; 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.TimerSpec; import org.apache.beam.sdk.values.PCollection; import org.apache.beam.sdk.values.TypeDescriptor; http://git-wip-us.apache.org/repos/asf/beam/blob/ca41af8f/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignatures.java ---------------------------------------------------------------------- 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 3dfca8c..bb191b1 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 @@ -44,6 +44,8 @@ import javax.annotation.Nullable; import org.apache.beam.sdk.coders.Coder; import org.apache.beam.sdk.state.State; import org.apache.beam.sdk.state.StateSpec; +import org.apache.beam.sdk.state.Timer; +import org.apache.beam.sdk.state.TimerSpec; import org.apache.beam.sdk.transforms.DoFn; import org.apache.beam.sdk.transforms.DoFn.StateId; import org.apache.beam.sdk.transforms.DoFn.TimerId; @@ -57,8 +59,6 @@ import org.apache.beam.sdk.transforms.reflect.DoFnSignature.TimerDeclaration; import org.apache.beam.sdk.transforms.splittabledofn.HasDefaultTracker; 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.TimerSpec; import org.apache.beam.sdk.util.common.ReflectHelpers; import org.apache.beam.sdk.values.PCollection; import org.apache.beam.sdk.values.TypeDescriptor; http://git-wip-us.apache.org/repos/asf/beam/blob/ca41af8f/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/AfterWatermark.java ---------------------------------------------------------------------- diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/AfterWatermark.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/AfterWatermark.java index 14a8c98..5648895 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/AfterWatermark.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/AfterWatermark.java @@ -24,8 +24,8 @@ import java.util.List; import java.util.Objects; import javax.annotation.Nullable; import org.apache.beam.sdk.annotations.Experimental; +import org.apache.beam.sdk.state.TimeDomain; import org.apache.beam.sdk.transforms.windowing.Trigger.OnceTrigger; -import org.apache.beam.sdk.util.TimeDomain; import org.joda.time.Instant; /** http://git-wip-us.apache.org/repos/asf/beam/blob/ca41af8f/sdks/java/core/src/main/java/org/apache/beam/sdk/util/ClassPath.java ---------------------------------------------------------------------- diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/ClassPath.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/ClassPath.java index 35632ed..271bce0 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/ClassPath.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/ClassPath.java @@ -19,6 +19,7 @@ package org.apache.beam.sdk.util; import static com.google.common.base.Preconditions.checkArgument; import static com.google.common.base.Preconditions.checkNotNull; + import com.google.common.annotations.Beta; import com.google.common.annotations.VisibleForTesting; import com.google.common.base.CharMatcher; http://git-wip-us.apache.org/repos/asf/beam/blob/ca41af8f/sdks/java/core/src/main/java/org/apache/beam/sdk/util/TimeDomain.java ---------------------------------------------------------------------- diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/TimeDomain.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/TimeDomain.java deleted file mode 100644 index b067ab8..0000000 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/TimeDomain.java +++ /dev/null @@ -1,43 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.beam.sdk.util; - -/** - * {@code TimeDomain} specifies whether an operation is based on - * timestamps of elements or current "real-world" time as reported while processing. - */ -public enum TimeDomain { - /** - * The {@code EVENT_TIME} domain corresponds to the timestamps on the elements. Time advances - * on the system watermark advances. - */ - EVENT_TIME, - - /** - * The {@code PROCESSING_TIME} domain corresponds to the current to the current (system) time. - * This is advanced during execution of the pipeline. - */ - PROCESSING_TIME, - - /** - * Same as the {@code PROCESSING_TIME} domain, except it won't fire a timer set for time - * {@code T} until all timers from earlier stages set for a time earlier than {@code T} have - * fired. - */ - SYNCHRONIZED_PROCESSING_TIME -} http://git-wip-us.apache.org/repos/asf/beam/blob/ca41af8f/sdks/java/core/src/main/java/org/apache/beam/sdk/util/Timer.java ---------------------------------------------------------------------- diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/Timer.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/Timer.java deleted file mode 100644 index 9727969..0000000 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/Timer.java +++ /dev/null @@ -1,78 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.beam.sdk.util; - -import org.apache.beam.sdk.annotations.Experimental; -import org.apache.beam.sdk.transforms.DoFn; -import org.joda.time.Duration; -import org.joda.time.Instant; - -/** - * A timer for a specified time domain that can be set to register the desire for further processing - * at particular time in its specified time domain. - * - * <p>See {@link TimeDomain} for details on the time domains available. - * - * <p>In a {@link DoFn}, a {@link Timer} is specified by a {@link TimerSpec} annotated with {@link - * DoFn.TimerId}. - * - * <p>An implementation of {@link Timer} is implicitly scoped - it may be scoped to a key and - * window, or a key, window, and trigger, etc. - * - * <p>A timer exists in one of two states: set or unset. A timer can be set only for a single time - * per scope. - * - * <p>Timer callbacks are not guaranteed to be called immediately according to the local view of the - * {@link TimeDomain}, but will be called at some time after the requested time, in timestamp - * order. - */ -@Experimental(Experimental.Kind.TIMERS) -public interface Timer { - /** - * Sets or resets the time in the timer's {@link TimeDomain} at which it should fire. If the timer - * was already set, resets it to the new requested time. - * - * <p>For {@link TimeDomain#PROCESSING_TIME}, the behavior is be unpredictable, since processing - * time timers are ignored after a window has expired. Instead, it is recommended to use - * {@link #setRelative()}. - */ - void set(Instant absoluteTime); - - /** - * Unsets this timer. It is permitted to {@code cancel()} whether or not the timer was actually - * set. - */ - void cancel(); - - /** - * Sets the timer relative to the current time, according to any offset and alignment specified. - * Using {@link #offset(Duration)} and {@link #align(Duration)}. - */ - void setRelative(); - - /** - * Set the align offset. - */ - Timer offset(Duration offset); - - /** - * Aligns a timestamp to the next boundary of {@code period}. - */ - Timer align(Duration period); - -} http://git-wip-us.apache.org/repos/asf/beam/blob/ca41af8f/sdks/java/core/src/main/java/org/apache/beam/sdk/util/TimerSpec.java ---------------------------------------------------------------------- diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/TimerSpec.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/TimerSpec.java deleted file mode 100644 index 5b7717b..0000000 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/TimerSpec.java +++ /dev/null @@ -1,30 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.beam.sdk.util; - -import java.io.Serializable; -import org.apache.beam.sdk.annotations.Experimental; -import org.apache.beam.sdk.annotations.Experimental.Kind; - -/** - * A specification for a {@link Timer}. This includes its {@link TimeDomain}. - */ -@Experimental(Kind.TIMERS) -public interface TimerSpec extends Serializable { - TimeDomain getTimeDomain(); -} http://git-wip-us.apache.org/repos/asf/beam/blob/ca41af8f/sdks/java/core/src/main/java/org/apache/beam/sdk/util/TimerSpecs.java ---------------------------------------------------------------------- diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/TimerSpecs.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/TimerSpecs.java deleted file mode 100644 index f289ccd..0000000 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/TimerSpecs.java +++ /dev/null @@ -1,41 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.beam.sdk.util; - -import com.google.auto.value.AutoValue; -import org.apache.beam.sdk.annotations.Experimental; -import org.apache.beam.sdk.annotations.Experimental.Kind; - -/** - * Static methods for working with {@link TimerSpec}. - */ -@Experimental(Kind.TIMERS) -public class TimerSpecs { - - public static TimerSpec timer(TimeDomain timeDomain) { - return new AutoValue_TimerSpecs_SimpleTimerSpec(timeDomain); - } - - /** - * A straightforward POJO {@link TimerSpec}. Package-level access for AutoValue. - */ - @AutoValue - abstract static class SimpleTimerSpec implements TimerSpec { - public abstract TimeDomain getTimeDomain(); - } -} http://git-wip-us.apache.org/repos/asf/beam/blob/ca41af8f/sdks/java/core/src/main/java/org/apache/beam/sdk/util/Timers.java ---------------------------------------------------------------------- diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/Timers.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/Timers.java deleted file mode 100644 index 7415d7a..0000000 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/Timers.java +++ /dev/null @@ -1,58 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.beam.sdk.util; - -import javax.annotation.Nullable; -import org.apache.beam.sdk.annotations.Experimental; -import org.joda.time.Instant; - -/** - * Interface for interacting with time. - */ -@Experimental(Experimental.Kind.TIMERS) -public interface Timers { - /** - * Sets a timer to fire when the event time watermark, the current processing time, or - * the synchronized processing time watermark surpasses a given timestamp. - * - * <p>See {@link TimeDomain} for details on the time domains available. - * - * <p>Timers are not guaranteed to fire immediately, but will be delivered at some time - * afterwards. - * - * <p>An implementation of {@link Timers} implicitly scopes timers that are set - they may - * be scoped to a key and window, or a key, window, and trigger, etc. - * - * @param timestamp the time at which the timer should be delivered - * @param timeDomain the domain that the {@code timestamp} applies to - */ - void setTimer(Instant timestamp, TimeDomain timeDomain); - - /** Removes the timer set in this context for the {@code timestmap} and {@code timeDomain}. */ - void deleteTimer(Instant timestamp, TimeDomain timeDomain); - - /** Returns the current processing time. */ - Instant currentProcessingTime(); - - /** Returns the current synchronized processing time or {@code null} if unknown. */ - @Nullable - Instant currentSynchronizedProcessingTime(); - - /** Returns the current event time. */ - Instant currentEventTime(); -} http://git-wip-us.apache.org/repos/asf/beam/blob/ca41af8f/sdks/java/core/src/test/java/org/apache/beam/sdk/io/DrunkWritableByteChannelFactory.java ---------------------------------------------------------------------- diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/DrunkWritableByteChannelFactory.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/DrunkWritableByteChannelFactory.java index 79f0996..6615a2e 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/DrunkWritableByteChannelFactory.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/DrunkWritableByteChannelFactory.java @@ -20,7 +20,6 @@ package org.apache.beam.sdk.io; import java.io.IOException; import java.nio.ByteBuffer; import java.nio.channels.WritableByteChannel; - import org.apache.beam.sdk.io.FileBasedSink.WritableByteChannelFactory; import org.apache.beam.sdk.util.MimeTypes; http://git-wip-us.apache.org/repos/asf/beam/blob/ca41af8f/sdks/java/core/src/test/java/org/apache/beam/sdk/io/TFRecordIOTest.java ---------------------------------------------------------------------- diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/TFRecordIOTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/TFRecordIOTest.java index 9a9e840..d564d3b 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/TFRecordIOTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/TFRecordIOTest.java @@ -30,7 +30,6 @@ import static org.junit.Assert.assertThat; import com.google.common.collect.Lists; import com.google.common.io.BaseEncoding; import com.google.common.io.ByteStreams; - import java.io.File; import java.io.FileInputStream; import java.io.FileOutputStream; @@ -43,7 +42,6 @@ import java.nio.file.attribute.BasicFileAttributes; import java.util.Arrays; import java.util.Collections; import java.util.List; - import org.apache.beam.sdk.coders.StringUtf8Coder; import org.apache.beam.sdk.testing.NeedsRunner; import org.apache.beam.sdk.testing.PAssert; http://git-wip-us.apache.org/repos/asf/beam/blob/ca41af8f/sdks/java/core/src/test/java/org/apache/beam/sdk/metrics/GaugeCellTest.java ---------------------------------------------------------------------- diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/metrics/GaugeCellTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/metrics/GaugeCellTest.java index d8ef928..7dbfc4b 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/metrics/GaugeCellTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/metrics/GaugeCellTest.java @@ -19,6 +19,7 @@ package org.apache.beam.sdk.metrics; import static org.hamcrest.Matchers.equalTo; import static org.junit.Assert.assertThat; + import org.junit.Test; /** http://git-wip-us.apache.org/repos/asf/beam/blob/ca41af8f/sdks/java/core/src/test/java/org/apache/beam/sdk/options/PipelineOptionsTest.java ---------------------------------------------------------------------- diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/options/PipelineOptionsTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/options/PipelineOptionsTest.java index a13aa01..5e3211f 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/options/PipelineOptionsTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/options/PipelineOptionsTest.java @@ -23,7 +23,6 @@ import static org.junit.Assert.fail; import com.fasterxml.jackson.annotation.JsonIgnore; import com.google.common.collect.ImmutableMap; - import java.util.HashSet; import java.util.List; import java.util.Map; http://git-wip-us.apache.org/repos/asf/beam/blob/ca41af8f/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java ---------------------------------------------------------------------- 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 1d41923..ffdf3d0 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 @@ -70,6 +70,10 @@ 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.TimeDomain; +import org.apache.beam.sdk.state.Timer; +import org.apache.beam.sdk.state.TimerSpec; +import org.apache.beam.sdk.state.TimerSpecs; import org.apache.beam.sdk.state.ValueState; import org.apache.beam.sdk.testing.NeedsRunner; import org.apache.beam.sdk.testing.PAssert; @@ -94,10 +98,6 @@ import org.apache.beam.sdk.transforms.windowing.GlobalWindow; import org.apache.beam.sdk.transforms.windowing.IntervalWindow; import org.apache.beam.sdk.transforms.windowing.SlidingWindows; import org.apache.beam.sdk.transforms.windowing.Window; -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.TimerSpecs; import org.apache.beam.sdk.util.common.ElementByteSizeObserver; import org.apache.beam.sdk.values.KV; import org.apache.beam.sdk.values.PCollection; http://git-wip-us.apache.org/repos/asf/beam/blob/ca41af8f/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/RegexTest.java ---------------------------------------------------------------------- diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/RegexTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/RegexTest.java index ceebcca..969c4ea 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/RegexTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/RegexTest.java @@ -20,13 +20,11 @@ package org.apache.beam.sdk.transforms; import java.io.Serializable; import java.util.Arrays; import java.util.List; - import org.apache.beam.sdk.testing.NeedsRunner; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; import org.apache.beam.sdk.values.KV; import org.apache.beam.sdk.values.PCollection; - import org.junit.Rule; import org.junit.Test; import org.junit.experimental.categories.Category; http://git-wip-us.apache.org/repos/asf/beam/blob/ca41af8f/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnInvokersTest.java ---------------------------------------------------------------------- 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 13e46d5..19b6092 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 @@ -40,6 +40,10 @@ import org.apache.beam.sdk.coders.CustomCoder; import org.apache.beam.sdk.coders.VarIntCoder; import org.apache.beam.sdk.state.StateSpec; import org.apache.beam.sdk.state.StateSpecs; +import org.apache.beam.sdk.state.TimeDomain; +import org.apache.beam.sdk.state.Timer; +import org.apache.beam.sdk.state.TimerSpec; +import org.apache.beam.sdk.state.TimerSpecs; import org.apache.beam.sdk.state.ValueState; import org.apache.beam.sdk.transforms.DoFn; import org.apache.beam.sdk.transforms.reflect.DoFnInvoker.FakeArgumentProvider; @@ -48,10 +52,6 @@ import org.apache.beam.sdk.transforms.splittabledofn.HasDefaultTracker; import org.apache.beam.sdk.transforms.splittabledofn.RestrictionTracker; import org.apache.beam.sdk.transforms.windowing.BoundedWindow; import org.apache.beam.sdk.transforms.windowing.IntervalWindow; -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.TimerSpecs; import org.apache.beam.sdk.util.UserCodeException; import org.joda.time.Instant; import org.junit.Before; http://git-wip-us.apache.org/repos/asf/beam/blob/ca41af8f/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnSignaturesTest.java ---------------------------------------------------------------------- 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 27e0b89..5b97a55 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 @@ -31,6 +31,10 @@ import org.apache.beam.sdk.coders.VarIntCoder; import org.apache.beam.sdk.coders.VarLongCoder; import org.apache.beam.sdk.state.StateSpec; import org.apache.beam.sdk.state.StateSpecs; +import org.apache.beam.sdk.state.TimeDomain; +import org.apache.beam.sdk.state.Timer; +import org.apache.beam.sdk.state.TimerSpec; +import org.apache.beam.sdk.state.TimerSpecs; import org.apache.beam.sdk.state.ValueState; import org.apache.beam.sdk.state.WatermarkHoldState; import org.apache.beam.sdk.transforms.DoFn; @@ -41,10 +45,6 @@ import org.apache.beam.sdk.transforms.reflect.DoFnSignature.Parameter.TimerParam import org.apache.beam.sdk.transforms.reflect.DoFnSignature.Parameter.WindowParameter; import org.apache.beam.sdk.transforms.reflect.DoFnSignaturesTestUtils.FakeDoFn; import org.apache.beam.sdk.transforms.windowing.BoundedWindow; -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.TimerSpecs; import org.apache.beam.sdk.values.KV; import org.apache.beam.sdk.values.TypeDescriptor; import org.hamcrest.Matcher; http://git-wip-us.apache.org/repos/asf/beam/blob/ca41af8f/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/OnTimerInvokersTest.java ---------------------------------------------------------------------- diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/OnTimerInvokersTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/OnTimerInvokersTest.java index 177f15f..d317952 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/OnTimerInvokersTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/OnTimerInvokersTest.java @@ -22,11 +22,11 @@ import static org.hamcrest.Matchers.theInstance; import static org.junit.Assert.assertThat; import static org.mockito.Mockito.when; +import org.apache.beam.sdk.state.TimeDomain; +import org.apache.beam.sdk.state.TimerSpec; +import org.apache.beam.sdk.state.TimerSpecs; import org.apache.beam.sdk.transforms.DoFn; import org.apache.beam.sdk.transforms.windowing.BoundedWindow; -import org.apache.beam.sdk.util.TimeDomain; -import org.apache.beam.sdk.util.TimerSpec; -import org.apache.beam.sdk.util.TimerSpecs; import org.junit.Before; import org.junit.Rule; import org.junit.Test; http://git-wip-us.apache.org/repos/asf/beam/blob/ca41af8f/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/testhelper/DoFnInvokersTestHelper.java ---------------------------------------------------------------------- diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/testhelper/DoFnInvokersTestHelper.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/testhelper/DoFnInvokersTestHelper.java index 95e7c49..f9fcc58 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/testhelper/DoFnInvokersTestHelper.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/testhelper/DoFnInvokersTestHelper.java @@ -21,12 +21,12 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNull; import static org.mockito.Mockito.verify; +import org.apache.beam.sdk.state.TimeDomain; +import org.apache.beam.sdk.state.TimerSpec; +import org.apache.beam.sdk.state.TimerSpecs; import org.apache.beam.sdk.transforms.DoFn; import org.apache.beam.sdk.transforms.reflect.DoFnInvokersTest; import org.apache.beam.sdk.transforms.windowing.BoundedWindow; -import org.apache.beam.sdk.util.TimeDomain; -import org.apache.beam.sdk.util.TimerSpec; -import org.apache.beam.sdk.util.TimerSpecs; /** * Test helper for {@link DoFnInvokersTest}, which needs to test package-private access to DoFns in