You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by tg...@apache.org on 2017/03/17 15:54:32 UTC
[4/4] beam git commit: Migrate ForwardingPTransform to
core-construction
Migrate ForwardingPTransform to core-construction
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/45feb34a
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/45feb34a
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/45feb34a
Branch: refs/heads/master
Commit: 45feb34a154cfac9fd54e18439bee6bf1be7de1b
Parents: 969acb7
Author: Thomas Groh <tg...@google.com>
Authored: Tue Mar 14 12:39:57 2017 -0700
Committer: Thomas Groh <tg...@google.com>
Committed: Fri Mar 17 08:54:02 2017 -0700
----------------------------------------------------------------------
runners/core-construction-java/pom.xml | 7 ++
.../core/construction/ForwardingPTransform.java | 62 ++++++++++
.../construction/ForwardingPTransformTest.java | 111 ++++++++++++++++++
.../beam/runners/direct/DirectGroupByKey.java | 1 +
.../runners/direct/ForwardingPTransform.java | 62 ----------
.../runners/direct/ViewEvaluatorFactory.java | 1 +
.../direct/ForwardingPTransformTest.java | 112 -------------------
7 files changed, 182 insertions(+), 174 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/45feb34a/runners/core-construction-java/pom.xml
----------------------------------------------------------------------
diff --git a/runners/core-construction-java/pom.xml b/runners/core-construction-java/pom.xml
index 50843ed..9619280 100644
--- a/runners/core-construction-java/pom.xml
+++ b/runners/core-construction-java/pom.xml
@@ -82,5 +82,12 @@
<artifactId>junit</artifactId>
<scope>test</scope>
</dependency>
+
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-all</artifactId>
+ <scope>test</scope>
+ </dependency>
+
</dependencies>
</project>
http://git-wip-us.apache.org/repos/asf/beam/blob/45feb34a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ForwardingPTransform.java
----------------------------------------------------------------------
diff --git a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ForwardingPTransform.java b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ForwardingPTransform.java
new file mode 100644
index 0000000..3bee281
--- /dev/null
+++ b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ForwardingPTransform.java
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.core.construction;
+
+import org.apache.beam.sdk.coders.CannotProvideCoderException;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.display.DisplayData;
+import org.apache.beam.sdk.values.PInput;
+import org.apache.beam.sdk.values.POutput;
+import org.apache.beam.sdk.values.TypedPValue;
+
+/**
+ * A base class for implementing {@link PTransform} overrides, which behave identically to the
+ * delegate transform but with overridden methods. Implementors are required to implement
+ * {@link #delegate()}, which returns the object to forward calls to, and {@link #expand(PInput)}.
+ */
+public abstract class ForwardingPTransform<InputT extends PInput, OutputT extends POutput>
+ extends PTransform<InputT, OutputT> {
+ protected abstract PTransform<InputT, OutputT> delegate();
+
+ @Override
+ public OutputT expand(InputT input) {
+ return delegate().expand(input);
+ }
+
+ @Override
+ public void validate(InputT input) {
+ delegate().validate(input);
+ }
+
+ @Override
+ public String getName() {
+ return delegate().getName();
+ }
+
+ @Override
+ public <T> Coder<T> getDefaultOutputCoder(InputT input, @SuppressWarnings("unused")
+ TypedPValue<T> output) throws CannotProvideCoderException {
+ return delegate().getDefaultOutputCoder(input, output);
+ }
+
+ @Override
+ public void populateDisplayData(DisplayData.Builder builder) {
+ builder.delegate(delegate());
+ }
+}
http://git-wip-us.apache.org/repos/asf/beam/blob/45feb34a/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/ForwardingPTransformTest.java
----------------------------------------------------------------------
diff --git a/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/ForwardingPTransformTest.java b/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/ForwardingPTransformTest.java
new file mode 100644
index 0000000..7d3bfd8
--- /dev/null
+++ b/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/ForwardingPTransformTest.java
@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.core.construction;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.junit.Assert.assertThat;
+import static org.mockito.Matchers.any;
+
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.display.DisplayData;
+import org.apache.beam.sdk.values.PCollection;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.MockitoAnnotations;
+
+/**
+ * Tests for {@link ForwardingPTransform}.
+ */
+@RunWith(JUnit4.class)
+public class ForwardingPTransformTest {
+ @Rule public ExpectedException thrown = ExpectedException.none();
+
+ @Mock private PTransform<PCollection<Integer>, PCollection<String>> delegate;
+
+ private ForwardingPTransform<PCollection<Integer>, PCollection<String>> forwarding;
+
+ @Before
+ public void setup() {
+ MockitoAnnotations.initMocks(this);
+ forwarding =
+ new ForwardingPTransform<PCollection<Integer>, PCollection<String>>() {
+ @Override
+ protected PTransform<PCollection<Integer>, PCollection<String>> delegate() {
+ return delegate;
+ }
+ };
+ }
+
+ @Test
+ public void applyDelegates() {
+ @SuppressWarnings("unchecked")
+ PCollection<Integer> collection = Mockito.mock(PCollection.class);
+ @SuppressWarnings("unchecked")
+ PCollection<String> output = Mockito.mock(PCollection.class);
+ Mockito.when(delegate.expand(collection)).thenReturn(output);
+ PCollection<String> result = forwarding.expand(collection);
+ assertThat(result, equalTo(output));
+ }
+
+ @Test
+ public void getNameDelegates() {
+ String name = "My_forwardingptransform-name;for!thisTest";
+ Mockito.when(delegate.getName()).thenReturn(name);
+ assertThat(forwarding.getName(), equalTo(name));
+ }
+
+ @Test
+ public void validateDelegates() {
+ @SuppressWarnings("unchecked")
+ PCollection<Integer> input = Mockito.mock(PCollection.class);
+ Mockito.doThrow(RuntimeException.class).when(delegate).validate(input);
+
+ thrown.expect(RuntimeException.class);
+ forwarding.validate(input);
+ }
+
+ @Test
+ public void getDefaultOutputCoderDelegates() throws Exception {
+ @SuppressWarnings("unchecked")
+ PCollection<Integer> input = Mockito.mock(PCollection.class);
+ @SuppressWarnings("unchecked")
+ PCollection<String> output = Mockito.mock(PCollection.class);
+ @SuppressWarnings("unchecked")
+ Coder<String> outputCoder = Mockito.mock(Coder.class);
+
+ Mockito.when(delegate.getDefaultOutputCoder(input, output)).thenReturn(outputCoder);
+ assertThat(forwarding.getDefaultOutputCoder(input, output), equalTo(outputCoder));
+ }
+
+ @Test
+ public void populateDisplayDataDelegates() {
+ Mockito.doThrow(RuntimeException.class)
+ .when(delegate)
+ .populateDisplayData(any(DisplayData.Builder.class));
+
+ thrown.expect(RuntimeException.class);
+ DisplayData.from(forwarding);
+ }
+}
http://git-wip-us.apache.org/repos/asf/beam/blob/45feb34a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectGroupByKey.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectGroupByKey.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectGroupByKey.java
index 6c10bd2..3605002 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectGroupByKey.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectGroupByKey.java
@@ -22,6 +22,7 @@ import static com.google.common.base.Preconditions.checkArgument;
import org.apache.beam.runners.core.KeyedWorkItem;
import org.apache.beam.runners.core.KeyedWorkItemCoder;
+import org.apache.beam.runners.core.construction.ForwardingPTransform;
import org.apache.beam.sdk.coders.CannotProvideCoderException;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.coders.IterableCoder;
http://git-wip-us.apache.org/repos/asf/beam/blob/45feb34a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ForwardingPTransform.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ForwardingPTransform.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ForwardingPTransform.java
deleted file mode 100644
index 97c0983..0000000
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ForwardingPTransform.java
+++ /dev/null
@@ -1,62 +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.direct;
-
-import org.apache.beam.sdk.coders.CannotProvideCoderException;
-import org.apache.beam.sdk.coders.Coder;
-import org.apache.beam.sdk.transforms.PTransform;
-import org.apache.beam.sdk.transforms.display.DisplayData;
-import org.apache.beam.sdk.values.PInput;
-import org.apache.beam.sdk.values.POutput;
-import org.apache.beam.sdk.values.TypedPValue;
-
-/**
- * A base class for implementing {@link PTransform} overrides, which behave identically to the
- * delegate transform but with overridden methods. Implementors are required to implement
- * {@link #delegate()}, which returns the object to forward calls to, and {@link #expand(PInput)}.
- */
-public abstract class ForwardingPTransform<InputT extends PInput, OutputT extends POutput>
- extends PTransform<InputT, OutputT> {
- protected abstract PTransform<InputT, OutputT> delegate();
-
- @Override
- public OutputT expand(InputT input) {
- return delegate().expand(input);
- }
-
- @Override
- public void validate(InputT input) {
- delegate().validate(input);
- }
-
- @Override
- public String getName() {
- return delegate().getName();
- }
-
- @Override
- public <T> Coder<T> getDefaultOutputCoder(InputT input, @SuppressWarnings("unused")
- TypedPValue<T> output) throws CannotProvideCoderException {
- return delegate().getDefaultOutputCoder(input, output);
- }
-
- @Override
- public void populateDisplayData(DisplayData.Builder builder) {
- builder.delegate(delegate());
- }
-}
http://git-wip-us.apache.org/repos/asf/beam/blob/45feb34a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ViewEvaluatorFactory.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ViewEvaluatorFactory.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ViewEvaluatorFactory.java
index 1548772..9dcbf9e 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ViewEvaluatorFactory.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ViewEvaluatorFactory.java
@@ -20,6 +20,7 @@ package org.apache.beam.runners.direct;
import com.google.common.collect.Iterables;
import java.util.ArrayList;
import java.util.List;
+import org.apache.beam.runners.core.construction.ForwardingPTransform;
import org.apache.beam.runners.core.construction.SingleInputOutputOverrideFactory;
import org.apache.beam.runners.direct.CommittedResult.OutputType;
import org.apache.beam.runners.direct.DirectRunner.PCollectionViewWriter;
http://git-wip-us.apache.org/repos/asf/beam/blob/45feb34a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ForwardingPTransformTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ForwardingPTransformTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ForwardingPTransformTest.java
deleted file mode 100644
index 6860a58..0000000
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ForwardingPTransformTest.java
+++ /dev/null
@@ -1,112 +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.direct;
-
-import static org.hamcrest.Matchers.equalTo;
-import static org.junit.Assert.assertThat;
-import static org.mockito.Matchers.any;
-import static org.mockito.Mockito.doThrow;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
-import org.apache.beam.sdk.coders.Coder;
-import org.apache.beam.sdk.transforms.PTransform;
-import org.apache.beam.sdk.transforms.display.DisplayData;
-import org.apache.beam.sdk.values.PCollection;
-import org.junit.Before;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.ExpectedException;
-import org.junit.runner.RunWith;
-import org.junit.runners.JUnit4;
-import org.mockito.Mock;
-import org.mockito.MockitoAnnotations;
-
-/**
- * Tests for {@link ForwardingPTransform}.
- */
-@RunWith(JUnit4.class)
-public class ForwardingPTransformTest {
- @Rule public ExpectedException thrown = ExpectedException.none();
-
- @Mock private PTransform<PCollection<Integer>, PCollection<String>> delegate;
-
- private ForwardingPTransform<PCollection<Integer>, PCollection<String>> forwarding;
-
- @Before
- public void setup() {
- MockitoAnnotations.initMocks(this);
- forwarding =
- new ForwardingPTransform<PCollection<Integer>, PCollection<String>>() {
- @Override
- protected PTransform<PCollection<Integer>, PCollection<String>> delegate() {
- return delegate;
- }
- };
- }
-
- @Test
- public void applyDelegates() {
- @SuppressWarnings("unchecked")
- PCollection<Integer> collection = mock(PCollection.class);
- @SuppressWarnings("unchecked")
- PCollection<String> output = mock(PCollection.class);
- when(delegate.expand(collection)).thenReturn(output);
- PCollection<String> result = forwarding.expand(collection);
- assertThat(result, equalTo(output));
- }
-
- @Test
- public void getNameDelegates() {
- String name = "My_forwardingptransform-name;for!thisTest";
- when(delegate.getName()).thenReturn(name);
- assertThat(forwarding.getName(), equalTo(name));
- }
-
- @Test
- public void validateDelegates() {
- @SuppressWarnings("unchecked")
- PCollection<Integer> input = mock(PCollection.class);
- doThrow(RuntimeException.class).when(delegate).validate(input);
-
- thrown.expect(RuntimeException.class);
- forwarding.validate(input);
- }
-
- @Test
- public void getDefaultOutputCoderDelegates() throws Exception {
- @SuppressWarnings("unchecked")
- PCollection<Integer> input = mock(PCollection.class);
- @SuppressWarnings("unchecked")
- PCollection<String> output = mock(PCollection.class);
- @SuppressWarnings("unchecked")
- Coder<String> outputCoder = mock(Coder.class);
-
- when(delegate.getDefaultOutputCoder(input, output)).thenReturn(outputCoder);
- assertThat(forwarding.getDefaultOutputCoder(input, output), equalTo(outputCoder));
- }
-
- @Test
- public void populateDisplayDataDelegates() {
- doThrow(RuntimeException.class)
- .when(delegate).populateDisplayData(any(DisplayData.Builder.class));
-
- thrown.expect(RuntimeException.class);
- DisplayData.from(forwarding);
- }
-}