You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by mjsax <gi...@git.apache.org> on 2016/02/05 00:17:13 UTC
[GitHub] flink pull request: [FLINK-2721] [Storm Compatibility] Add Tuple m...
GitHub user mjsax opened a pull request:
https://github.com/apache/flink/pull/1591
[FLINK-2721] [Storm Compatibility] Add Tuple meta information
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/mjsax/flink flink-2721-tupleMetaInfo
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/1591.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #1591
----
commit 034b1ae44f1ea9eb95c33d0e6cca31e1c888aa0b
Author: mjsax <mj...@informatik.hu-berlin.de>
Date: 2016-02-04T23:02:41Z
[FLINK-2721] [Storm Compatibility] Add Tuple meta information
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2721] [Storm Compatibility] Add Tuple m...
Posted by mxm <gi...@git.apache.org>.
Github user mxm commented on a diff in the pull request:
https://github.com/apache/flink/pull/1591#discussion_r52102026
--- Diff: flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/StormTuple.java ---
@@ -41,55 +39,61 @@
*/
public class StormTuple<IN> implements backtype.storm.tuple.Tuple {
- /** The Storm representation of the original Flink tuple */
+ /** The Storm representation of the original Flink tuple. */
private final Values stormTuple;
- /** The schema (ie, ordered field names) of the tuple */
+ /** The schema (ie, ordered field names) of this tuple. */
private final Fields schema;
-
- /** The task id where this tuple is processed */
- private final int taskId;
- /** The producer of this tuple */
+ /** The task ID where this tuple was produced. */
+ private final int producerTaskId;
+ /** The input stream from which this tuple was received. */
private final String producerStreamId;
- /** The producer's component id of this tuple */
+ /** The producer's component ID of this tupl.e */
--- End diff --
typo
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2721] [Storm Compatibility] Add Tuple m...
Posted by mxm <gi...@git.apache.org>.
Github user mxm commented on a diff in the pull request:
https://github.com/apache/flink/pull/1591#discussion_r52102056
--- Diff: flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/wrappers/SpoutCollectorTest.java ---
@@ -75,12 +75,71 @@ public void testSpoutStormCollector() throws InstantiationException, IllegalAcce
}
}
+ @SuppressWarnings({ "rawtypes", "unchecked" })
+ @Test
+ public void testSpoutStormCollectorWithTaskId() throws InstantiationException, IllegalAccessException {
+ for (int numberOfAttributes = 0; numberOfAttributes < 25; ++numberOfAttributes) {
+ final SourceContext flinkCollector = mock(SourceContext.class);
+ final int taskId = 42;
+ final String streamId = "streamId";
+
+ HashMap<String, Integer> attributes = new HashMap<String, Integer>();
+ attributes.put(streamId, numberOfAttributes);
+
+ SpoutCollector<?> collector = new SpoutCollector(attributes, taskId, flinkCollector);
+
+ final Values tuple = new Values();
+ final Tuple flinkTuple = Tuple.getTupleClass(numberOfAttributes + 1).newInstance();
+
+ for (int i = 0; i < numberOfAttributes; ++i) {
+ tuple.add(new Integer(this.r.nextInt()));
+ flinkTuple.setField(tuple.get(i), i);
+ }
+ flinkTuple.setField(taskId, numberOfAttributes);
+
+ final List<Integer> taskIds;
+ final Object messageId = new Integer(this.r.nextInt());
+
+ taskIds = collector.emit(streamId, tuple, messageId);
+
+ Assert.assertNull(taskIds);
+
+ verify(flinkCollector).collect(flinkTuple);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ @Test(expected = UnsupportedOperationException.class)
+ public void testToManyAttributes() {
--- End diff --
typo
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2721] [Storm Compatibility] Add Tuple m...
Posted by mxm <gi...@git.apache.org>.
Github user mxm commented on a diff in the pull request:
https://github.com/apache/flink/pull/1591#discussion_r52102060
--- Diff: flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/api/FlinkTopology.java ---
@@ -229,9 +229,24 @@ private void translateTopology() {
boolean makeProgress = true;
while (bolts.size() > 0) {
if (!makeProgress) {
- throw new RuntimeException(
- "Unable to build Topology. Could not connect the following bolts: "
- + bolts.keySet());
+ StringBuilder strBld = new StringBuilder();
+ strBld.append("Unable to build Topology. Could not connect the following bolts:");
+ for (String boltId : bolts.keySet()) {
+ strBld.append("\n ");
+ strBld.append(boltId);
+ strBld.append(": missing input streams [");
+ for (Entry<GlobalStreamId, Grouping> streams : unprocessdInputsPerBolt
+ .get(boltId)) {
+ strBld.append("'");
+ strBld.append(streams.getKey().get_streamId());
+ strBld.append("' from '");
+ strBld.append(streams.getKey().get_componentId());
+ strBld.append("'; ");
+ }
+ strBld.append("]");
--- End diff --
Nice. That's helpful for debugging.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2721] [Storm Compatibility] Add Tuple m...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/flink/pull/1591
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2721] [Storm Compatibility] Add Tuple m...
Posted by mxm <gi...@git.apache.org>.
Github user mxm commented on a diff in the pull request:
https://github.com/apache/flink/pull/1591#discussion_r52102057
--- Diff: flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/wrappers/SpoutCollectorTest.java ---
@@ -75,12 +75,71 @@ public void testSpoutStormCollector() throws InstantiationException, IllegalAcce
}
}
+ @SuppressWarnings({ "rawtypes", "unchecked" })
+ @Test
+ public void testSpoutStormCollectorWithTaskId() throws InstantiationException, IllegalAccessException {
+ for (int numberOfAttributes = 0; numberOfAttributes < 25; ++numberOfAttributes) {
+ final SourceContext flinkCollector = mock(SourceContext.class);
+ final int taskId = 42;
+ final String streamId = "streamId";
+
+ HashMap<String, Integer> attributes = new HashMap<String, Integer>();
+ attributes.put(streamId, numberOfAttributes);
+
+ SpoutCollector<?> collector = new SpoutCollector(attributes, taskId, flinkCollector);
+
+ final Values tuple = new Values();
+ final Tuple flinkTuple = Tuple.getTupleClass(numberOfAttributes + 1).newInstance();
+
+ for (int i = 0; i < numberOfAttributes; ++i) {
+ tuple.add(new Integer(this.r.nextInt()));
+ flinkTuple.setField(tuple.get(i), i);
+ }
+ flinkTuple.setField(taskId, numberOfAttributes);
+
+ final List<Integer> taskIds;
+ final Object messageId = new Integer(this.r.nextInt());
+
+ taskIds = collector.emit(streamId, tuple, messageId);
+
+ Assert.assertNull(taskIds);
+
+ verify(flinkCollector).collect(flinkTuple);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ @Test(expected = UnsupportedOperationException.class)
+ public void testToManyAttributes() {
+ HashMap<String, Integer> attributes = new HashMap<String, Integer>();
+ attributes.put("", 26);
+
+ new SpoutCollector<Object>(attributes, -1, mock(SourceContext.class));
+ }
+
+ @SuppressWarnings("unchecked")
+ @Test(expected = UnsupportedOperationException.class)
+ public void testToManyAttributesWithTaskId() {
--- End diff --
typo
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2721] [Storm Compatibility] Add Tuple m...
Posted by mjsax <gi...@git.apache.org>.
Github user mjsax commented on the pull request:
https://github.com/apache/flink/pull/1591#issuecomment-181032182
Extended `VerifyMetaDataBolt.java` to check if the message ID is of "type" unanchored and fixed the typos.
Merging this.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2721] [Storm Compatibility] Add Tuple m...
Posted by mxm <gi...@git.apache.org>.
Github user mxm commented on a diff in the pull request:
https://github.com/apache/flink/pull/1591#discussion_r52102118
--- Diff: flink-contrib/flink-storm-examples/src/test/java/org/apache/flink/storm/tests/operators/VerifyMetaDataBolt.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.storm.tests.operators;
+
+import java.util.Map;
+
+import backtype.storm.task.OutputCollector;
+import backtype.storm.task.TopologyContext;
+import backtype.storm.topology.OutputFieldsDeclarer;
+import backtype.storm.topology.base.BaseRichBolt;
+import backtype.storm.tuple.Fields;
+import backtype.storm.tuple.Tuple;
+import backtype.storm.tuple.Values;
+
+public class VerifyMetaDataBolt extends BaseRichBolt {
+ private static final long serialVersionUID = 1353222852073800478L;
+
+ public static final String STREAM_ID = "boltMeta";
+
+ private OutputCollector collector;
+ private TopologyContext context;
+
+ public static boolean errorOccured = false;
+
+ @SuppressWarnings("rawtypes")
+ @Override
+ public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
+ this.collector = collector;
+ this.context = context;
+ }
+
+ @Override
+ public void execute(Tuple input) {
+ if (!input.getSourceComponent().equals(input.getString(0))
+ || !input.getSourceStreamId().equals(input.getString(1))
+ || !input.getSourceGlobalStreamid().get_componentId().equals(input.getString(0))
+ || !input.getSourceGlobalStreamid().get_streamId().equals(input.getString(1))
+ || input.getSourceTask() != input.getInteger(2).intValue()) {
+ errorOccured = true;
--- End diff --
Should the message id also be verified here sine it belongs to the meta data?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2721] [Storm Compatibility] Add Tuple m...
Posted by mjsax <gi...@git.apache.org>.
Github user mjsax commented on the pull request:
https://github.com/apache/flink/pull/1591#issuecomment-180098460
@mxm I would like to get this into 1.0 release. Would be great if you could review it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2721] [Storm Compatibility] Add Tuple m...
Posted by mxm <gi...@git.apache.org>.
Github user mxm commented on the pull request:
https://github.com/apache/flink/pull/1591#issuecomment-180765494
Looks like the meta data is now completely available :). +1 for merging this.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2721] [Storm Compatibility] Add Tuple m...
Posted by mxm <gi...@git.apache.org>.
Github user mxm commented on the pull request:
https://github.com/apache/flink/pull/1591#issuecomment-180288305
@mjsax Would be good to merge this. I'll take a look later.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---