You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by ke...@apache.org on 2016/11/12 02:28:48 UTC
[30/39] incubator-beam git commit: BEAM-261 Make translators package
private.
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/5553c603/runners/apex/src/test/java/org/apache/beam/runners/apex/translators/ParDoBoundTranslatorTest.java
----------------------------------------------------------------------
diff --git a/runners/apex/src/test/java/org/apache/beam/runners/apex/translators/ParDoBoundTranslatorTest.java b/runners/apex/src/test/java/org/apache/beam/runners/apex/translators/ParDoBoundTranslatorTest.java
deleted file mode 100644
index 2379a9e..0000000
--- a/runners/apex/src/test/java/org/apache/beam/runners/apex/translators/ParDoBoundTranslatorTest.java
+++ /dev/null
@@ -1,340 +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.runners.apex.translators;
-
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
-
-import com.datatorrent.api.DAG;
-import com.datatorrent.api.Sink;
-import com.datatorrent.lib.util.KryoCloneUtils;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Sets;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.List;
-import java.util.regex.Pattern;
-
-import org.apache.beam.runners.apex.ApexPipelineOptions;
-import org.apache.beam.runners.apex.ApexRunner;
-import org.apache.beam.runners.apex.ApexRunnerResult;
-import org.apache.beam.runners.apex.TestApexRunner;
-import org.apache.beam.runners.apex.translators.functions.ApexParDoOperator;
-import org.apache.beam.runners.apex.translators.io.ApexReadUnboundedInputOperator;
-import org.apache.beam.runners.apex.translators.utils.ApexStateInternals;
-import org.apache.beam.runners.apex.translators.utils.ApexStreamTuple;
-import org.apache.beam.sdk.Pipeline;
-import org.apache.beam.sdk.coders.Coder;
-import org.apache.beam.sdk.coders.SerializableCoder;
-import org.apache.beam.sdk.coders.VarIntCoder;
-import org.apache.beam.sdk.options.PipelineOptionsFactory;
-import org.apache.beam.sdk.testing.PAssert;
-import org.apache.beam.sdk.transforms.Create;
-import org.apache.beam.sdk.transforms.OldDoFn;
-import org.apache.beam.sdk.transforms.ParDo;
-import org.apache.beam.sdk.transforms.Sum;
-import org.apache.beam.sdk.transforms.View;
-import org.apache.beam.sdk.util.WindowedValue;
-import org.apache.beam.sdk.util.WindowingStrategy;
-import org.apache.beam.sdk.values.PCollection;
-import org.apache.beam.sdk.values.PCollectionTuple;
-import org.apache.beam.sdk.values.PCollectionView;
-import org.apache.beam.sdk.values.TupleTag;
-import org.apache.beam.sdk.values.TupleTagList;
-import org.junit.Assert;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.JUnit4;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * integration test for {@link ParDoBoundTranslator}.
- */
-@RunWith(JUnit4.class)
-public class ParDoBoundTranslatorTest {
- private static final Logger LOG = LoggerFactory.getLogger(ParDoBoundTranslatorTest.class);
- private static final long SLEEP_MILLIS = 500;
- private static final long TIMEOUT_MILLIS = 30000;
-
- @Test
- public void test() throws Exception {
- ApexPipelineOptions options = PipelineOptionsFactory.create()
- .as(ApexPipelineOptions.class);
- options.setApplicationName("ParDoBound");
- options.setRunner(ApexRunner.class);
-
- Pipeline p = Pipeline.create(options);
-
- List<Integer> collection = Lists.newArrayList(1, 2, 3, 4, 5);
- List<Integer> expected = Lists.newArrayList(6, 7, 8, 9, 10);
- p.apply(Create.of(collection).withCoder(SerializableCoder.of(Integer.class)))
- .apply(ParDo.of(new Add(5)))
- .apply(ParDo.of(new EmbeddedCollector()));
-
- ApexRunnerResult result = (ApexRunnerResult) p.run();
- DAG dag = result.getApexDAG();
-
- DAG.OperatorMeta om = dag.getOperatorMeta("Create.Values");
- Assert.assertNotNull(om);
- Assert.assertEquals(om.getOperator().getClass(), ApexReadUnboundedInputOperator.class);
-
- om = dag.getOperatorMeta("ParDo(Add)");
- Assert.assertNotNull(om);
- Assert.assertEquals(om.getOperator().getClass(), ApexParDoOperator.class);
-
- long timeout = System.currentTimeMillis() + TIMEOUT_MILLIS;
- while (System.currentTimeMillis() < timeout) {
- if (EmbeddedCollector.RESULTS.containsAll(expected)) {
- break;
- }
- LOG.info("Waiting for expected results.");
- Thread.sleep(SLEEP_MILLIS);
- }
- Assert.assertEquals(Sets.newHashSet(expected), EmbeddedCollector.RESULTS);
- }
-
- @SuppressWarnings("serial")
- private static class Add extends OldDoFn<Integer, Integer> {
- private Integer number;
- private PCollectionView<Integer> sideInputView;
-
- private Add(Integer number) {
- this.number = number;
- }
-
- private Add(PCollectionView<Integer> sideInputView) {
- this.sideInputView = sideInputView;
- }
-
- @Override
- public void processElement(ProcessContext c) throws Exception {
- if (sideInputView != null) {
- number = c.sideInput(sideInputView);
- }
- c.output(c.element() + number);
- }
- }
-
- private static class EmbeddedCollector extends OldDoFn<Object, Void> {
- private static final long serialVersionUID = 1L;
- protected static final HashSet<Object> RESULTS = new HashSet<>();
-
- public EmbeddedCollector() {
- RESULTS.clear();
- }
-
- @Override
- public void processElement(ProcessContext c) throws Exception {
- RESULTS.add(c.element());
- }
- }
-
- private static Throwable runExpectingAssertionFailure(Pipeline pipeline) {
- // We cannot use thrown.expect(AssertionError.class) because the AssertionError
- // is first caught by JUnit and causes a test failure.
- try {
- pipeline.run();
- } catch (AssertionError exc) {
- return exc;
- }
- fail("assertion should have failed");
- throw new RuntimeException("unreachable");
- }
-
- @Test
- public void testAssertionFailure() throws Exception {
- ApexPipelineOptions options = PipelineOptionsFactory.create()
- .as(ApexPipelineOptions.class);
- options.setRunner(TestApexRunner.class);
- Pipeline pipeline = Pipeline.create(options);
-
- PCollection<Integer> pcollection = pipeline
- .apply(Create.of(1, 2, 3, 4));
- PAssert.that(pcollection).containsInAnyOrder(2, 1, 4, 3, 7);
-
- Throwable exc = runExpectingAssertionFailure(pipeline);
- Pattern expectedPattern = Pattern.compile(
- "Expected: iterable over \\[((<4>|<7>|<3>|<2>|<1>)(, )?){5}\\] in any order");
- // A loose pattern, but should get the job done.
- assertTrue(
- "Expected error message from PAssert with substring matching "
- + expectedPattern
- + " but the message was \""
- + exc.getMessage()
- + "\"",
- expectedPattern.matcher(exc.getMessage()).find());
- }
-
- @Test
- public void testContainsInAnyOrder() throws Exception {
- ApexPipelineOptions options = PipelineOptionsFactory.create().as(ApexPipelineOptions.class);
- options.setRunner(TestApexRunner.class);
- Pipeline pipeline = Pipeline.create(options);
- PCollection<Integer> pcollection = pipeline.apply(Create.of(1, 2, 3, 4));
- PAssert.that(pcollection).containsInAnyOrder(2, 1, 4, 3);
- // TODO: terminate faster based on processed assertion vs. auto-shutdown
- pipeline.run();
- }
-
- @Test
- public void testSerialization() throws Exception {
- ApexPipelineOptions options = PipelineOptionsFactory.create()
- .as(ApexPipelineOptions.class);
- options.setRunner(TestApexRunner.class);
- Pipeline pipeline = Pipeline.create(options);
- Coder<WindowedValue<Integer>> coder = WindowedValue.getValueOnlyCoder(VarIntCoder.of());
-
- PCollectionView<Integer> singletonView = pipeline.apply(Create.of(1))
- .apply(Sum.integersGlobally().asSingletonView());
-
- ApexParDoOperator<Integer, Integer> operator = new ApexParDoOperator<>(options,
- new Add(singletonView), new TupleTag<Integer>(), TupleTagList.empty().getAll(),
- WindowingStrategy.globalDefault(),
- Collections.<PCollectionView<?>>singletonList(singletonView),
- coder,
- new ApexStateInternals.ApexStateInternalsFactory<Void>()
- );
- operator.setup(null);
- operator.beginWindow(0);
- WindowedValue<Integer> wv1 = WindowedValue.valueInGlobalWindow(1);
- WindowedValue<Iterable<?>> sideInput = WindowedValue.<Iterable<?>>valueInGlobalWindow(
- Lists.<Integer>newArrayList(22));
- operator.input.process(ApexStreamTuple.DataTuple.of(wv1)); // pushed back input
-
- final List<Object> results = Lists.newArrayList();
- Sink<Object> sink = new Sink<Object>() {
- @Override
- public void put(Object tuple) {
- results.add(tuple);
- }
- @Override
- public int getCount(boolean reset) {
- return 0;
- }
- };
-
- // verify pushed back input checkpointing
- Assert.assertNotNull("Serialization", operator = KryoCloneUtils.cloneObject(operator));
- operator.output.setSink(sink);
- operator.setup(null);
- operator.beginWindow(1);
- WindowedValue<Integer> wv2 = WindowedValue.valueInGlobalWindow(2);
- operator.sideInput1.process(ApexStreamTuple.DataTuple.of(sideInput));
- Assert.assertEquals("number outputs", 1, results.size());
- Assert.assertEquals("result", WindowedValue.valueInGlobalWindow(23),
- ((ApexStreamTuple.DataTuple) results.get(0)).getValue());
-
- // verify side input checkpointing
- results.clear();
- Assert.assertNotNull("Serialization", operator = KryoCloneUtils.cloneObject(operator));
- operator.output.setSink(sink);
- operator.setup(null);
- operator.beginWindow(2);
- operator.input.process(ApexStreamTuple.DataTuple.of(wv2));
- Assert.assertEquals("number outputs", 1, results.size());
- Assert.assertEquals("result", WindowedValue.valueInGlobalWindow(24),
- ((ApexStreamTuple.DataTuple) results.get(0)).getValue());
- }
-
- @Test
- public void testMultiOutputParDoWithSideInputs() throws Exception {
- ApexPipelineOptions options = PipelineOptionsFactory.create().as(ApexPipelineOptions.class);
- options.setRunner(ApexRunner.class); // non-blocking run
- Pipeline pipeline = Pipeline.create(options);
-
- List<Integer> inputs = Arrays.asList(3, -42, 666);
- final TupleTag<String> mainOutputTag = new TupleTag<>("main");
- final TupleTag<Void> sideOutputTag = new TupleTag<>("sideOutput");
-
- PCollectionView<Integer> sideInput1 = pipeline
- .apply("CreateSideInput1", Create.of(11))
- .apply("ViewSideInput1", View.<Integer>asSingleton());
- PCollectionView<Integer> sideInputUnread = pipeline
- .apply("CreateSideInputUnread", Create.of(-3333))
- .apply("ViewSideInputUnread", View.<Integer>asSingleton());
- PCollectionView<Integer> sideInput2 = pipeline
- .apply("CreateSideInput2", Create.of(222))
- .apply("ViewSideInput2", View.<Integer>asSingleton());
-
- PCollectionTuple outputs = pipeline
- .apply(Create.of(inputs))
- .apply(ParDo.withSideInputs(sideInput1)
- .withSideInputs(sideInputUnread)
- .withSideInputs(sideInput2)
- .withOutputTags(mainOutputTag, TupleTagList.of(sideOutputTag))
- .of(new TestMultiOutputWithSideInputsFn(
- Arrays.asList(sideInput1, sideInput2),
- Arrays.<TupleTag<String>>asList())));
-
- outputs.get(mainOutputTag).apply(ParDo.of(new EmbeddedCollector()));
- ApexRunnerResult result = (ApexRunnerResult) pipeline.run();
-
- HashSet<String> expected = Sets.newHashSet("processing: 3: [11, 222]",
- "processing: -42: [11, 222]", "processing: 666: [11, 222]");
- long timeout = System.currentTimeMillis() + TIMEOUT_MILLIS;
- while (System.currentTimeMillis() < timeout) {
- if (EmbeddedCollector.RESULTS.containsAll(expected)) {
- break;
- }
- LOG.info("Waiting for expected results.");
- Thread.sleep(SLEEP_MILLIS);
- }
- result.cancel();
- Assert.assertEquals(Sets.newHashSet(expected), EmbeddedCollector.RESULTS);
- }
-
- private static class TestMultiOutputWithSideInputsFn extends OldDoFn<Integer, String> {
- private static final long serialVersionUID = 1L;
-
- final List<PCollectionView<Integer>> sideInputViews = new ArrayList<>();
- final List<TupleTag<String>> sideOutputTupleTags = new ArrayList<>();
-
- public TestMultiOutputWithSideInputsFn(List<PCollectionView<Integer>> sideInputViews,
- List<TupleTag<String>> sideOutputTupleTags) {
- this.sideInputViews.addAll(sideInputViews);
- this.sideOutputTupleTags.addAll(sideOutputTupleTags);
- }
-
- @Override
- public void processElement(ProcessContext c) throws Exception {
- outputToAllWithSideInputs(c, "processing: " + c.element());
- }
-
- private void outputToAllWithSideInputs(ProcessContext c, String value) {
- if (!sideInputViews.isEmpty()) {
- List<Integer> sideInputValues = new ArrayList<>();
- for (PCollectionView<Integer> sideInputView : sideInputViews) {
- sideInputValues.add(c.sideInput(sideInputView));
- }
- value += ": " + sideInputValues;
- }
- c.output(value);
- for (TupleTag<String> sideOutputTupleTag : sideOutputTupleTags) {
- c.sideOutput(sideOutputTupleTag,
- sideOutputTupleTag.getId() + ": " + value);
- }
- }
-
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/5553c603/runners/apex/src/test/java/org/apache/beam/runners/apex/translators/ReadUnboundTranslatorTest.java
----------------------------------------------------------------------
diff --git a/runners/apex/src/test/java/org/apache/beam/runners/apex/translators/ReadUnboundTranslatorTest.java b/runners/apex/src/test/java/org/apache/beam/runners/apex/translators/ReadUnboundTranslatorTest.java
deleted file mode 100644
index 71c5354..0000000
--- a/runners/apex/src/test/java/org/apache/beam/runners/apex/translators/ReadUnboundTranslatorTest.java
+++ /dev/null
@@ -1,129 +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.runners.apex.translators;
-
-import com.datatorrent.api.DAG;
-import com.google.common.collect.ContiguousSet;
-import com.google.common.collect.DiscreteDomain;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Range;
-import com.google.common.collect.Sets;
-
-import java.util.HashSet;
-import java.util.List;
-import java.util.Set;
-
-import org.apache.beam.runners.apex.ApexPipelineOptions;
-import org.apache.beam.runners.apex.ApexRunner;
-import org.apache.beam.runners.apex.ApexRunnerResult;
-import org.apache.beam.runners.apex.translators.io.ApexReadUnboundedInputOperator;
-import org.apache.beam.runners.apex.translators.utils.CollectionSource;
-import org.apache.beam.sdk.Pipeline;
-import org.apache.beam.sdk.coders.StringUtf8Coder;
-import org.apache.beam.sdk.io.CountingSource;
-import org.apache.beam.sdk.io.Read;
-import org.apache.beam.sdk.options.PipelineOptionsFactory;
-import org.apache.beam.sdk.transforms.OldDoFn;
-import org.apache.beam.sdk.transforms.ParDo;
-import org.junit.Assert;
-import org.junit.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * integration test for {@link ReadUnboundedTranslator}.
- */
-public class ReadUnboundTranslatorTest {
- private static final Logger LOG = LoggerFactory.getLogger(ReadUnboundTranslatorTest.class);
-
- @Test
- public void test() throws Exception {
- ApexPipelineOptions options = PipelineOptionsFactory.create()
- .as(ApexPipelineOptions.class);
- EmbeddedCollector.RESULTS.clear();
- options.setApplicationName("ReadUnbound");
- options.setRunner(ApexRunner.class);
- Pipeline p = Pipeline.create(options);
-
- List<String> collection = Lists.newArrayList("1", "2", "3", "4", "5");
- CollectionSource<String> source = new CollectionSource<>(collection, StringUtf8Coder.of());
- p.apply(Read.from(source))
- .apply(ParDo.of(new EmbeddedCollector()));
-
- ApexRunnerResult result = (ApexRunnerResult) p.run();
- DAG dag = result.getApexDAG();
- DAG.OperatorMeta om = dag.getOperatorMeta("Read(CollectionSource)");
- Assert.assertNotNull(om);
- Assert.assertEquals(om.getOperator().getClass(), ApexReadUnboundedInputOperator.class);
-
- long timeout = System.currentTimeMillis() + 30000;
- while (System.currentTimeMillis() < timeout) {
- if (EmbeddedCollector.RESULTS.containsAll(collection)) {
- break;
- }
- LOG.info("Waiting for expected results.");
- Thread.sleep(1000);
- }
- Assert.assertEquals(Sets.newHashSet(collection), EmbeddedCollector.RESULTS);
- }
-
- @Test
- public void testReadBounded() throws Exception {
- ApexPipelineOptions options = PipelineOptionsFactory.create()
- .as(ApexPipelineOptions.class);
- EmbeddedCollector.RESULTS.clear();
- options.setApplicationName("ReadBounded");
- options.setRunner(ApexRunner.class);
- Pipeline p = Pipeline.create(options);
-
- Set<Long> expected = ContiguousSet.create(Range.closedOpen(0L, 10L), DiscreteDomain.longs());
- p.apply(Read.from(CountingSource.upTo(10)))
- .apply(ParDo.of(new EmbeddedCollector()));
-
- ApexRunnerResult result = (ApexRunnerResult) p.run();
- DAG dag = result.getApexDAG();
- DAG.OperatorMeta om = dag.getOperatorMeta("Read(BoundedCountingSource)");
- Assert.assertNotNull(om);
- Assert.assertEquals(om.getOperator().getClass(), ApexReadUnboundedInputOperator.class);
-
- long timeout = System.currentTimeMillis() + 30000;
- while (System.currentTimeMillis() < timeout) {
- if (EmbeddedCollector.RESULTS.containsAll(expected)) {
- break;
- }
- LOG.info("Waiting for expected results.");
- Thread.sleep(1000);
- }
- Assert.assertEquals(Sets.newHashSet(expected), EmbeddedCollector.RESULTS);
- }
-
- @SuppressWarnings("serial")
- private static class EmbeddedCollector extends OldDoFn<Object, Void> {
- protected static final HashSet<Object> RESULTS = new HashSet<>();
-
- public EmbeddedCollector() {
- }
-
- @Override
- public void processElement(ProcessContext c) throws Exception {
- RESULTS.add(c.element());
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/5553c603/runners/apex/src/test/java/org/apache/beam/runners/apex/translators/utils/ApexStateInternalsTest.java
----------------------------------------------------------------------
diff --git a/runners/apex/src/test/java/org/apache/beam/runners/apex/translators/utils/ApexStateInternalsTest.java b/runners/apex/src/test/java/org/apache/beam/runners/apex/translators/utils/ApexStateInternalsTest.java
deleted file mode 100644
index 055d98c..0000000
--- a/runners/apex/src/test/java/org/apache/beam/runners/apex/translators/utils/ApexStateInternalsTest.java
+++ /dev/null
@@ -1,361 +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.runners.apex.translators.utils;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertThat;
-
-import com.datatorrent.lib.util.KryoCloneUtils;
-
-import java.util.Arrays;
-
-import org.apache.beam.sdk.coders.StringUtf8Coder;
-import org.apache.beam.sdk.coders.VarIntCoder;
-import org.apache.beam.sdk.transforms.Sum;
-import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
-import org.apache.beam.sdk.transforms.windowing.IntervalWindow;
-import org.apache.beam.sdk.transforms.windowing.OutputTimeFns;
-import org.apache.beam.sdk.util.state.AccumulatorCombiningState;
-import org.apache.beam.sdk.util.state.BagState;
-import org.apache.beam.sdk.util.state.CombiningState;
-import org.apache.beam.sdk.util.state.ReadableState;
-import org.apache.beam.sdk.util.state.StateMerging;
-import org.apache.beam.sdk.util.state.StateNamespace;
-import org.apache.beam.sdk.util.state.StateNamespaceForTest;
-import org.apache.beam.sdk.util.state.StateTag;
-import org.apache.beam.sdk.util.state.StateTags;
-import org.apache.beam.sdk.util.state.ValueState;
-import org.apache.beam.sdk.util.state.WatermarkHoldState;
-import org.hamcrest.Matchers;
-import org.joda.time.Instant;
-import org.junit.Before;
-import org.junit.Test;
-
-/**
- * Tests for {@link ApexStateInternals}. This is based on the tests for
- * {@code InMemoryStateInternals}.
- */
-public class ApexStateInternalsTest {
- private static final BoundedWindow WINDOW_1 = new IntervalWindow(new Instant(0), new Instant(10));
- private static final StateNamespace NAMESPACE_1 = new StateNamespaceForTest("ns1");
- private static final StateNamespace NAMESPACE_2 = new StateNamespaceForTest("ns2");
- private static final StateNamespace NAMESPACE_3 = new StateNamespaceForTest("ns3");
-
- private static final StateTag<Object, ValueState<String>> STRING_VALUE_ADDR =
- StateTags.value("stringValue", StringUtf8Coder.of());
- private static final StateTag<Object, AccumulatorCombiningState<Integer, int[], Integer>>
- SUM_INTEGER_ADDR = StateTags.combiningValueFromInputInternal(
- "sumInteger", VarIntCoder.of(), new Sum.SumIntegerFn());
- private static final StateTag<Object, BagState<String>> STRING_BAG_ADDR =
- StateTags.bag("stringBag", StringUtf8Coder.of());
- private static final StateTag<Object, WatermarkHoldState<BoundedWindow>>
- WATERMARK_EARLIEST_ADDR =
- StateTags.watermarkStateInternal("watermark", OutputTimeFns.outputAtEarliestInputTimestamp());
- private static final StateTag<Object, WatermarkHoldState<BoundedWindow>>
- WATERMARK_LATEST_ADDR =
- StateTags.watermarkStateInternal("watermark", OutputTimeFns.outputAtLatestInputTimestamp());
- private static final StateTag<Object, WatermarkHoldState<BoundedWindow>> WATERMARK_EOW_ADDR =
- StateTags.watermarkStateInternal("watermark", OutputTimeFns.outputAtEndOfWindow());
-
- private ApexStateInternals<String> underTest;
-
- @Before
- public void initStateInternals() {
- underTest = new ApexStateInternals<>(null);
- }
-
- @Test
- public void testBag() throws Exception {
- BagState<String> value = underTest.state(NAMESPACE_1, STRING_BAG_ADDR);
-
- assertEquals(value, underTest.state(NAMESPACE_1, STRING_BAG_ADDR));
- assertFalse(value.equals(underTest.state(NAMESPACE_2, STRING_BAG_ADDR)));
-
- assertThat(value.read(), Matchers.emptyIterable());
- value.add("hello");
- assertThat(value.read(), Matchers.containsInAnyOrder("hello"));
-
- value.add("world");
- assertThat(value.read(), Matchers.containsInAnyOrder("hello", "world"));
-
- value.clear();
- assertThat(value.read(), Matchers.emptyIterable());
- assertEquals(underTest.state(NAMESPACE_1, STRING_BAG_ADDR), value);
-
- }
-
- @Test
- public void testBagIsEmpty() throws Exception {
- BagState<String> value = underTest.state(NAMESPACE_1, STRING_BAG_ADDR);
-
- assertThat(value.isEmpty().read(), Matchers.is(true));
- ReadableState<Boolean> readFuture = value.isEmpty();
- value.add("hello");
- assertThat(readFuture.read(), Matchers.is(false));
-
- value.clear();
- assertThat(readFuture.read(), Matchers.is(true));
- }
-
- @Test
- public void testMergeBagIntoSource() throws Exception {
- BagState<String> bag1 = underTest.state(NAMESPACE_1, STRING_BAG_ADDR);
- BagState<String> bag2 = underTest.state(NAMESPACE_2, STRING_BAG_ADDR);
-
- bag1.add("Hello");
- bag2.add("World");
- bag1.add("!");
-
- StateMerging.mergeBags(Arrays.asList(bag1, bag2), bag1);
-
- // Reading the merged bag gets both the contents
- assertThat(bag1.read(), Matchers.containsInAnyOrder("Hello", "World", "!"));
- assertThat(bag2.read(), Matchers.emptyIterable());
- }
-
- @Test
- public void testMergeBagIntoNewNamespace() throws Exception {
- BagState<String> bag1 = underTest.state(NAMESPACE_1, STRING_BAG_ADDR);
- BagState<String> bag2 = underTest.state(NAMESPACE_2, STRING_BAG_ADDR);
- BagState<String> bag3 = underTest.state(NAMESPACE_3, STRING_BAG_ADDR);
-
- bag1.add("Hello");
- bag2.add("World");
- bag1.add("!");
-
- StateMerging.mergeBags(Arrays.asList(bag1, bag2, bag3), bag3);
-
- // Reading the merged bag gets both the contents
- assertThat(bag3.read(), Matchers.containsInAnyOrder("Hello", "World", "!"));
- assertThat(bag1.read(), Matchers.emptyIterable());
- assertThat(bag2.read(), Matchers.emptyIterable());
- }
-
- @Test
- public void testCombiningValue() throws Exception {
- CombiningState<Integer, Integer> value = underTest.state(NAMESPACE_1, SUM_INTEGER_ADDR);
-
- // State instances are cached, but depend on the namespace.
- assertEquals(value, underTest.state(NAMESPACE_1, SUM_INTEGER_ADDR));
- assertFalse(value.equals(underTest.state(NAMESPACE_2, SUM_INTEGER_ADDR)));
-
- assertThat(value.read(), Matchers.equalTo(0));
- value.add(2);
- assertThat(value.read(), Matchers.equalTo(2));
-
- value.add(3);
- assertThat(value.read(), Matchers.equalTo(5));
-
- value.clear();
- assertThat(value.read(), Matchers.equalTo(0));
- assertEquals(underTest.state(NAMESPACE_1, SUM_INTEGER_ADDR), value);
- }
-
- @Test
- public void testCombiningIsEmpty() throws Exception {
- CombiningState<Integer, Integer> value = underTest.state(NAMESPACE_1, SUM_INTEGER_ADDR);
-
- assertThat(value.isEmpty().read(), Matchers.is(true));
- ReadableState<Boolean> readFuture = value.isEmpty();
- value.add(5);
- assertThat(readFuture.read(), Matchers.is(false));
-
- value.clear();
- assertThat(readFuture.read(), Matchers.is(true));
- }
-
- @Test
- public void testMergeCombiningValueIntoSource() throws Exception {
- AccumulatorCombiningState<Integer, int[], Integer> value1 =
- underTest.state(NAMESPACE_1, SUM_INTEGER_ADDR);
- AccumulatorCombiningState<Integer, int[], Integer> value2 =
- underTest.state(NAMESPACE_2, SUM_INTEGER_ADDR);
-
- value1.add(5);
- value2.add(10);
- value1.add(6);
-
- assertThat(value1.read(), Matchers.equalTo(11));
- assertThat(value2.read(), Matchers.equalTo(10));
-
- // Merging clears the old values and updates the result value.
- StateMerging.mergeCombiningValues(Arrays.asList(value1, value2), value1);
-
- assertThat(value1.read(), Matchers.equalTo(21));
- assertThat(value2.read(), Matchers.equalTo(0));
- }
-
- @Test
- public void testMergeCombiningValueIntoNewNamespace() throws Exception {
- AccumulatorCombiningState<Integer, int[], Integer> value1 =
- underTest.state(NAMESPACE_1, SUM_INTEGER_ADDR);
- AccumulatorCombiningState<Integer, int[], Integer> value2 =
- underTest.state(NAMESPACE_2, SUM_INTEGER_ADDR);
- AccumulatorCombiningState<Integer, int[], Integer> value3 =
- underTest.state(NAMESPACE_3, SUM_INTEGER_ADDR);
-
- value1.add(5);
- value2.add(10);
- value1.add(6);
-
- StateMerging.mergeCombiningValues(Arrays.asList(value1, value2), value3);
-
- // Merging clears the old values and updates the result value.
- assertThat(value1.read(), Matchers.equalTo(0));
- assertThat(value2.read(), Matchers.equalTo(0));
- assertThat(value3.read(), Matchers.equalTo(21));
- }
-
- @Test
- public void testWatermarkEarliestState() throws Exception {
- WatermarkHoldState<BoundedWindow> value =
- underTest.state(NAMESPACE_1, WATERMARK_EARLIEST_ADDR);
-
- // State instances are cached, but depend on the namespace.
- assertEquals(value, underTest.state(NAMESPACE_1, WATERMARK_EARLIEST_ADDR));
- assertFalse(value.equals(underTest.state(NAMESPACE_2, WATERMARK_EARLIEST_ADDR)));
-
- assertThat(value.read(), Matchers.nullValue());
- value.add(new Instant(2000));
- assertThat(value.read(), Matchers.equalTo(new Instant(2000)));
-
- value.add(new Instant(3000));
- assertThat(value.read(), Matchers.equalTo(new Instant(2000)));
-
- value.add(new Instant(1000));
- assertThat(value.read(), Matchers.equalTo(new Instant(1000)));
-
- value.clear();
- assertThat(value.read(), Matchers.equalTo(null));
- assertEquals(underTest.state(NAMESPACE_1, WATERMARK_EARLIEST_ADDR), value);
- }
-
- @Test
- public void testWatermarkLatestState() throws Exception {
- WatermarkHoldState<BoundedWindow> value =
- underTest.state(NAMESPACE_1, WATERMARK_LATEST_ADDR);
-
- // State instances are cached, but depend on the namespace.
- assertEquals(value, underTest.state(NAMESPACE_1, WATERMARK_LATEST_ADDR));
- assertFalse(value.equals(underTest.state(NAMESPACE_2, WATERMARK_LATEST_ADDR)));
-
- assertThat(value.read(), Matchers.nullValue());
- value.add(new Instant(2000));
- assertThat(value.read(), Matchers.equalTo(new Instant(2000)));
-
- value.add(new Instant(3000));
- assertThat(value.read(), Matchers.equalTo(new Instant(3000)));
-
- value.add(new Instant(1000));
- assertThat(value.read(), Matchers.equalTo(new Instant(3000)));
-
- value.clear();
- assertThat(value.read(), Matchers.equalTo(null));
- assertEquals(underTest.state(NAMESPACE_1, WATERMARK_LATEST_ADDR), value);
- }
-
- @Test
- public void testWatermarkEndOfWindowState() throws Exception {
- WatermarkHoldState<BoundedWindow> value = underTest.state(NAMESPACE_1, WATERMARK_EOW_ADDR);
-
- // State instances are cached, but depend on the namespace.
- assertEquals(value, underTest.state(NAMESPACE_1, WATERMARK_EOW_ADDR));
- assertFalse(value.equals(underTest.state(NAMESPACE_2, WATERMARK_EOW_ADDR)));
-
- assertThat(value.read(), Matchers.nullValue());
- value.add(new Instant(2000));
- assertThat(value.read(), Matchers.equalTo(new Instant(2000)));
-
- value.clear();
- assertThat(value.read(), Matchers.equalTo(null));
- assertEquals(underTest.state(NAMESPACE_1, WATERMARK_EOW_ADDR), value);
- }
-
- @Test
- public void testWatermarkStateIsEmpty() throws Exception {
- WatermarkHoldState<BoundedWindow> value =
- underTest.state(NAMESPACE_1, WATERMARK_EARLIEST_ADDR);
-
- assertThat(value.isEmpty().read(), Matchers.is(true));
- ReadableState<Boolean> readFuture = value.isEmpty();
- value.add(new Instant(1000));
- assertThat(readFuture.read(), Matchers.is(false));
-
- value.clear();
- assertThat(readFuture.read(), Matchers.is(true));
- }
-
- @Test
- public void testMergeEarliestWatermarkIntoSource() throws Exception {
- WatermarkHoldState<BoundedWindow> value1 =
- underTest.state(NAMESPACE_1, WATERMARK_EARLIEST_ADDR);
- WatermarkHoldState<BoundedWindow> value2 =
- underTest.state(NAMESPACE_2, WATERMARK_EARLIEST_ADDR);
-
- value1.add(new Instant(3000));
- value2.add(new Instant(5000));
- value1.add(new Instant(4000));
- value2.add(new Instant(2000));
-
- // Merging clears the old values and updates the merged value.
- StateMerging.mergeWatermarks(Arrays.asList(value1, value2), value1, WINDOW_1);
-
- assertThat(value1.read(), Matchers.equalTo(new Instant(2000)));
- assertThat(value2.read(), Matchers.equalTo(null));
- }
-
- @Test
- public void testMergeLatestWatermarkIntoSource() throws Exception {
- WatermarkHoldState<BoundedWindow> value1 =
- underTest.state(NAMESPACE_1, WATERMARK_LATEST_ADDR);
- WatermarkHoldState<BoundedWindow> value2 =
- underTest.state(NAMESPACE_2, WATERMARK_LATEST_ADDR);
- WatermarkHoldState<BoundedWindow> value3 =
- underTest.state(NAMESPACE_3, WATERMARK_LATEST_ADDR);
-
- value1.add(new Instant(3000));
- value2.add(new Instant(5000));
- value1.add(new Instant(4000));
- value2.add(new Instant(2000));
-
- // Merging clears the old values and updates the result value.
- StateMerging.mergeWatermarks(Arrays.asList(value1, value2), value3, WINDOW_1);
-
- // Merging clears the old values and updates the result value.
- assertThat(value3.read(), Matchers.equalTo(new Instant(5000)));
- assertThat(value1.read(), Matchers.equalTo(null));
- assertThat(value2.read(), Matchers.equalTo(null));
- }
-
- @Test
- public void testSerialization() throws Exception {
- ApexStateInternals<String> original = new ApexStateInternals<String>(null);
- ValueState<String> value = original.state(NAMESPACE_1, STRING_VALUE_ADDR);
- assertEquals(original.state(NAMESPACE_1, STRING_VALUE_ADDR), value);
- value.write("hello");
-
- ApexStateInternals<String> cloned;
- assertNotNull("Serialization", cloned = KryoCloneUtils.cloneObject(original));
- ValueState<String> clonedValue = cloned.state(NAMESPACE_1, STRING_VALUE_ADDR);
- assertThat(clonedValue.read(), Matchers.equalTo("hello"));
- assertEquals(cloned.state(NAMESPACE_1, STRING_VALUE_ADDR), value);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/5553c603/runners/apex/src/test/java/org/apache/beam/runners/apex/translators/utils/CollectionSource.java
----------------------------------------------------------------------
diff --git a/runners/apex/src/test/java/org/apache/beam/runners/apex/translators/utils/CollectionSource.java b/runners/apex/src/test/java/org/apache/beam/runners/apex/translators/utils/CollectionSource.java
deleted file mode 100644
index c368bb2..0000000
--- a/runners/apex/src/test/java/org/apache/beam/runners/apex/translators/utils/CollectionSource.java
+++ /dev/null
@@ -1,136 +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.runners.apex.translators.utils;
-
-import java.io.IOException;
-import java.io.Serializable;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.Iterator;
-import java.util.List;
-import java.util.NoSuchElementException;
-
-import javax.annotation.Nullable;
-
-import org.apache.beam.sdk.coders.Coder;
-import org.apache.beam.sdk.io.UnboundedSource;
-import org.apache.beam.sdk.options.PipelineOptions;
-import org.joda.time.Instant;
-
-/**
- * collection as {@link UnboundedSource}, used for tests.
- */
-public class CollectionSource<T> extends UnboundedSource<T, UnboundedSource.CheckpointMark> {
- private static final long serialVersionUID = 1L;
- private final Collection<T> collection;
- private final Coder<T> coder;
-
- public CollectionSource(Collection<T> collection, Coder<T> coder) {
- this.collection = collection;
- this.coder = coder;
- }
-
- @Override
- public List<? extends UnboundedSource<T, CheckpointMark>> generateInitialSplits(
- int desiredNumSplits, PipelineOptions options) throws Exception {
- return Collections.singletonList(this);
- }
-
- @Override
- public UnboundedReader<T> createReader(PipelineOptions options,
- @Nullable UnboundedSource.CheckpointMark checkpointMark) {
- return new CollectionReader<>(collection, this);
- }
-
- @Nullable
- @Override
- public Coder<CheckpointMark> getCheckpointMarkCoder() {
- return null;
- }
-
- @Override
- public void validate() {
- }
-
- @Override
- public Coder<T> getDefaultOutputCoder() {
- return coder;
- }
-
- private static class CollectionReader<T> extends UnboundedSource.UnboundedReader<T>
- implements Serializable {
-
- private T current;
- private final CollectionSource<T> source;
- private final Collection<T> collection;
- private Iterator<T> iterator;
-
- public CollectionReader(Collection<T> collection, CollectionSource<T> source) {
- this.collection = collection;
- this.source = source;
- }
-
- @Override
- public boolean start() throws IOException {
- if (null == iterator) {
- iterator = collection.iterator();
- }
- return advance();
- }
-
- @Override
- public boolean advance() throws IOException {
- if (iterator.hasNext()) {
- current = iterator.next();
- return true;
- } else {
- return false;
- }
- }
-
- @Override
- public Instant getWatermark() {
- return Instant.now();
- }
-
- @Override
- public UnboundedSource.CheckpointMark getCheckpointMark() {
- return null;
- }
-
- @Override
- public UnboundedSource<T, ?> getCurrentSource() {
- return source;
- }
-
- @Override
- public T getCurrent() throws NoSuchElementException {
- return current;
- }
-
- @Override
- public Instant getCurrentTimestamp() throws NoSuchElementException {
- return Instant.now();
- }
-
- @Override
- public void close() throws IOException {
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/5553c603/runners/apex/src/test/java/org/apache/beam/runners/apex/translators/utils/PipelineOptionsTest.java
----------------------------------------------------------------------
diff --git a/runners/apex/src/test/java/org/apache/beam/runners/apex/translators/utils/PipelineOptionsTest.java b/runners/apex/src/test/java/org/apache/beam/runners/apex/translators/utils/PipelineOptionsTest.java
deleted file mode 100644
index e67efa9..0000000
--- a/runners/apex/src/test/java/org/apache/beam/runners/apex/translators/utils/PipelineOptionsTest.java
+++ /dev/null
@@ -1,84 +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.runners.apex.translators.utils;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-
-import com.datatorrent.common.util.FSStorageAgent;
-import com.esotericsoftware.kryo.serializers.FieldSerializer.Bind;
-import com.esotericsoftware.kryo.serializers.JavaSerializer;
-
-import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
-
-import org.apache.beam.runners.apex.ApexPipelineOptions;
-import org.apache.beam.sdk.options.Default;
-import org.apache.beam.sdk.options.Description;
-import org.apache.beam.sdk.options.PipelineOptionsFactory;
-import org.junit.BeforeClass;
-import org.junit.Test;
-
-/**
- * Tests the serialization of PipelineOptions.
- */
-public class PipelineOptionsTest {
-
- /**
- * Interface for testing.
- */
- public interface MyOptions extends ApexPipelineOptions {
- @Description("Bla bla bla")
- @Default.String("Hello")
- String getTestOption();
- void setTestOption(String value);
- }
-
- private static class MyOptionsWrapper {
- private MyOptionsWrapper() {
- this(null); // required for Kryo
- }
- private MyOptionsWrapper(ApexPipelineOptions options) {
- this.options = new SerializablePipelineOptions(options);
- }
- @Bind(JavaSerializer.class)
- private final SerializablePipelineOptions options;
- }
-
- private static MyOptions options;
-
- private static final String[] args = new String[]{"--testOption=nothing"};
-
- @BeforeClass
- public static void beforeTest() {
- options = PipelineOptionsFactory.fromArgs(args).as(MyOptions.class);
- }
-
- @Test
- public void testSerialization() {
- MyOptionsWrapper wrapper = new MyOptionsWrapper(PipelineOptionsTest.options);
- ByteArrayOutputStream bos = new ByteArrayOutputStream();
- FSStorageAgent.store(bos, wrapper);
-
- ByteArrayInputStream bis = new ByteArrayInputStream(bos.toByteArray());
- MyOptionsWrapper wrapperCopy = (MyOptionsWrapper) FSStorageAgent.retrieve(bis);
- assertNotNull(wrapperCopy.options);
- assertEquals("nothing", wrapperCopy.options.get().as(MyOptions.class).getTestOption());
- }
-
-}