You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by mb...@apache.org on 2015/06/15 11:32:53 UTC
[03/27] flink git commit: [storm-compat] Added tests for Storm
compatibility API
[storm-compat] Added tests for Storm compatibility API
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/0a8b63aa
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/0a8b63aa
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/0a8b63aa
Branch: refs/heads/master
Commit: 0a8b63aaece33419ccb569bfaa9b2ea5ada1573c
Parents: 2bf76be
Author: mjsax <mj...@informatik.hu-berlin.de>
Authored: Thu May 14 12:26:43 2015 +0200
Committer: mbalassi <mb...@apache.org>
Committed: Sun Jun 14 22:58:47 2015 +0200
----------------------------------------------------------------------
.../api/FlinkOutputFieldsDeclarerTest.java | 177 +++++++++++++++++++
.../api/FlinkTopologyContextTest.java | 78 ++++++++
.../api/FlinkTopologyTest.java | 65 +++++++
.../stormcompatibility/util/AbstractTest.java | 42 +++++
.../wrappers/FiniteTestSpout.java | 83 +++++++++
.../wrappers/TestCollector.java | 42 +++++
6 files changed, 487 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/0a8b63aa/flink-staging/flink-streaming/flink-storm-compatibility/src/test/java/org/apache/flink/stormcompatibility/api/FlinkOutputFieldsDeclarerTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-storm-compatibility/src/test/java/org/apache/flink/stormcompatibility/api/FlinkOutputFieldsDeclarerTest.java b/flink-staging/flink-streaming/flink-storm-compatibility/src/test/java/org/apache/flink/stormcompatibility/api/FlinkOutputFieldsDeclarerTest.java
new file mode 100644
index 0000000..1902835
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-storm-compatibility/src/test/java/org/apache/flink/stormcompatibility/api/FlinkOutputFieldsDeclarerTest.java
@@ -0,0 +1,177 @@
+/*
+ * 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.flink.stormcompatibility.api;
+
+import java.util.LinkedList;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.stormcompatibility.util.AbstractTest;
+import org.junit.Assert;
+import org.junit.Test;
+
+import backtype.storm.tuple.Fields;
+import backtype.storm.utils.Utils;
+
+
+
+
+
+public class FlinkOutputFieldsDeclarerTest extends AbstractTest {
+
+ @Test
+ public void testDeclare() {
+ for(int i = 0; i < 4; ++i) {
+ for(int j = 0; j <= 25; ++j) {
+ this.runDeclareTest(i, j);
+ }
+ }
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testDeclareSimpleToManyAttributes() {
+ this.runDeclareTest(0, 26);
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testDeclareNonDirectToManyAttributes() {
+ this.runDeclareTest(1, 26);
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testDeclareDefaultStreamToManyAttributes() {
+ this.runDeclareTest(2, 26);
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testDeclareFullToManyAttributes() {
+ this.runDeclareTest(3, 26);
+ }
+
+ private void runDeclareTest(final int testCase, final int numberOfAttributes) {
+ final FlinkOutputFieldsDeclarer declarere = new FlinkOutputFieldsDeclarer();
+
+ final String[] attributes = new String[numberOfAttributes];
+ for(int i = 0; i < numberOfAttributes; ++i) {
+ attributes[i] = "a" + i;
+ }
+
+ switch(testCase) {
+ case 0:
+ this.declareSimple(declarere, attributes);
+ break;
+ case 1:
+ this.declareNonDirect(declarere, attributes);
+ break;
+ case 2:
+ this.declareDefaultStream(declarere, attributes);
+ break;
+ default:
+ this.declareFull(declarere, attributes);
+ }
+
+
+ final TypeInformation<?> type = declarere.getOutputType();
+
+ if(numberOfAttributes == 0) {
+ Assert.assertNull(type);
+ } else {
+ Assert.assertEquals(numberOfAttributes, type.getArity());
+ if(numberOfAttributes == 1) {
+ Assert.assertFalse(type.isTupleType());
+ } else {
+ Assert.assertTrue(type.isTupleType());
+ }
+ }
+ }
+
+ private void declareSimple(final FlinkOutputFieldsDeclarer declarere, final String[] attributes) {
+ declarere.declare(new Fields(attributes));
+ }
+
+ private void declareNonDirect(final FlinkOutputFieldsDeclarer declarere, final String[] attributes) {
+ declarere.declare(false, new Fields(attributes));
+ }
+
+ private void declareDefaultStream(final FlinkOutputFieldsDeclarer declarere, final String[] attributes) {
+ declarere.declareStream(Utils.DEFAULT_STREAM_ID, new Fields(attributes));
+ }
+
+ private void declareFull(final FlinkOutputFieldsDeclarer declarere, final String[] attributes) {
+ declarere.declareStream(Utils.DEFAULT_STREAM_ID, false, new Fields(attributes));
+ }
+
+ @Test(expected = UnsupportedOperationException.class)
+ public void testDeclareDirect() {
+ new FlinkOutputFieldsDeclarer().declare(true, null);
+ }
+
+ @Test(expected = UnsupportedOperationException.class)
+ public void testDeclareNonDefaultStrem() {
+ new FlinkOutputFieldsDeclarer().declareStream("dummy", null);
+ }
+
+ @Test(expected = UnsupportedOperationException.class)
+ public void testDeclareDirect2() {
+ new FlinkOutputFieldsDeclarer().declareStream(Utils.DEFAULT_STREAM_ID, true, null);
+ }
+
+ @Test(expected = UnsupportedOperationException.class)
+ public void testDeclareNonDefaultStrem2() {
+ new FlinkOutputFieldsDeclarer().declareStream("dummy", this.r.nextBoolean(), null);
+ }
+
+ @Test
+ public void testGetGroupingFieldIndexes() {
+ final int numberOfAttributes = 5 + this.r.nextInt(21);
+ final String[] attributes = new String[numberOfAttributes];
+ for(int i = 0; i < numberOfAttributes; ++i) {
+ attributes[i] = "a" + i;
+ }
+
+ final FlinkOutputFieldsDeclarer declarere = new FlinkOutputFieldsDeclarer();
+ declarere.declare(new Fields(attributes));
+
+ final int numberOfKeys = 1 + this.r.nextInt(25);
+ final LinkedList<String> groupingFields = new LinkedList<String>();
+ final boolean[] indexes = new boolean[numberOfAttributes];
+
+ for(int i = 0; i < numberOfAttributes; ++i) {
+ if(this.r.nextInt(26) < numberOfKeys) {
+ groupingFields.add(attributes[i]);
+ indexes[i] = true;
+ } else {
+ indexes[i] = false;
+ }
+ }
+
+ final int[] expectedResult = new int[groupingFields.size()];
+ int j = 0;
+ for(int i = 0; i < numberOfAttributes; ++i) {
+ if(indexes[i]) {
+ expectedResult[j++] = i;
+ }
+ }
+
+ final int[] result = declarere.getGroupingFieldIndexes(groupingFields);
+
+ Assert.assertEquals(expectedResult.length, result.length);
+ for(int i = 0; i < expectedResult.length; ++i) {
+ Assert.assertEquals(expectedResult[i], result[i]);
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/0a8b63aa/flink-staging/flink-streaming/flink-storm-compatibility/src/test/java/org/apache/flink/stormcompatibility/api/FlinkTopologyContextTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-storm-compatibility/src/test/java/org/apache/flink/stormcompatibility/api/FlinkTopologyContextTest.java b/flink-staging/flink-streaming/flink-storm-compatibility/src/test/java/org/apache/flink/stormcompatibility/api/FlinkTopologyContextTest.java
new file mode 100644
index 0000000..029a400
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-storm-compatibility/src/test/java/org/apache/flink/stormcompatibility/api/FlinkTopologyContextTest.java
@@ -0,0 +1,78 @@
+/*
+ * 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.flink.stormcompatibility.api;
+
+import org.junit.Test;
+
+import backtype.storm.metric.api.ICombiner;
+import backtype.storm.metric.api.IMetric;
+import backtype.storm.metric.api.IReducer;
+
+
+
+
+
+public class FlinkTopologyContextTest {
+
+ @Test(expected = UnsupportedOperationException.class)
+ public void testAddTaskHook() {
+ new FlinkTopologyContext(null, null, null).addTaskHook(null);
+ }
+
+ @Test(expected = UnsupportedOperationException.class)
+ public void testGetHooks() {
+ new FlinkTopologyContext(null, null, null).getHooks();
+ }
+
+ @SuppressWarnings("rawtypes")
+ @Test(expected = UnsupportedOperationException.class)
+ public void testRegisteredMetric1() {
+ new FlinkTopologyContext(null, null, null).registerMetric(null, (ICombiner)null, 0);
+ }
+
+ @SuppressWarnings("rawtypes")
+ @Test(expected = UnsupportedOperationException.class)
+ public void testRegisteredMetric2() {
+ new FlinkTopologyContext(null, null, null).registerMetric(null, (IReducer)null, 0);
+ }
+
+ @Test(expected = UnsupportedOperationException.class)
+ public void testRegisteredMetric3() {
+ new FlinkTopologyContext(null, null, null).registerMetric(null, (IMetric)null, 0);
+ }
+
+ @Test(expected = UnsupportedOperationException.class)
+ public void testGetRegisteredMetricByName() {
+ new FlinkTopologyContext(null, null, null).getRegisteredMetricByName(null);
+ }
+
+ @Test(expected = UnsupportedOperationException.class)
+ public void testSetAllSubscribedState() {
+ new FlinkTopologyContext(null, null, null).setAllSubscribedState(null);
+ }
+
+ @Test(expected = UnsupportedOperationException.class)
+ public void testSetSubscribedState1() {
+ new FlinkTopologyContext(null, null, null).setSubscribedState(null, null);
+ }
+
+ @Test(expected = UnsupportedOperationException.class)
+ public void testSetSubscribedState2() {
+ new FlinkTopologyContext(null, null, null).setSubscribedState(null, null, null);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/0a8b63aa/flink-staging/flink-streaming/flink-storm-compatibility/src/test/java/org/apache/flink/stormcompatibility/api/FlinkTopologyTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-storm-compatibility/src/test/java/org/apache/flink/stormcompatibility/api/FlinkTopologyTest.java b/flink-staging/flink-streaming/flink-storm-compatibility/src/test/java/org/apache/flink/stormcompatibility/api/FlinkTopologyTest.java
new file mode 100644
index 0000000..4f1f40f
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-storm-compatibility/src/test/java/org/apache/flink/stormcompatibility/api/FlinkTopologyTest.java
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.stormcompatibility.api;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+
+
+
+
+public class FlinkTopologyTest {
+
+ @Test
+ public void testDefaultParallelism() {
+ final FlinkTopology topology = new FlinkTopology(null);
+ Assert.assertEquals(1, topology.getParallelism());
+ }
+
+ @Test(expected = UnsupportedOperationException.class)
+ public void testExecute() throws Exception {
+ new FlinkTopology(null).execute();
+ }
+
+ @Test(expected = UnsupportedOperationException.class)
+ public void testExecuteWithName() throws Exception {
+ new FlinkTopology(null).execute(null);
+ }
+
+ @Test
+ public void testNumberOfTasks() {
+ final FlinkTopology topology = new FlinkTopology(null);
+
+ Assert.assertEquals(0, topology.getNumberOfTasks());
+
+ topology.increaseNumberOfTasks(3);
+ Assert.assertEquals(3, topology.getNumberOfTasks());
+
+ topology.increaseNumberOfTasks(2);
+ Assert.assertEquals(5, topology.getNumberOfTasks());
+
+ topology.increaseNumberOfTasks(8);
+ Assert.assertEquals(13, topology.getNumberOfTasks());
+ }
+
+ @Test(expected = AssertionError.class)
+ public void testAssert() {
+ new FlinkTopology(null).increaseNumberOfTasks(0);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/0a8b63aa/flink-staging/flink-streaming/flink-storm-compatibility/src/test/java/org/apache/flink/stormcompatibility/util/AbstractTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-storm-compatibility/src/test/java/org/apache/flink/stormcompatibility/util/AbstractTest.java b/flink-staging/flink-streaming/flink-storm-compatibility/src/test/java/org/apache/flink/stormcompatibility/util/AbstractTest.java
new file mode 100644
index 0000000..100ca87
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-storm-compatibility/src/test/java/org/apache/flink/stormcompatibility/util/AbstractTest.java
@@ -0,0 +1,42 @@
+/*
+ * 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.flink.stormcompatibility.util;
+
+import java.util.Random;
+
+import org.junit.Before;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+
+
+
+public abstract class AbstractTest {
+ private static final Logger LOG = LoggerFactory.getLogger(AbstractTest.class);
+
+ protected long seed;
+ protected Random r;
+
+ @Before
+ public void prepare() {
+ this.seed = System.currentTimeMillis();
+ this.r = new Random(this.seed);
+ LOG.info("Test seed: {}", new Long(this.seed));
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/0a8b63aa/flink-staging/flink-streaming/flink-storm-compatibility/src/test/java/org/apache/flink/stormcompatibility/wrappers/FiniteTestSpout.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-storm-compatibility/src/test/java/org/apache/flink/stormcompatibility/wrappers/FiniteTestSpout.java b/flink-staging/flink-streaming/flink-storm-compatibility/src/test/java/org/apache/flink/stormcompatibility/wrappers/FiniteTestSpout.java
new file mode 100644
index 0000000..52136a9
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-storm-compatibility/src/test/java/org/apache/flink/stormcompatibility/wrappers/FiniteTestSpout.java
@@ -0,0 +1,83 @@
+/*
+ * 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.flink.stormcompatibility.wrappers;
+
+import java.util.Map;
+
+import backtype.storm.spout.SpoutOutputCollector;
+import backtype.storm.task.TopologyContext;
+import backtype.storm.topology.IRichSpout;
+import backtype.storm.topology.OutputFieldsDeclarer;
+import backtype.storm.tuple.Fields;
+import backtype.storm.tuple.Values;
+
+
+
+
+
+class FiniteTestSpout implements IRichSpout {
+ private static final long serialVersionUID = 7992419478267824279L;
+
+ private int numberOfOutputTuples;
+ private SpoutOutputCollector collector;
+
+
+
+ public FiniteTestSpout(final int numberOfOutputTuples) {
+ this.numberOfOutputTuples = numberOfOutputTuples;
+ }
+
+
+
+ @Override
+ public void open(@SuppressWarnings("rawtypes") final Map conf, final TopologyContext context, @SuppressWarnings("hiding") final SpoutOutputCollector collector) {
+ this.collector = collector;
+ }
+
+ @Override
+ public void close() {/* nothing to do */}
+
+ @Override
+ public void activate() {/* nothing to do */}
+
+ @Override
+ public void deactivate() {/* nothing to do */}
+
+ @Override
+ public void nextTuple() {
+ if(--this.numberOfOutputTuples >= 0) {
+ this.collector.emit(new Values(new Integer(this.numberOfOutputTuples)));
+ }
+ }
+
+ @Override
+ public void ack(final Object msgId) {/* nothing to do */}
+
+ @Override
+ public void fail(final Object msgId) {/* nothing to do */}
+
+ @Override
+ public void declareOutputFields(final OutputFieldsDeclarer declarer) {
+ declarer.declare(new Fields("dummy"));
+ }
+
+ @Override
+ public Map<String, Object> getComponentConfiguration() {
+ return null;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/0a8b63aa/flink-staging/flink-streaming/flink-storm-compatibility/src/test/java/org/apache/flink/stormcompatibility/wrappers/TestCollector.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-storm-compatibility/src/test/java/org/apache/flink/stormcompatibility/wrappers/TestCollector.java b/flink-staging/flink-streaming/flink-storm-compatibility/src/test/java/org/apache/flink/stormcompatibility/wrappers/TestCollector.java
new file mode 100644
index 0000000..fc2768d
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-storm-compatibility/src/test/java/org/apache/flink/stormcompatibility/wrappers/TestCollector.java
@@ -0,0 +1,42 @@
+/*
+ * 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.flink.stormcompatibility.wrappers;
+
+import java.util.LinkedList;
+
+import org.apache.flink.api.java.tuple.Tuple1;
+import org.apache.flink.util.Collector;
+
+
+
+
+
+class TestCollector implements Collector<Tuple1<Integer>> {
+ public LinkedList<Tuple1<Integer>> result = new LinkedList<Tuple1<Integer>>();
+
+ public TestCollector() {}
+
+ @Override
+ public void collect(final Tuple1<Integer> record) {
+ this.result.add(record.copy());
+ }
+
+ @Override
+ public void close() {/* nothing to to */}
+
+}