[
https://issues.apache.org/jira/browse/FLINK-2721?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15135763#comment-15135763
]
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_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?
> 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)
