http://git-wip-us.apache.org/repos/asf/flink/blob/4cb96708/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/AbstractStormCollector.java ---------------------------------------------------------------------- diff --git a/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/AbstractStormCollector.java b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/AbstractStormCollector.java new file mode 100644 index 0000000..85d895c --- /dev/null +++ b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/AbstractStormCollector.java @@ -0,0 +1,143 @@ +/* + * 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.wrappers; + +import org.apache.flink.api.java.tuple.Tuple; +import org.apache.flink.api.java.tuple.Tuple0; +import org.apache.flink.api.java.tuple.Tuple25; +import org.apache.flink.storm.util.SplitStreamType; + +import java.util.HashMap; +import java.util.List; +import java.util.Map.Entry; + +/** + * A {@link AbstractStormCollector} transforms Storm tuples to Flink tuples. + */ +abstract class AbstractStormCollector<OUT> { + + /** Flink output tuple of concrete type {@link Tuple0} to {@link Tuple25} per output stream. */ + protected final HashMap<String, Tuple> outputTuple = new HashMap<String, Tuple>(); + /** Flink split tuple. Used, if multiple output streams are declared. */ + private final SplitStreamType<Object> splitTuple = new SplitStreamType<Object>(); + /** + * The number of attributes of the output tuples per stream. (Determines the concrete type of {@link #outputTuple}). + * If {@link #numberOfAttributes} is zero, {@link #outputTuple} is not used and "raw" data type is used. + */ + protected final HashMap<String, Integer> numberOfAttributes; + /** Indicates of multiple output stream are declared and thus {@link SplitStreamType} must be used as output. */ + private final boolean split; + /** Is set to {@code true} each time a tuple is emitted. */ + boolean tupleEmitted = false; + + /** + * Instantiates a new {@link AbstractStormCollector} that emits Flink tuples via {@link #doEmit(Object)}. If the + * number of attributes is negative, any output type is supported (ie, raw type). If the number of attributes is + * between 0 and 25, the output type is {@link Tuple0} to {@link Tuple25}, respectively. + * + * @param numberOfAttributes + * The number of attributes of the emitted tuples per output stream. + * @throws UnsupportedOperationException + * if the specified number of attributes is greater than 25 + */ + AbstractStormCollector(final HashMap<String, Integer> numberOfAttributes) + throws UnsupportedOperationException { + assert (numberOfAttributes != null); + + this.numberOfAttributes = numberOfAttributes; + this.split = this.numberOfAttributes.size() > 1; + + for (Entry<String, Integer> outputStream : numberOfAttributes.entrySet()) { + final int numAtt = outputStream.getValue(); + assert (numAtt >= -1); + + if (numAtt > 25) { + throw new UnsupportedOperationException( + "Flink cannot handle more then 25 attributes, but " + numAtt + + " are declared for stream '" + outputStream.getKey() + + "' by the given bolt"); + } else if (numAtt >= 0) { + try { + this.outputTuple.put(outputStream.getKey(), + org.apache.flink.api.java.tuple.Tuple.getTupleClass(numAtt) + .newInstance()); + } catch (final InstantiationException e) { + throw new RuntimeException(e); + } catch (final IllegalAccessException e) { + throw new RuntimeException(e); + } + + } + } + } + + /** + * Transforms a Storm tuple into a Flink tuple of type {@code OUT} and emits this tuple via {@link #doEmit(Object)} + * to the specified output stream. + * + * @param The + * The output stream id. + * @param tuple + * The Storm tuple to be emitted. + * @return the return value of {@link #doEmit(Object)} + */ + @SuppressWarnings("unchecked") + protected final List<Integer> tansformAndEmit(final String streamId, final List<Object> tuple) { + List<Integer> taskIds; + + final int numAtt = this.numberOfAttributes.get(streamId); + if (numAtt > -1) { + assert (tuple.size() == numAtt); + Tuple out = this.outputTuple.get(streamId); + for (int i = 0; i < numAtt; ++i) { + out.setField(tuple.get(i), i); + } + if (this.split) { + this.splitTuple.streamId = streamId; + this.splitTuple.value = out; + + taskIds = doEmit((OUT) this.splitTuple); + } else { + taskIds = doEmit((OUT) out); + } + + } else { + assert (tuple.size() == 1); + if (split) { + this.splitTuple.streamId = streamId; + this.splitTuple.value = tuple.get(0); + + taskIds = doEmit((OUT) this.splitTuple); + } else { + taskIds = doEmit((OUT) tuple.get(0)); + } + } + this.tupleEmitted = true; + + return taskIds; + } + + /** + * Emits a Flink tuple. + * + * @param flinkTuple + * The tuple to be emitted. + * @return the IDs of the tasks this tuple was sent to + */ + protected abstract List<Integer> doEmit(OUT flinkTuple); + +}
http://git-wip-us.apache.org/repos/asf/flink/blob/4cb96708/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/BoltCollector.java ---------------------------------------------------------------------- diff --git a/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/BoltCollector.java b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/BoltCollector.java new file mode 100644 index 0000000..58fd098 --- /dev/null +++ b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/BoltCollector.java @@ -0,0 +1,89 @@ +/* + * 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.wrappers; + +import backtype.storm.task.IOutputCollector; +import backtype.storm.tuple.Tuple; + +import org.apache.flink.api.java.tuple.Tuple0; +import org.apache.flink.api.java.tuple.Tuple25; +import org.apache.flink.streaming.api.operators.Output; +import org.apache.flink.util.Collector; + +import java.util.Collection; +import java.util.HashMap; +import java.util.List; + +/** + * A {@link BoltCollector} is used by {@link BoltWrapper} to provided an Storm compatible + * output collector to the wrapped bolt. It transforms the emitted Storm tuples into Flink tuples + * and emits them via the provide {@link Output} object. + */ +class BoltCollector<OUT> extends AbstractStormCollector<OUT> implements IOutputCollector { + + /** The Flink output Collector */ + private final Collector<OUT> flinkOutput; + + /** + * Instantiates a new {@link BoltCollector} that emits Flink tuples to the given Flink output object. If the + * number of attributes is negative, any output type is supported (ie, raw type). If the number of attributes is + * between 0 and 25, the output type is {@link Tuple0} to {@link Tuple25}, respectively. + * + * @param numberOfAttributes + * The number of attributes of the emitted tuples per output stream. + * @param flinkOutput + * The Flink output object to be used. + * @throws UnsupportedOperationException + * if the specified number of attributes is greater than 25 + */ + BoltCollector(final HashMap<String, Integer> numberOfAttributes, + final Collector<OUT> flinkOutput) throws UnsupportedOperationException { + super(numberOfAttributes); + assert (flinkOutput != null); + this.flinkOutput = flinkOutput; + } + + @Override + protected List<Integer> doEmit(final OUT flinkTuple) { + this.flinkOutput.collect(flinkTuple); + // TODO + return null; + } + + @Override + public void reportError(final Throwable error) { + // not sure, if Flink can support this + } + + @Override + public List<Integer> emit(final String streamId, final Collection<Tuple> anchors, final List<Object> tuple) { + return this.tansformAndEmit(streamId, tuple); + } + + @Override + public void emitDirect(final int taskId, final String streamId, final Collection<Tuple> anchors, final List<Object> tuple) { + throw new UnsupportedOperationException("Direct emit is not supported by Flink"); + } + + @Override + public void ack(final Tuple input) {} + + @Override + public void fail(final Tuple input) {} + +} http://git-wip-us.apache.org/repos/asf/flink/blob/4cb96708/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/BoltWrapper.java ---------------------------------------------------------------------- diff --git a/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/BoltWrapper.java b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/BoltWrapper.java new file mode 100644 index 0000000..b16fc09 --- /dev/null +++ b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/BoltWrapper.java @@ -0,0 +1,260 @@ +/* + * 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.wrappers; + +import java.util.Collection; +import java.util.HashMap; + +import backtype.storm.generated.StormTopology; +import backtype.storm.task.OutputCollector; +import backtype.storm.task.TopologyContext; +import backtype.storm.topology.IRichBolt; +import backtype.storm.tuple.Fields; + +import org.apache.flink.api.common.ExecutionConfig.GlobalJobParameters; +import org.apache.flink.api.java.tuple.Tuple0; +import org.apache.flink.api.java.tuple.Tuple1; +import org.apache.flink.api.java.tuple.Tuple25; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.storm.util.SplitStreamType; +import org.apache.flink.storm.util.StormConfig; +import org.apache.flink.streaming.api.operators.AbstractStreamOperator; +import org.apache.flink.streaming.api.operators.OneInputStreamOperator; +import org.apache.flink.streaming.api.operators.TimestampedCollector; +import org.apache.flink.streaming.api.watermark.Watermark; +import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; + +import com.google.common.collect.Sets; + +/** + * A {@link BoltWrapper} wraps an {@link IRichBolt} in order to execute the Storm bolt within a Flink Streaming + * program. It takes the Flink input tuples of type {@code IN} and transforms them into {@link StormTuple}s that the + * bolt can process. Furthermore, it takes the bolt's output tuples and transforms them into Flink tuples of type + * {@code OUT} (see {@link AbstractStormCollector} for supported types).<br /> + * <br /> + * <strong>CAUTION: currently, only simple bolts are supported! (ie, bolts that do not use the Storm configuration + * <code>Map</code> or <code>TopologyContext</code> that is provided by the bolt's <code>open(..)</code> method. + * Furthermore, acking and failing of tuples as well as accessing tuple attributes by field names is not supported so + * far.</strong> + */ +public class BoltWrapper<IN, OUT> extends AbstractStreamOperator<OUT> implements OneInputStreamOperator<IN, OUT> { + private static final long serialVersionUID = -4788589118464155835L; + + /** The wrapped Storm {@link IRichBolt bolt}. */ + private final IRichBolt bolt; + /** Number of attributes of the bolt's output tuples per stream. */ + private final HashMap<String, Integer> numberOfAttributes; + /** The schema (ie, ordered field names) of the input stream. */ + private final Fields inputSchema; + /** The original Storm topology. */ + protected StormTopology stormTopology; + /** + * We have to use this because Operators must output + * {@link org.apache.flink.streaming.runtime.streamrecord.StreamRecord}. + */ + private TimestampedCollector<OUT> flinkCollector; + + /** + * Instantiates a new {@link BoltWrapper} that wraps the given Storm {@link IRichBolt bolt} such that it can be + * used within a Flink streaming program. As no input schema is defined, attribute-by-name access in only possible + * for POJO input types. The output type will be one of {@link Tuple0} to {@link Tuple25} depending on the bolt's + * declared number of attributes. + * + * @param bolt + * The Storm {@link IRichBolt bolt} to be used. + * @throws IllegalArgumentException + * If the number of declared output attributes is not with range [0;25]. + */ + public BoltWrapper(final IRichBolt bolt) throws IllegalArgumentException { + this(bolt, null, (Collection<String>) null); + } + + /** + * Instantiates a new {@link BoltWrapper} that wraps the given Storm {@link IRichBolt bolt} such that it can be + * used within a Flink streaming program. The given input schema enable attribute-by-name access for input types + * {@link Tuple0} to {@link Tuple25}. The output type will be one of {@link Tuple0} to {@link Tuple25} depending on + * the bolt's declared number of attributes. + * + * @param bolt + * The Storm {@link IRichBolt bolt} to be used. + * @param inputSchema + * The schema (ie, ordered field names) of the input stream. + * @throws IllegalArgumentException + * If the number of declared output attributes is not with range [0;25]. + */ + public BoltWrapper(final IRichBolt bolt, final Fields inputSchema) + throws IllegalArgumentException { + this(bolt, inputSchema, (Collection<String>) null); + } + + /** + * Instantiates a new {@link BoltWrapper} that wraps the given Storm {@link IRichBolt bolt} such that it can be + * used within a Flink streaming program. As no input schema is defined, attribute-by-name access in only possible + * for POJO input types. The output type can be any type if parameter {@code rawOutput} is {@code true} and the + * bolt's number of declared output tuples is 1. If {@code rawOutput} is {@code false} the output type will be one + * of {@link Tuple0} to {@link Tuple25} depending on the bolt's declared number of attributes. + * + * @param bolt + * The Storm {@link IRichBolt bolt} to be used. + * @param rawOutputs + * Contains stream names if a single attribute output stream, should not be of type {@link Tuple1} but be + * of a raw type. + * @throws IllegalArgumentException + * If {@code rawOuput} is {@code true} and the number of declared output attributes is not 1 or if + * {@code rawOuput} is {@code false} and the number of declared output attributes is not with range + * [1;25]. + */ + public BoltWrapper(final IRichBolt bolt, final String[] rawOutputs) + throws IllegalArgumentException { + this(bolt, null, Sets.newHashSet(rawOutputs)); + } + + /** + * Instantiates a new {@link BoltWrapper} that wraps the given Storm {@link IRichBolt bolt} such that it can be + * used within a Flink streaming program. As no input schema is defined, attribute-by-name access in only possible + * for POJO input types. The output type can be any type if parameter {@code rawOutput} is {@code true} and the + * bolt's number of declared output tuples is 1. If {@code rawOutput} is {@code false} the output type will be one + * of {@link Tuple0} to {@link Tuple25} depending on the bolt's declared number of attributes. + * + * @param bolt + * The Storm {@link IRichBolt bolt} to be used. + * @param rawOutputs + * Contains stream names if a single attribute output stream, should not be of type {@link Tuple1} but be + * of a raw type. + * @throws IllegalArgumentException + * If {@code rawOuput} is {@code true} and the number of declared output attributes is not 1 or if + * {@code rawOuput} is {@code false} and the number of declared output attributes is not with range + * [1;25]. + */ + public BoltWrapper(final IRichBolt bolt, final Collection<String> rawOutputs) + throws IllegalArgumentException { + this(bolt, null, rawOutputs); + } + + /** + * Instantiates a new {@link BoltWrapper} that wraps the given Storm {@link IRichBolt bolt} such that it can be + * used within a Flink streaming program. The given input schema enable attribute-by-name access for input types + * {@link Tuple0} to {@link Tuple25}. The output type can be any type if parameter {@code rawOutput} is {@code true} + * and the bolt's number of declared output tuples is 1. If {@code rawOutput} is {@code false} the output type will + * be one of {@link Tuple0} to {@link Tuple25} depending on the bolt's declared number of attributes. + * + * @param bolt + * The Storm {@link IRichBolt bolt} to be used. + * @param inputSchema + * The schema (ie, ordered field names) of the input stream. + * @param rawOutputs + * Contains stream names if a single attribute output stream, should not be of type {@link Tuple1} but be + * of a raw type. + * @throws IllegalArgumentException + * If {@code rawOuput} is {@code true} and the number of declared output attributes is not 1 or if + * {@code rawOuput} is {@code false} and the number of declared output attributes is not with range + * [0;25]. + */ + public BoltWrapper(final IRichBolt bolt, final Fields inputSchema, + final String[] rawOutputs) throws IllegalArgumentException { + this(bolt, inputSchema, Sets.newHashSet(rawOutputs)); + } + + /** + * Instantiates a new {@link BoltWrapper} that wraps the given Storm {@link IRichBolt bolt} such that it can be + * used within a Flink streaming program. The given input schema enable attribute-by-name access for input types + * {@link Tuple0} to {@link Tuple25}. The output type can be any type if parameter {@code rawOutput} is {@code true} + * and the bolt's number of declared output tuples is 1. If {@code rawOutput} is {@code false} the output type will + * be one of {@link Tuple0} to {@link Tuple25} depending on the bolt's declared number of attributes. + * + * @param bolt + * The Storm {@link IRichBolt bolt} to be used. + * @param inputSchema + * The schema (ie, ordered field names) of the input stream. + * @param rawOutputs + * Contains stream names if a single attribute output stream, should not be of type {@link Tuple1} but be + * of a raw type. + * @throws IllegalArgumentException + * If {@code rawOuput} is {@code true} and the number of declared output attributes is not 1 or if + * {@code rawOuput} is {@code false} and the number of declared output attributes is not with range + * [0;25]. + */ + public BoltWrapper(final IRichBolt bolt, final Fields inputSchema, + final Collection<String> rawOutputs) throws IllegalArgumentException { + this.bolt = bolt; + this.inputSchema = inputSchema; + this.numberOfAttributes = WrapperSetupHelper.getNumberOfAttributes(bolt, rawOutputs); + } + + /** + * Sets the original Storm topology. + * + * @param stormTopology + * The original Storm topology. + */ + public void setStormTopology(StormTopology stormTopology) { + this.stormTopology = stormTopology; + } + + @Override + public void open(final Configuration parameters) throws Exception { + super.open(parameters); + + this.flinkCollector = new TimestampedCollector<OUT>(output); + OutputCollector stormCollector = null; + + if (this.numberOfAttributes.size() > 0) { + stormCollector = new OutputCollector(new BoltCollector<OUT>( + this.numberOfAttributes, flinkCollector)); + } + + GlobalJobParameters config = super.executionConfig.getGlobalJobParameters(); + StormConfig stormConfig = new StormConfig(); + + if (config != null) { + if (config instanceof StormConfig) { + stormConfig = (StormConfig) config; + } else { + stormConfig.putAll(config.toMap()); + } + } + + final TopologyContext topologyContext = WrapperSetupHelper.createTopologyContext( + super.runtimeContext, this.bolt, this.stormTopology, stormConfig); + + this.bolt.prepare(stormConfig, topologyContext, stormCollector); + } + + @Override + public void dispose() { + this.bolt.cleanup(); + } + + @SuppressWarnings("unchecked") + @Override + public void processElement(final StreamRecord<IN> element) throws Exception { + this.flinkCollector.setTimestamp(element.getTimestamp()); + IN value = element.getValue(); + if (value instanceof SplitStreamType) { + this.bolt.execute(new StormTuple<IN>(((SplitStreamType<IN>) value).value, + inputSchema)); + } else { + this.bolt.execute(new StormTuple<IN>(value, inputSchema)); + } + } + + @Override + public void processWatermark(Watermark mark) throws Exception { + this.output.emitWatermark(mark); + } + +} http://git-wip-us.apache.org/repos/asf/flink/blob/4cb96708/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/FlinkTopologyContext.java ---------------------------------------------------------------------- diff --git a/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/FlinkTopologyContext.java b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/FlinkTopologyContext.java new file mode 100644 index 0000000..68368bf --- /dev/null +++ b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/FlinkTopologyContext.java @@ -0,0 +1,165 @@ +/* + * 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.wrappers; + +import backtype.storm.generated.StormTopology; +import backtype.storm.hooks.ITaskHook; +import backtype.storm.metric.api.CombinedMetric; +import backtype.storm.metric.api.ICombiner; +import backtype.storm.metric.api.IMetric; +import backtype.storm.metric.api.IReducer; +import backtype.storm.metric.api.ReducedMetric; +import backtype.storm.state.ISubscribedState; +import backtype.storm.task.TopologyContext; +import backtype.storm.tuple.Fields; + +import java.util.Collection; +import java.util.List; +import java.util.Map; + +import clojure.lang.Atom; + +/** + * {@link FlinkTopologyContext} is a {@link TopologyContext} that overwrites certain method that are not applicable when + * a Storm topology is executed within Flink. + */ +final class FlinkTopologyContext extends TopologyContext { + + /** + * Instantiates a new {@link FlinkTopologyContext} for a given Storm topology. The context object is instantiated + * for each parallel task + */ + FlinkTopologyContext(final StormTopology topology, + @SuppressWarnings("rawtypes") final Map stormConf, + final Map<Integer, String> taskToComponent, final Map<String, List<Integer>> componentToSortedTasks, + final Map<String, Map<String, Fields>> componentToStreamToFields, final String stormId, final String codeDir, + final String pidDir, final Integer taskId, final Integer workerPort, final List<Integer> workerTasks, + final Map<String, Object> defaultResources, final Map<String, Object> userResources, + final Map<String, Object> executorData, @SuppressWarnings("rawtypes") final Map registeredMetrics, + final Atom openOrPrepareWasCalled) { + super(topology, stormConf, taskToComponent, componentToSortedTasks, componentToStreamToFields, stormId, + codeDir, pidDir, taskId, workerPort, workerTasks, defaultResources, userResources, executorData, + registeredMetrics, openOrPrepareWasCalled); + } + + /** + * Not supported by Flink. + * + * @throws UnsupportedOperationException + * at every invocation + */ + @Override + public void addTaskHook(final ITaskHook hook) { + throw new UnsupportedOperationException("Task hooks are not supported by Flink"); + } + + /** + * Not supported by Flink. + * + * @throws UnsupportedOperationException + * at every invocation + */ + @Override + public Collection<ITaskHook> getHooks() { + throw new UnsupportedOperationException("Task hooks are not supported by Flink"); + } + + /** + * Not supported by Flink. + * + * @throws UnsupportedOperationException + * at every invocation + */ + @Override + public IMetric getRegisteredMetricByName(final String name) { + throw new UnsupportedOperationException("Metrics are not supported by Flink"); + + } + + /** + * Not supported by Flink. + * + * @throws UnsupportedOperationException + * at every invocation + */ + @SuppressWarnings("rawtypes") + @Override + public CombinedMetric registerMetric(final String name, final ICombiner combiner, final int timeBucketSizeInSecs) { + throw new UnsupportedOperationException("Metrics are not supported by Flink"); + } + + /** + * Not supported by Flink. + * + * @throws UnsupportedOperationException + * at every invocation + */ + @SuppressWarnings("rawtypes") + @Override + public ReducedMetric registerMetric(final String name, final IReducer combiner, final int timeBucketSizeInSecs) { + throw new UnsupportedOperationException("Metrics are not supported by Flink"); + } + + /** + * Not supported by Flink. + * + * @throws UnsupportedOperationException + * at every invocation + */ + @SuppressWarnings("unchecked") + @Override + public IMetric registerMetric(final String name, final IMetric metric, final int timeBucketSizeInSecs) { + throw new UnsupportedOperationException("Metrics are not supported by Flink"); + } + + /** + * Not supported by Flink. + * + * @throws UnsupportedOperationException + * at every invocation + */ + @Override + public <T extends ISubscribedState> T setAllSubscribedState(final T obj) { + throw new UnsupportedOperationException("Not supported by Flink"); + + } + + /** + * Not supported by Flink. + * + * @throws UnsupportedOperationException + * at every invocation + */ + @Override + public <T extends ISubscribedState> T setSubscribedState(final String componentId, final T obj) { + throw new UnsupportedOperationException("Not supported by Flink"); + } + + /** + * Not supported by Flink. + * + * @throws UnsupportedOperationException + * at every invocation + */ + @Override + public <T extends ISubscribedState> T setSubscribedState(final String componentId, final String streamId, final T + obj) { + throw new UnsupportedOperationException("Not supported by Flink"); + } + +} http://git-wip-us.apache.org/repos/asf/flink/blob/4cb96708/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/SetupOutputFieldsDeclarer.java ---------------------------------------------------------------------- diff --git a/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/SetupOutputFieldsDeclarer.java b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/SetupOutputFieldsDeclarer.java new file mode 100644 index 0000000..507305b --- /dev/null +++ b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/SetupOutputFieldsDeclarer.java @@ -0,0 +1,66 @@ +/* + * 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.wrappers; + +import java.util.HashMap; + +import backtype.storm.topology.OutputFieldsDeclarer; +import backtype.storm.tuple.Fields; +import backtype.storm.utils.Utils; + +/** + * {@link SetupOutputFieldsDeclarer} is used by {@link WrapperSetupHelper} to determine the output streams and + * number of attributes declared by the wrapped spout's or bolt's {@code declare(...)}/{@code declareStream(...)} + * method. + */ +class SetupOutputFieldsDeclarer implements OutputFieldsDeclarer { + + /** The declared output streams and schemas. */ + HashMap<String, Fields> outputStreams = new HashMap<String, Fields>(); + /** The number of attributes for each declared stream by the wrapped operator. */ + HashMap<String, Integer> outputSchemas = new HashMap<String, Integer>(); + + @Override + public void declare(final Fields fields) { + this.declareStream(Utils.DEFAULT_STREAM_ID, false, fields); + } + + @Override + public void declare(final boolean direct, final Fields fields) { + this.declareStream(Utils.DEFAULT_STREAM_ID, direct, fields); + } + + @Override + public void declareStream(final String streamId, final Fields fields) { + this.declareStream(streamId, false, fields); + } + + @Override + public void declareStream(final String streamId, final boolean direct, final Fields fields) { + if (streamId == null) { + throw new IllegalArgumentException("Stream ID cannot be null."); + } + if (direct) { + throw new UnsupportedOperationException("Direct emit is not supported by Flink"); + } + + this.outputStreams.put(streamId, fields); + this.outputSchemas.put(streamId, fields.size()); + } + +} http://git-wip-us.apache.org/repos/asf/flink/blob/4cb96708/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/SpoutCollector.java ---------------------------------------------------------------------- diff --git a/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/SpoutCollector.java b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/SpoutCollector.java new file mode 100644 index 0000000..91fc090 --- /dev/null +++ b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/SpoutCollector.java @@ -0,0 +1,81 @@ +/* + * 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.wrappers; + +import backtype.storm.spout.ISpoutOutputCollector; + +import org.apache.flink.api.java.tuple.Tuple0; +import org.apache.flink.api.java.tuple.Tuple25; +import org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext; + +import java.util.HashMap; +import java.util.List; + +/** + * A {@link SpoutCollector} is used by {@link AbstractStormSpoutWrapper} to provided an Storm + * compatible output collector to the wrapped spout. It transforms the emitted Storm tuples into + * Flink tuples and emits them via the provide {@link SourceContext} object. + */ +class SpoutCollector<OUT> extends AbstractStormCollector<OUT> implements ISpoutOutputCollector { + + /** The Flink source context object */ + private final SourceContext<OUT> flinkContext; + + /** + * Instantiates a new {@link SpoutCollector} that emits Flink tuples to the given Flink source context. If the + * number of attributes is specified as zero, any output type is supported. If the number of attributes is between 0 + * to 25, the output type is {@link Tuple0} to {@link Tuple25}, respectively. + * + * @param numberOfAttributes + * The number of attributes of the emitted tuples. + * @param flinkContext + * The Flink source context to be used. + * @throws UnsupportedOperationException + * if the specified number of attributes is greater than 25 + */ + SpoutCollector(final HashMap<String, Integer> numberOfAttributes, + final SourceContext<OUT> flinkContext) throws UnsupportedOperationException { + super(numberOfAttributes); + assert (flinkContext != null); + this.flinkContext = flinkContext; + } + + @Override + protected List<Integer> doEmit(final OUT flinkTuple) { + this.flinkContext.collect(flinkTuple); + // TODO + return null; + } + + @Override + public void reportError(final Throwable error) { + // not sure, if Flink can support this + } + + @Override + public List<Integer> emit(final String streamId, final List<Object> tuple, final Object messageId) { + return this.tansformAndEmit(streamId, tuple); + } + + + @Override + public void emitDirect(final int taskId, final String streamId, final List<Object> tuple, final Object messageId) { + throw new UnsupportedOperationException("Direct emit is not supported by Flink"); + } + +} http://git-wip-us.apache.org/repos/asf/flink/blob/4cb96708/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/SpoutWrapper.java ---------------------------------------------------------------------- diff --git a/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/SpoutWrapper.java b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/SpoutWrapper.java new file mode 100644 index 0000000..914a19d --- /dev/null +++ b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/SpoutWrapper.java @@ -0,0 +1,274 @@ +/* + * 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.wrappers; + +import java.util.Collection; +import java.util.HashMap; + +import backtype.storm.generated.StormTopology; +import backtype.storm.spout.SpoutOutputCollector; +import backtype.storm.topology.IRichSpout; + +import org.apache.flink.api.common.ExecutionConfig.GlobalJobParameters; +import org.apache.flink.api.java.tuple.Tuple0; +import org.apache.flink.api.java.tuple.Tuple1; +import org.apache.flink.api.java.tuple.Tuple25; +import org.apache.flink.storm.util.FiniteSpout; +import org.apache.flink.storm.util.StormConfig; +import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction; +import org.apache.flink.streaming.runtime.tasks.StreamingRuntimeContext; + +import com.google.common.collect.Sets; + +/** + * A {@link SpoutWrapper} wraps an {@link IRichSpout} in order to execute it within a Flink Streaming program. It + * takes the spout's output tuples and transforms them into Flink tuples of type {@code OUT} (see + * {@link SpoutCollector} for supported types).<br /> + * <br /> + * Per default, {@link SpoutWrapper} calls the wrapped spout's {@link IRichSpout#nextTuple() nextTuple()} method in + * an infinite loop.<br /> + * Alternatively, {@link SpoutWrapper} can call {@link IRichSpout#nextTuple() nextTuple()} for a finite number of + * times and terminate automatically afterwards (for finite input streams). The number of {@code nextTuple()} calls can + * be specified as a certain number of invocations or can be undefined. In the undefined case, {@link SpoutWrapper} + * terminates if no record was emitted to the output collector for the first time during a call to + * {@link IRichSpout#nextTuple() nextTuple()}.<br /> + * If the given spout implements {@link FiniteSpout} interface and {@link #numberOfInvocations} is not provided or + * is {@code null}, {@link SpoutWrapper} calls {@link IRichSpout#nextTuple() nextTuple()} method until + * {@link FiniteSpout#reachedEnd()} returns true. + */ +public final class SpoutWrapper<OUT> extends RichParallelSourceFunction<OUT> { + private static final long serialVersionUID = -218340336648247605L; + + /** Number of attributes of the spouts's output tuples per stream. */ + private final HashMap<String, Integer> numberOfAttributes; + /** The wrapped {@link IRichSpout spout}. */ + private final IRichSpout spout; + /** The wrapper of the given Flink collector. */ + private SpoutCollector<OUT> collector; + /** Indicates, if the source is still running or was canceled. */ + private volatile boolean isRunning = true; + /** The number of {@link IRichSpout#nextTuple()} calls. */ + private Integer numberOfInvocations; // do not use int -> null indicates an infinite loop + /** The original Storm topology. */ + private StormTopology stormTopology; + + /** + * Instantiates a new {@link SpoutWrapper} that calls the {@link IRichSpout#nextTuple() nextTuple()} method of + * the given {@link IRichSpout spout} in an infinite loop. The output type will be one of {@link Tuple0} to + * {@link Tuple25} depending on the spout's declared number of attributes. + * + * @param spout + * The {@link IRichSpout spout} to be used. + * @throws IllegalArgumentException + * If the number of declared output attributes is not with range [0;25]. + */ + public SpoutWrapper(final IRichSpout spout) throws IllegalArgumentException { + this(spout, (Collection<String>) null, null); + } + + /** + * Instantiates a new {@link SpoutWrapper} that calls the {@link IRichSpout#nextTuple() nextTuple()} method of + * the given {@link IRichSpout spout} a finite number of times. The output type will be one of {@link Tuple0} to + * {@link Tuple25} depending on the spout's declared number of attributes. + * + * @param spout + * The {@link IRichSpout spout} to be used. + * @param numberOfInvocations + * The number of calls to {@link IRichSpout#nextTuple()}. If value is negative, {@link SpoutWrapper} + * terminates if no tuple was emitted for the first time. If value is {@code null}, finite invocation is + * disabled. + * @throws IllegalArgumentException + * If the number of declared output attributes is not with range [0;25]. + */ + public SpoutWrapper(final IRichSpout spout, final Integer numberOfInvocations) + throws IllegalArgumentException { + this(spout, (Collection<String>) null, numberOfInvocations); + } + + /** + * Instantiates a new {@link SpoutWrapper} that calls the {@link IRichSpout#nextTuple() nextTuple()} method of + * the given {@link IRichSpout spout} in an infinite loop. The output type can be any type if parameter + * {@code rawOutput} is {@code true} and the spout's number of declared output tuples is 1. If {@code rawOutput} is + * {@code false} the output type will be one of {@link Tuple0} to {@link Tuple25} depending on the spout's declared + * number of attributes. + * + * @param spout + * The {@link IRichSpout spout} to be used. + * @param rawOutputs + * Contains stream names if a single attribute output stream, should not be of type {@link Tuple1} but be + * of a raw type. (Can be {@code null}.) + * @throws IllegalArgumentException + * If {@code rawOuput} is {@code true} and the number of declared output attributes is not 1 or if + * {@code rawOuput} is {@code false} and the number of declared output attributes is not with range + * [0;25]. + */ + public SpoutWrapper(final IRichSpout spout, final String[] rawOutputs) + throws IllegalArgumentException { + this(spout, Sets.newHashSet(rawOutputs), null); + } + + /** + * Instantiates a new {@link SpoutWrapper} that calls the {@link IRichSpout#nextTuple() nextTuple()} method of + * the given {@link IRichSpout spout} a finite number of times. The output type can be any type if parameter + * {@code rawOutput} is {@code true} and the spout's number of declared output tuples is 1. If {@code rawOutput} is + * {@code false} the output type will be one of {@link Tuple0} to {@link Tuple25} depending on the spout's declared + * number of attributes. + * + * @param spout + * The {@link IRichSpout spout} to be used. + * @param rawOutputs + * Contains stream names if a single attribute output stream, should not be of type {@link Tuple1} but be + * of a raw type. (Can be {@code null}.) + * @param numberOfInvocations + * The number of calls to {@link IRichSpout#nextTuple()}. If value is negative, {@link SpoutWrapper} + * terminates if no tuple was emitted for the first time. If value is {@code null}, finite invocation is + * disabled. + * @throws IllegalArgumentException + * If {@code rawOuput} is {@code true} and the number of declared output attributes is not 1 or if + * {@code rawOuput} is {@code false} and the number of declared output attributes is not with range + * [0;25]. + */ + public SpoutWrapper(final IRichSpout spout, final String[] rawOutputs, + final Integer numberOfInvocations) throws IllegalArgumentException { + this(spout, Sets.newHashSet(rawOutputs), numberOfInvocations); + } + + /** + * Instantiates a new {@link SpoutWrapper} that calls the {@link IRichSpout#nextTuple() nextTuple()} method of + * the given {@link IRichSpout spout} in an infinite loop. The output type can be any type if parameter + * {@code rawOutput} is {@code true} and the spout's number of declared output tuples is 1. If {@code rawOutput} is + * {@code false} the output type will be one of {@link Tuple0} to {@link Tuple25} depending on the spout's declared + * number of attributes. + * + * @param spout + * The {@link IRichSpout spout} to be used. + * @param rawOutputs + * Contains stream names if a single attribute output stream, should not be of type {@link Tuple1} but be + * of a raw type. (Can be {@code null}.) + * @throws IllegalArgumentException + * If {@code rawOuput} is {@code true} and the number of declared output attributes is not 1 or if + * {@code rawOuput} is {@code false} and the number of declared output attributes is not with range + * [0;25]. + */ + public SpoutWrapper(final IRichSpout spout, final Collection<String> rawOutputs) + throws IllegalArgumentException { + this(spout, rawOutputs, null); + } + + /** + * Instantiates a new {@link SpoutWrapper} that calls the {@link IRichSpout#nextTuple() nextTuple()} method of + * the given {@link IRichSpout spout} a finite number of times. The output type can be any type if parameter + * {@code rawOutput} is {@code true} and the spout's number of declared output tuples is 1. If {@code rawOutput} is + * {@code false} the output type will be one of {@link Tuple0} to {@link Tuple25} depending on the spout's declared + * number of attributes. + * + * @param spout + * The {@link IRichSpout spout} to be used. + * @param rawOutputs + * Contains stream names if a single attribute output stream, should not be of type {@link Tuple1} but be + * of a raw type. (Can be {@code null}.) + * @param numberOfInvocations + * The number of calls to {@link IRichSpout#nextTuple()}. If value is negative, {@link SpoutWrapper} + * terminates if no tuple was emitted for the first time. If value is {@code null}, finite invocation is + * disabled. + * @throws IllegalArgumentException + * If {@code rawOuput} is {@code true} and the number of declared output attributes is not 1 or if + * {@code rawOuput} is {@code false} and the number of declared output attributes is not with range + * [0;25]. + */ + public SpoutWrapper(final IRichSpout spout, final Collection<String> rawOutputs, + final Integer numberOfInvocations) throws IllegalArgumentException { + this.spout = spout; + this.numberOfAttributes = WrapperSetupHelper.getNumberOfAttributes(spout, rawOutputs); + this.numberOfInvocations = numberOfInvocations; + } + + /** + * Sets the original Storm topology. + * + * @param stormTopology + * The original Storm topology. + */ + public void setStormTopology(StormTopology stormTopology) { + this.stormTopology = stormTopology; + } + + @Override + public final void run(final SourceContext<OUT> ctx) throws Exception { + this.collector = new SpoutCollector<OUT>(this.numberOfAttributes, ctx); + + GlobalJobParameters config = super.getRuntimeContext().getExecutionConfig() + .getGlobalJobParameters(); + StormConfig stormConfig = new StormConfig(); + + if (config != null) { + if (config instanceof StormConfig) { + stormConfig = (StormConfig) config; + } else { + stormConfig.putAll(config.toMap()); + } + } + + this.spout.open(stormConfig, WrapperSetupHelper.createTopologyContext( + (StreamingRuntimeContext) super.getRuntimeContext(), this.spout, + this.stormTopology, stormConfig), new SpoutOutputCollector(this.collector)); + this.spout.activate(); + + if (numberOfInvocations == null) { + if (this.spout instanceof FiniteSpout) { + final FiniteSpout finiteSpout = (FiniteSpout) this.spout; + + while (this.isRunning && !finiteSpout.reachedEnd()) { + finiteSpout.nextTuple(); + } + } else { + while (this.isRunning) { + this.spout.nextTuple(); + } + } + } else { + int counter = this.numberOfInvocations; + if (counter >= 0) { + while ((--counter >= 0) && this.isRunning) { + this.spout.nextTuple(); + } + } else { + do { + this.collector.tupleEmitted = false; + this.spout.nextTuple(); + } while (this.collector.tupleEmitted && this.isRunning); + } + } + } + + /** + * {@inheritDoc} + * <p/> + * Sets the {@link #isRunning} flag to {@code false}. + */ + @Override + public void cancel() { + this.isRunning = false; + } + + @Override + public void close() throws Exception { + this.spout.close(); + } + +} http://git-wip-us.apache.org/repos/asf/flink/blob/4cb96708/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/StormTuple.java ---------------------------------------------------------------------- diff --git a/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/StormTuple.java b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/StormTuple.java new file mode 100644 index 0000000..c9ab8e5 --- /dev/null +++ b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/StormTuple.java @@ -0,0 +1,327 @@ +/* + * 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.wrappers; + +/* + * We do neither import + * backtype.storm.tuple.Tuple; + * nor + * org.apache.flink.api.java.tuple.Tuple + * to avoid confusion + */ + +import backtype.storm.generated.GlobalStreamId; +import backtype.storm.tuple.Fields; +import backtype.storm.tuple.MessageId; +import backtype.storm.tuple.Values; + +import java.lang.reflect.Field; +import java.lang.reflect.Method; +import java.util.List; + +/** + * {@link StormTuple} converts a Flink tuple of type {@code IN} into a Storm tuple. + */ +class StormTuple<IN> implements backtype.storm.tuple.Tuple { + + /** The Storm representation of the original Flink tuple */ + private final Values stormTuple; + /** The schema (ie, ordered field names) of the tuple */ + private final Fields schema; + + /** + * Create a new Storm tuple from the given Flink tuple. The provided {@code nameIndexMap} is ignored for raw input + * types. + * + * @param flinkTuple + * The Flink tuple to be converted. + * @param schema + * The schema (ie, ordered field names) of the tuple. + */ + StormTuple(final IN flinkTuple, final Fields schema) { + if (flinkTuple instanceof org.apache.flink.api.java.tuple.Tuple) { + this.schema = schema; + final org.apache.flink.api.java.tuple.Tuple t = (org.apache.flink.api.java.tuple.Tuple) flinkTuple; + + final int numberOfAttributes = t.getArity(); + this.stormTuple = new Values(); + for (int i = 0; i < numberOfAttributes; ++i) { + this.stormTuple.add(t.getField(i)); + } + } else { + this.schema = null; + this.stormTuple = new Values(flinkTuple); + } + } + + @Override + public int size() { + return this.stormTuple.size(); + } + + @Override + public boolean contains(final String field) { + if (this.schema != null) { + return this.schema.contains(field); + } + + try { + this.getPublicMemberField(field); + return true; + } catch (NoSuchFieldException f) { + try { + this.getGetterMethod(field); + return true; + } catch (Exception g) { + return false; + } + } catch (Exception e) { + return false; + } + } + + @Override + public Fields getFields() { + return this.schema; + } + + @Override + public int fieldIndex(final String field) { + return this.schema.fieldIndex(field); + } + + @Override + public List<Object> select(final Fields selector) { + return this.schema.select(selector, this.stormTuple); + } + + @Override + public Object getValue(final int i) { + return this.stormTuple.get(i); + } + + @Override + public String getString(final int i) { + return (String) this.stormTuple.get(i); + } + + @Override + public Integer getInteger(final int i) { + return (Integer) this.stormTuple.get(i); + } + + @Override + public Long getLong(final int i) { + return (Long) this.stormTuple.get(i); + } + + @Override + public Boolean getBoolean(final int i) { + return (Boolean) this.stormTuple.get(i); + } + + @Override + public Short getShort(final int i) { + return (Short) this.stormTuple.get(i); + } + + @Override + public Byte getByte(final int i) { + return (Byte) this.stormTuple.get(i); + } + + @Override + public Double getDouble(final int i) { + return (Double) this.stormTuple.get(i); + } + + @Override + public Float getFloat(final int i) { + return (Float) this.stormTuple.get(i); + } + + @Override + public byte[] getBinary(final int i) { + return (byte[]) this.stormTuple.get(i); + } + + private Field getPublicMemberField(final String field) throws Exception { + assert (this.stormTuple.size() == 1); + return this.stormTuple.get(0).getClass().getField(field); + } + + private Method getGetterMethod(final String field) throws Exception { + assert (this.stormTuple.size() == 1); + return this.stormTuple + .get(0) + .getClass() + .getMethod("get" + Character.toUpperCase(field.charAt(0)) + field.substring(1), + (Class<?>[]) null); + } + + private Object getValueByPublicMember(final String field) throws Exception { + assert (this.stormTuple.size() == 1); + return getPublicMemberField(field).get(this.stormTuple.get(0)); + } + + private Object getValueByGetter(final String field) throws Exception { + assert (this.stormTuple.size() == 1); + return getGetterMethod(field).invoke(this.stormTuple.get(0), (Object[]) null); + } + + @SuppressWarnings("unchecked") + public <T> T getValueByName(final String field) { + if (this.schema != null) { + return (T) this.getValue(this.schema.fieldIndex(field)); + } + assert (this.stormTuple.size() == 1); + + Exception e; + try { + // try public member + return (T) getValueByPublicMember(field); + } catch (NoSuchFieldException f) { + try { + // try getter-method + return (T) getValueByGetter(field); + } catch (Exception g) { + e = g; + } + } catch (Exception f) { + e = f; + } + + throw new RuntimeException("Could not access field <" + field + ">", e); + } + + @Override + public Object getValueByField(final String field) { + return getValueByName(field); + } + + @Override + public String getStringByField(final String field) { + return getValueByName(field); + } + + @Override + public Integer getIntegerByField(final String field) { + return getValueByName(field); + } + + @Override + public Long getLongByField(final String field) { + return getValueByName(field); + } + + @Override + public Boolean getBooleanByField(final String field) { + return getValueByName(field); + } + + @Override + public Short getShortByField(final String field) { + return getValueByName(field); + } + + @Override + public Byte getByteByField(final String field) { + return getValueByName(field); + } + + @Override + public Double getDoubleByField(final String field) { + return getValueByName(field); + } + + @Override + public Float getFloatByField(final String field) { + return getValueByName(field); + } + + @Override + public byte[] getBinaryByField(final String field) { + return getValueByName(field); + } + + @Override + public List<Object> getValues() { + return this.stormTuple; + } + + @Override + public GlobalStreamId getSourceGlobalStreamid() { + // not sure if Flink can support this + throw new UnsupportedOperationException(); + } + + @Override + public String getSourceComponent() { + // not sure if Flink can support this + throw new UnsupportedOperationException(); + } + + @Override + public int getSourceTask() { + // not sure if Flink can support this + throw new UnsupportedOperationException(); + } + + @Override + public String getSourceStreamId() { + // not sure if Flink can support this + throw new UnsupportedOperationException(); + } + + @Override + public MessageId getMessageId() { + // not sure if Flink can support this + throw new UnsupportedOperationException(); + } + + @Override + public int hashCode() { + final int prime = 31; + int result = 1; + result = (prime * result) + ((this.stormTuple == null) ? 0 : this.stormTuple.hashCode()); + return result; + } + + @Override + public boolean equals(final Object obj) { + if (this == obj) { + return true; + } + if (obj == null) { + return false; + } + if (this.getClass() != obj.getClass()) { + return false; + } + final StormTuple<?> other = (StormTuple<?>) obj; + if (this.stormTuple == null) { + if (other.stormTuple != null) { + return false; + } + } else if (!this.stormTuple.equals(other.stormTuple)) { + return false; + } + return true; + } + +} http://git-wip-us.apache.org/repos/asf/flink/blob/4cb96708/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/WrapperSetupHelper.java ---------------------------------------------------------------------- diff --git a/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/WrapperSetupHelper.java b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/WrapperSetupHelper.java new file mode 100644 index 0000000..d529b6a --- /dev/null +++ b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/WrapperSetupHelper.java @@ -0,0 +1,266 @@ +/* + * 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.wrappers; + +import backtype.storm.Config; +import backtype.storm.generated.Bolt; +import backtype.storm.generated.ComponentCommon; +import backtype.storm.generated.SpoutSpec; +import backtype.storm.generated.StateSpoutSpec; +import backtype.storm.generated.StormTopology; +import backtype.storm.generated.StreamInfo; +import backtype.storm.task.TopologyContext; +import backtype.storm.topology.IComponent; +import backtype.storm.topology.IRichBolt; +import backtype.storm.topology.IRichSpout; +import backtype.storm.tuple.Fields; + +import org.apache.flink.streaming.runtime.tasks.StreamingRuntimeContext; + +import clojure.lang.Atom; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Map.Entry; + +/** + * {@link WrapperSetupHelper} is an helper class used by {@link AbstractStormSpoutWrapper} and + * {@link BoltWrapper}. + */ +class WrapperSetupHelper { + + /** The configuration key for the topology name. */ + final static String TOPOLOGY_NAME = "storm.topology.name"; + + /** + * Computes the number of output attributes used by a {@link AbstractStormSpoutWrapper} or {@link BoltWrapper} + * per declared output stream. The number is {@code -1} for raw output type or a value within range [0;25] for + * output type {@link org.apache.flink.api.java.tuple.Tuple0 Tuple0} to + * {@link org.apache.flink.api.java.tuple.Tuple25 Tuple25}. + * + * @param spoutOrBolt + * The Storm {@link IRichSpout spout} or {@link IRichBolt bolt} to be used. + * @param rawOutputs + * Contains stream names if a single attribute output stream, should not be of type + * {@link org.apache.flink.api.java.tuple.Tuple1 Tuple1} but be of a raw type. (Can be {@code null}.) + * @return The number of attributes to be used for each stream. + * @throws IllegalArgumentException + * If {@code rawOuput} is {@code true} and the number of declared output attributes is not 1 or if + * {@code rawOuput} is {@code false} and the number of declared output attributes is not with range + * [0;25]. + */ + static HashMap<String, Integer> getNumberOfAttributes(final IComponent spoutOrBolt, + final Collection<String> rawOutputs) + throws IllegalArgumentException { + final SetupOutputFieldsDeclarer declarer = new SetupOutputFieldsDeclarer(); + spoutOrBolt.declareOutputFields(declarer); + + for (Entry<String, Integer> schema : declarer.outputSchemas.entrySet()) { + int declaredNumberOfAttributes = schema.getValue(); + if ((declaredNumberOfAttributes < 0) || (declaredNumberOfAttributes > 25)) { + throw new IllegalArgumentException( + "Provided bolt declares non supported number of output attributes. Must be in range [0;25] but " + + "was " + declaredNumberOfAttributes); + } + + if (rawOutputs != null && rawOutputs.contains(schema.getKey())) { + if (declaredNumberOfAttributes != 1) { + throw new IllegalArgumentException( + "Ouput type is requested to be raw type, but provided bolt declares more then one output " + + "attribute."); + } + schema.setValue(-1); + } + } + + return declarer.outputSchemas; + } + + /** Used to computed unique task IDs for a Storm topology. */ + private static int tid; + + /** + * Creates a {@link TopologyContext} for a Spout or Bolt instance (ie, Flink task / Storm executor). + * + * @param context + * The Flink runtime context. + * @param spoutOrBolt + * The Spout or Bolt this context is created for. + * @param stormTopology + * The original Storm topology. + * @param stormConfig + * The user provided configuration. + * @return The created {@link TopologyContext}. + */ + @SuppressWarnings({ "rawtypes", "unchecked" }) + static synchronized TopologyContext createTopologyContext( + final StreamingRuntimeContext context, final IComponent spoutOrBolt, + StormTopology stormTopology, Map stormConfig) { + String operatorName = context.getTaskName(); + if (operatorName.startsWith("Source: ")) { + // prefix "Source: " is inserted by Flink sources by default -- need to get rid of it here + operatorName = operatorName.substring(8); + } + final int dop = context.getNumberOfParallelSubtasks(); + + final Map<Integer, String> taskToComponents = new HashMap<Integer, String>(); + final Map<String, List<Integer>> componentToSortedTasks = new HashMap<String, List<Integer>>(); + final Map<String, Map<String, Fields>> componentToStreamToFields = new HashMap<String, Map<String, Fields>>(); + String stormId = (String) stormConfig.get(TOPOLOGY_NAME); + String codeDir = null; // not supported + String pidDir = null; // not supported + Integer taskId = null; + Integer workerPort = null; // not supported + List<Integer> workerTasks = new ArrayList<Integer>(); + final Map<String, Object> defaultResources = new HashMap<String, Object>(); + final Map<String, Object> userResources = new HashMap<String, Object>(); + final Map<String, Object> executorData = new HashMap<String, Object>(); + final Map registeredMetrics = new HashMap(); + Atom openOrPrepareWasCalled = null; + + if (stormTopology == null) { + // embedded mode + ComponentCommon common = new ComponentCommon(); + common.set_parallelism_hint(dop); + + HashMap<String, SpoutSpec> spouts = new HashMap<String, SpoutSpec>(); + HashMap<String, Bolt> bolts = new HashMap<String, Bolt>(); + if (spoutOrBolt instanceof IRichSpout) { + spouts.put(operatorName, new SpoutSpec(null, common)); + } else { + assert (spoutOrBolt instanceof IRichBolt); + bolts.put(operatorName, new Bolt(null, common)); + } + stormTopology = new StormTopology(spouts, bolts, new HashMap<String, StateSpoutSpec>()); + + taskId = context.getIndexOfThisSubtask(); + + List<Integer> sortedTasks = new ArrayList<Integer>(dop); + for (int i = 1; i <= dop; ++i) { + taskToComponents.put(i, operatorName); + sortedTasks.add(i); + } + componentToSortedTasks.put(operatorName, sortedTasks); + + SetupOutputFieldsDeclarer declarer = new SetupOutputFieldsDeclarer(); + spoutOrBolt.declareOutputFields(declarer); + componentToStreamToFields.put(operatorName, declarer.outputStreams); + } else { + // whole topology is built (ie, FlinkTopologyBuilder is used) + Map<String, SpoutSpec> spouts = stormTopology.get_spouts(); + Map<String, Bolt> bolts = stormTopology.get_bolts(); + Map<String, StateSpoutSpec> stateSpouts = stormTopology.get_state_spouts(); + + tid = 1; + + for (Entry<String, SpoutSpec> spout : spouts.entrySet()) { + Integer rc = processSingleOperator(spout.getKey(), spout.getValue().get_common(), + operatorName, context.getIndexOfThisSubtask(), dop, taskToComponents, + componentToSortedTasks, componentToStreamToFields); + if (rc != null) { + taskId = rc; + } + } + for (Entry<String, Bolt> bolt : bolts.entrySet()) { + Integer rc = processSingleOperator(bolt.getKey(), bolt.getValue().get_common(), + operatorName, context.getIndexOfThisSubtask(), dop, taskToComponents, + componentToSortedTasks, componentToStreamToFields); + if (rc != null) { + taskId = rc; + } + } + for (Entry<String, StateSpoutSpec> stateSpout : stateSpouts.entrySet()) { + Integer rc = taskId = processSingleOperator(stateSpout.getKey(), stateSpout + .getValue().get_common(), operatorName, context.getIndexOfThisSubtask(), + dop, taskToComponents, componentToSortedTasks, componentToStreamToFields); + if (rc != null) { + taskId = rc; + } + } + assert (taskId != null); + } + + if (!stormConfig.containsKey(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS)) { + stormConfig.put(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS, 30); // Storm default value + } + + return new FlinkTopologyContext(stormTopology, stormConfig, taskToComponents, + componentToSortedTasks, componentToStreamToFields, stormId, codeDir, pidDir, + taskId, workerPort, workerTasks, defaultResources, userResources, executorData, + registeredMetrics, openOrPrepareWasCalled); + } + + /** + * Sets up {@code taskToComponents}, {@code componentToSortedTasks}, and {@code componentToStreamToFields} for a + * single instance of a Spout or Bolt (ie, task or executor). Furthermore, is computes the unique task-id. + * + * @param componentId + * The ID of the Spout/Bolt in the topology. + * @param common + * The common operator object (that is all Spouts and Bolts have). + * @param operatorName + * The Flink operator name. + * @param index + * The index of the currently processed tasks with its operator. + * @param dop + * The parallelism of the operator. + * @param taskToComponents + * OUTPUT: A map from all task IDs of the topology to their component IDs. + * @param componentToSortedTasks + * OUTPUT: A map from all component IDs to their sorted list of corresponding task IDs. + * @param componentToStreamToFields + * OUTPUT: A map from all component IDs to there output streams and output fields. + * + * @return A unique task ID if the currently processed Spout or Bolt ({@code componentId}) is equal to the current + * Flink operator ({@link operatorName}) -- {@code null} otherwise. + */ + private static Integer processSingleOperator(final String componentId, + final ComponentCommon common, final String operatorName, final int index, + final int dop, final Map<Integer, String> taskToComponents, + final Map<String, List<Integer>> componentToSortedTasks, + final Map<String, Map<String, Fields>> componentToStreamToFields) { + final int parallelism_hint = common.get_parallelism_hint(); + Integer taskId = null; + + if (componentId.equals(operatorName)) { + taskId = tid + index; + } + + List<Integer> sortedTasks = new ArrayList<Integer>(dop); + for (int i = 0; i < parallelism_hint; ++i) { + taskToComponents.put(tid, componentId); + sortedTasks.add(tid); + ++tid; + } + componentToSortedTasks.put(componentId, sortedTasks); + + if (componentId.equals(operatorName)) { + } + + Map<String, Fields> outputStreams = new HashMap<String, Fields>(); + for(Entry<String, StreamInfo> outStream : common.get_streams().entrySet()) { + outputStreams.put(outStream.getKey(), new Fields(outStream.getValue().get_output_fields())); + } + componentToStreamToFields.put(componentId, outputStreams); + + return taskId; + } + +} http://git-wip-us.apache.org/repos/asf/flink/blob/4cb96708/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/api/FlinkOutputFieldsDeclarerTest.java ---------------------------------------------------------------------- diff --git a/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/api/FlinkOutputFieldsDeclarerTest.java b/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/api/FlinkOutputFieldsDeclarerTest.java new file mode 100644 index 0000000..49de476 --- /dev/null +++ b/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/api/FlinkOutputFieldsDeclarerTest.java @@ -0,0 +1,194 @@ +/* + * 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.api; + +import backtype.storm.tuple.Fields; +import backtype.storm.utils.Utils; + +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.api.java.typeutils.GenericTypeInfo; +import org.apache.flink.storm.api.FlinkOutputFieldsDeclarer; +import org.apache.flink.storm.util.AbstractTest; +import org.junit.Assert; +import org.junit.Test; + +import java.util.LinkedList; + +public class FlinkOutputFieldsDeclarerTest extends AbstractTest { + + + + @Test + public void testNull() { + Assert.assertNull(new FlinkOutputFieldsDeclarer().getOutputType(null)); + } + + @Test + public void testDeclare() { + for (int i = 0; i < 2; ++i) { // test case: simple / non-direct + for (int j = 1; j < 2; ++j) { // number of streams + for (int k = 0; k <= 25; ++k) { // number of attributes + this.runDeclareTest(i, j, k); + } + } + } + } + + @Test(expected = IllegalArgumentException.class) + public void testDeclareSimpleToManyAttributes() { + this.runDeclareTest(0, this.r.nextBoolean() ? 1 : 2, 26); + } + + @Test(expected = IllegalArgumentException.class) + public void testDeclareNonDirectToManyAttributes() { + this.runDeclareTest(1, this.r.nextBoolean() ? 1 : 2, 26); + } + + @Test(expected = IllegalArgumentException.class) + public void testDeclareDefaultStreamToManyAttributes() { + this.runDeclareTest(2, this.r.nextBoolean() ? 1 : 2, 26); + } + + @Test(expected = IllegalArgumentException.class) + public void testDeclareFullToManyAttributes() { + this.runDeclareTest(3, this.r.nextBoolean() ? 1 : 2, 26); + } + + private void runDeclareTest(final int testCase, final int numberOfStreams, + final int numberOfAttributes) { + final FlinkOutputFieldsDeclarer declarer = new FlinkOutputFieldsDeclarer(); + + String[] streams = null; + if (numberOfStreams > 1 || r.nextBoolean()) { + streams = new String[numberOfStreams]; + for (int i = 0; i < numberOfStreams; ++i) { + streams[i] = "stream" + i; + } + } + + final String[] attributes = new String[numberOfAttributes]; + for (int i = 0; i < attributes.length; ++i) { + attributes[i] = "a" + i; + } + + switch (testCase) { + case 0: + this.declareSimple(declarer, streams, attributes); + break; + default: + this.declareNonDirect(declarer, streams, attributes); + } + + if (streams == null) { + streams = new String[] { Utils.DEFAULT_STREAM_ID }; + } + + for (String stream : streams) { + final TypeInformation<?> type = declarer.getOutputType(stream); + + if (numberOfAttributes == 1) { + Assert.assertEquals(type.getClass(), GenericTypeInfo.class); + Assert.assertEquals(type.getTypeClass(), Object.class); + } else { + Assert.assertEquals(numberOfAttributes, type.getArity()); + Assert.assertTrue(type.isTupleType()); + } + } + } + + private void declareSimple(final FlinkOutputFieldsDeclarer declarer, final String[] streams, + final String[] attributes) { + + if (streams != null) { + for (String stream : streams) { + declarer.declareStream(stream, new Fields(attributes)); + } + } else { + declarer.declare(new Fields(attributes)); + } + } + + private void declareNonDirect(final FlinkOutputFieldsDeclarer declarer, final String[] streams, + final String[] attributes) { + + if (streams != null) { + for (String stream : streams) { + declarer.declareStream(stream, false, new Fields(attributes)); + } + } else { + declarer.declare(false, new Fields(attributes)); + } + } + + @Test(expected = IllegalArgumentException.class) + public void testUndeclared() { + final FlinkOutputFieldsDeclarer declarer = new FlinkOutputFieldsDeclarer(); + declarer.getOutputType("unknownStreamId"); + } + + @Test(expected = UnsupportedOperationException.class) + public void testDeclareDirect() { + new FlinkOutputFieldsDeclarer().declare(true, null); + } + + @Test(expected = UnsupportedOperationException.class) + public void testDeclareDirect2() { + new FlinkOutputFieldsDeclarer().declareStream(Utils.DEFAULT_STREAM_ID, true, 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 declarer = new FlinkOutputFieldsDeclarer(); + declarer.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 = declarer.getGroupingFieldIndexes(Utils.DEFAULT_STREAM_ID, + 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/4cb96708/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/api/FlinkTopologyBuilderTest.java ---------------------------------------------------------------------- diff --git a/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/api/FlinkTopologyBuilderTest.java b/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/api/FlinkTopologyBuilderTest.java new file mode 100644 index 0000000..e6fb8e5 --- /dev/null +++ b/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/api/FlinkTopologyBuilderTest.java @@ -0,0 +1,77 @@ +/* + * 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.api; + +import org.apache.flink.storm.api.FlinkTopologyBuilder; +import org.apache.flink.storm.util.TestDummyBolt; +import org.apache.flink.storm.util.TestDummySpout; +import org.apache.flink.storm.util.TestSink; +import org.junit.Test; + +import backtype.storm.tuple.Fields; + +public class FlinkTopologyBuilderTest { + + @Test(expected = RuntimeException.class) + public void testUnknowSpout() { + FlinkTopologyBuilder builder = new FlinkTopologyBuilder(); + builder.setSpout("spout", new TestSpout()); + builder.setBolt("bolt", new TestBolt()).shuffleGrouping("unknown"); + builder.createTopology(); + } + + @Test(expected = RuntimeException.class) + public void testUnknowBolt() { + FlinkTopologyBuilder builder = new FlinkTopologyBuilder(); + builder.setSpout("spout", new TestSpout()); + builder.setBolt("bolt1", new TestBolt()).shuffleGrouping("spout"); + builder.setBolt("bolt2", new TestBolt()).shuffleGrouping("unknown"); + builder.createTopology(); + } + + @Test(expected = RuntimeException.class) + public void testUndeclaredStream() { + FlinkTopologyBuilder builder = new FlinkTopologyBuilder(); + builder.setSpout("spout", new TestSpout()); + builder.setBolt("bolt", new TestBolt()).shuffleGrouping("spout"); + builder.createTopology(); + } + + @Test + public void testFieldsGroupingOnMultipleSpoutOutputStreams() { + FlinkTopologyBuilder flinkBuilder = new FlinkTopologyBuilder(); + + flinkBuilder.setSpout("spout", new TestDummySpout()); + flinkBuilder.setBolt("sink", new TestSink()).fieldsGrouping("spout", + TestDummySpout.spoutStreamId, new Fields("id")); + + flinkBuilder.createTopology(); + } + + @Test + public void testFieldsGroupingOnMultipleBoltOutputStreams() { + FlinkTopologyBuilder flinkBuilder = new FlinkTopologyBuilder(); + + flinkBuilder.setSpout("spout", new TestDummySpout()); + flinkBuilder.setBolt("bolt", new TestDummyBolt()).shuffleGrouping("spout"); + flinkBuilder.setBolt("sink", new TestSink()).fieldsGrouping("bolt", + TestDummyBolt.groupingStreamId, new Fields("id")); + + flinkBuilder.createTopology(); + } + +} http://git-wip-us.apache.org/repos/asf/flink/blob/4cb96708/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/api/FlinkTopologyTest.java ---------------------------------------------------------------------- diff --git a/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/api/FlinkTopologyTest.java b/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/api/FlinkTopologyTest.java new file mode 100644 index 0000000..9d04ca5 --- /dev/null +++ b/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/api/FlinkTopologyTest.java @@ -0,0 +1,63 @@ +/* + * 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.api; + +import org.apache.flink.storm.api.FlinkTopology; +import org.junit.Assert; +import org.junit.Test; + +public class FlinkTopologyTest { + + @Test + public void testDefaultParallelism() { + final FlinkTopology topology = new FlinkTopology(); + Assert.assertEquals(1, topology.getParallelism()); + } + + @Test(expected = UnsupportedOperationException.class) + public void testExecute() throws Exception { + new FlinkTopology().execute(); + } + + @Test(expected = UnsupportedOperationException.class) + public void testExecuteWithName() throws Exception { + new FlinkTopology().execute(null); + } + + @Test + public void testNumberOfTasks() { + final FlinkTopology topology = new FlinkTopology(); + + 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().increaseNumberOfTasks(0); + } + +} http://git-wip-us.apache.org/repos/asf/flink/blob/4cb96708/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/api/TestBolt.java ---------------------------------------------------------------------- diff --git a/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/api/TestBolt.java b/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/api/TestBolt.java new file mode 100644 index 0000000..74ea67e --- /dev/null +++ b/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/api/TestBolt.java @@ -0,0 +1,48 @@ +/* + * 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.api; + +import java.util.Map; + +import backtype.storm.task.OutputCollector; +import backtype.storm.task.TopologyContext; +import backtype.storm.topology.IRichBolt; +import backtype.storm.topology.OutputFieldsDeclarer; +import backtype.storm.tuple.Tuple; + +public class TestBolt implements IRichBolt { + private static final long serialVersionUID = -667148827441397683L; + + @SuppressWarnings("rawtypes") + @Override + public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {} + + @Override + public void execute(Tuple input) {} + + @Override + public void cleanup() {} + + @Override + public void declareOutputFields(OutputFieldsDeclarer declarer) {} + + @Override + public Map<String, Object> getComponentConfiguration() { + return null; + } + +} http://git-wip-us.apache.org/repos/asf/flink/blob/4cb96708/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/api/TestSpout.java ---------------------------------------------------------------------- diff --git a/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/api/TestSpout.java b/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/api/TestSpout.java new file mode 100644 index 0000000..4abb604 --- /dev/null +++ b/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/api/TestSpout.java @@ -0,0 +1,59 @@ +/* + * 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.api; + +import java.util.Map; + +import backtype.storm.spout.SpoutOutputCollector; +import backtype.storm.task.TopologyContext; +import backtype.storm.topology.IRichSpout; +import backtype.storm.topology.OutputFieldsDeclarer; + +public class TestSpout implements IRichSpout { + private static final long serialVersionUID = -4884029383198924007L; + + @SuppressWarnings("rawtypes") + @Override + public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {} + + @Override + public void close() {} + + @Override + public void activate() {} + + @Override + public void deactivate() {} + + @Override + public void nextTuple() {} + + @Override + public void ack(Object msgId) {} + + @Override + public void fail(Object msgId) {} + + @Override + public void declareOutputFields(OutputFieldsDeclarer declarer) {} + + @Override + public Map<String, Object> getComponentConfiguration() { + return null; + } + +} http://git-wip-us.apache.org/repos/asf/flink/blob/4cb96708/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/api/TestTopologyBuilder.java ---------------------------------------------------------------------- diff --git a/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/api/TestTopologyBuilder.java b/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/api/TestTopologyBuilder.java new file mode 100644 index 0000000..7bea94c --- /dev/null +++ b/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/api/TestTopologyBuilder.java @@ -0,0 +1,29 @@ +/* + * 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.api; + +import org.apache.flink.storm.api.FlinkTopologyBuilder; + +import backtype.storm.generated.StormTopology; + +public class TestTopologyBuilder extends FlinkTopologyBuilder { + @Override + public StormTopology getStormTopology() { + return super.getStormTopology(); + } +}
