[
https://issues.apache.org/jira/browse/FLINK-2721?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15135761#comment-15135761
]
ASF GitHub Bot commented on FLINK-2721:
---------------------------------------
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
> Add Tuple meta information
> --------------------------
>
> Key: FLINK-2721
> URL: https://issues.apache.org/jira/browse/FLINK-2721
> Project: Flink
> Issue Type: New Feature
> Components: Storm Compatibility
> Reporter: Matthias J. Sax
> Assignee: Matthias J. Sax
> Priority: Minor
>
> In {{Bolt.execute(Tuple input)}} the given input tuple contains meta
> information about its origin (like source component name, stream id, source
> task ID).
> This meta information in currently not provided by Flink and the
> corresponding methods throw an {{UnsupportedOperationException}}.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
