flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From mj...@apache.org
Subject [11/15] flink git commit: [Storm Compatibility] Maven module restucturing and cleanup - removed storm-parent; renamed storm-core and storm-examples - updated internal Java package structure * renamed package "stormcompatibility" to "storm" *
Date Tue, 06 Oct 2015 11:31:36 GMT
http://git-wip-us.apache.org/repos/asf/flink/blob/4cb96708/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/main/java/org/apache/flink/stormcompatibility/wrappers/StormSpoutWrapper.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/main/java/org/apache/flink/stormcompatibility/wrappers/StormSpoutWrapper.java b/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/main/java/org/apache/flink/stormcompatibility/wrappers/StormSpoutWrapper.java
deleted file mode 100644
index 300b241..0000000
--- a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/main/java/org/apache/flink/stormcompatibility/wrappers/StormSpoutWrapper.java
+++ /dev/null
@@ -1,105 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.stormcompatibility.wrappers;
-
-import java.util.Collection;
-
-import backtype.storm.topology.IRichSpout;
-
-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 com.google.common.collect.Sets;
-
-/**
- * A {@link StormSpoutWrapper} is an {@link AbstractStormSpoutWrapper} that calls the wrapped spout's
- * {@link IRichSpout#nextTuple() nextTuple()} method in in infinite loop.
- */
-public class StormSpoutWrapper<OUT> extends AbstractStormSpoutWrapper<OUT> {
-	private static final long serialVersionUID = -218340336648247605L;
-
-	/**
-	 * Instantiates a new {@link StormSpoutWrapper} that wraps the given Storm {@link IRichSpout spout} such that it can
-	 * be used within a Flink streaming program. The output type will be one of {@link Tuple0} to {@link Tuple25}
-	 * depending on the spout's declared number of attributes.
-	 * 
-	 * @param spout
-	 *            The Storm {@link IRichSpout spout} to be used.
-	 * @throws IllegalArgumentException
-	 *             If the number of declared output attributes is not with range [0;25].
-	 */
-	public StormSpoutWrapper(final IRichSpout spout) throws IllegalArgumentException {
-		super(spout, null);
-	}
-
-	/**
-	 * Instantiates a new {@link StormSpoutWrapper} that wraps the given Storm {@link IRichSpout spout} such that it can
-	 * be used within a Flink streaming program. 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 Storm {@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 StormSpoutWrapper(final IRichSpout spout, final String[] rawOutputs)
-			throws IllegalArgumentException {
-		super(spout, Sets.newHashSet(rawOutputs));
-	}
-
-	/**
-	 * Instantiates a new {@link StormSpoutWrapper} that wraps the given Storm {@link IRichSpout spout} such that it can
-	 * be used within a Flink streaming program. 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 Storm {@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 StormSpoutWrapper(final IRichSpout spout, final Collection<String> rawOutputs)
-			throws IllegalArgumentException {
-		super(spout, rawOutputs);
-	}
-
-	/**
-	 * Calls {@link IRichSpout#nextTuple()} in an infinite loop until {@link #cancel()} is called.
-	 */
-	@Override
-	protected void execute() {
-		while (super.isRunning) {
-			super.spout.nextTuple();
-		}
-	}
-
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/4cb96708/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/main/java/org/apache/flink/stormcompatibility/wrappers/StormTuple.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/main/java/org/apache/flink/stormcompatibility/wrappers/StormTuple.java b/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/main/java/org/apache/flink/stormcompatibility/wrappers/StormTuple.java
deleted file mode 100644
index 07d94b4..0000000
--- a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/main/java/org/apache/flink/stormcompatibility/wrappers/StormTuple.java
+++ /dev/null
@@ -1,327 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.stormcompatibility.wrappers;
-
-/*
- * 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.
-	 */
-	public 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-compatibility/flink-storm-compatibility-core/src/main/java/org/apache/flink/stormcompatibility/wrappers/StormWrapperSetupHelper.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/main/java/org/apache/flink/stormcompatibility/wrappers/StormWrapperSetupHelper.java b/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/main/java/org/apache/flink/stormcompatibility/wrappers/StormWrapperSetupHelper.java
deleted file mode 100644
index 891497e..0000000
--- a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/main/java/org/apache/flink/stormcompatibility/wrappers/StormWrapperSetupHelper.java
+++ /dev/null
@@ -1,268 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.stormcompatibility.wrappers;
-
-import 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.stormcompatibility.util.FlinkOutputFieldsDeclarer;
-import org.apache.flink.stormcompatibility.util.FlinkTopologyContext;
-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 StormWrapperSetupHelper} is an helper class used by {@link AbstractStormSpoutWrapper} and
- * {@link StormBoltWrapper}.
- */
-class StormWrapperSetupHelper {
-
-	/** 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 StormBoltWrapper}
-	 * 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].
-	 */
-	public 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" })
-	public 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);
-
-			FlinkOutputFieldsDeclarer declarer = new FlinkOutputFieldsDeclarer();
-			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-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/api/FlinkTopologyBuilderTest.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/api/FlinkTopologyBuilderTest.java b/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/api/FlinkTopologyBuilderTest.java
deleted file mode 100644
index 0dd9b1c..0000000
--- a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/api/FlinkTopologyBuilderTest.java
+++ /dev/null
@@ -1,76 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.stormcompatibility.api;
-
-import org.apache.flink.stormcompatibility.util.TestDummyBolt;
-import org.apache.flink.stormcompatibility.util.TestDummySpout;
-import org.apache.flink.stormcompatibility.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-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/api/FlinkTopologyTest.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/api/FlinkTopologyTest.java b/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/api/FlinkTopologyTest.java
deleted file mode 100644
index c98c9a3..0000000
--- a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/api/FlinkTopologyTest.java
+++ /dev/null
@@ -1,62 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.stormcompatibility.api;
-
-import org.junit.Assert;
-import org.junit.Test;
-
-public class FlinkTopologyTest {
-
-	@Test
-	public void testDefaultParallelism() {
-		final FlinkTopology topology = new FlinkTopology();
-		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-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/api/TestBolt.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/api/TestBolt.java b/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/api/TestBolt.java
deleted file mode 100644
index 2e4a534..0000000
--- a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/api/TestBolt.java
+++ /dev/null
@@ -1,48 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.stormcompatibility.api;
-
-import java.util.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-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/api/TestSpout.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/api/TestSpout.java b/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/api/TestSpout.java
deleted file mode 100644
index 146218f..0000000
--- a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/api/TestSpout.java
+++ /dev/null
@@ -1,59 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.stormcompatibility.api;
-
-import java.util.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-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/api/TestTopologyBuilder.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/api/TestTopologyBuilder.java b/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/api/TestTopologyBuilder.java
deleted file mode 100644
index f664e58..0000000
--- a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/api/TestTopologyBuilder.java
+++ /dev/null
@@ -1,27 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.stormcompatibility.api;
-
-import backtype.storm.generated.StormTopology;
-
-public class TestTopologyBuilder extends FlinkTopologyBuilder {
-	@Override
-	public StormTopology getStormTopology() {
-		return super.getStormTopology();
-	}
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/4cb96708/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/util/AbstractTest.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/util/AbstractTest.java b/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/util/AbstractTest.java
deleted file mode 100644
index 94a50cf..0000000
--- a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/util/AbstractTest.java
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.stormcompatibility.util;
-
-import org.junit.Before;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.Random;
-
-public abstract class AbstractTest {
-	private static final Logger LOG = LoggerFactory.getLogger(AbstractTest.class);
-
-	protected long seed;
-	protected Random r;
-
-	@Before
-	public void prepare() {
-		this.seed = System.currentTimeMillis();
-		this.r = new Random(this.seed);
-		LOG.info("Test seed: {}", new Long(this.seed));
-	}
-
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/4cb96708/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/util/FiniteTestSpout.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/util/FiniteTestSpout.java b/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/util/FiniteTestSpout.java
deleted file mode 100644
index 1891873..0000000
--- a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/util/FiniteTestSpout.java
+++ /dev/null
@@ -1,77 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.stormcompatibility.util;
-
-import backtype.storm.spout.SpoutOutputCollector;
-import backtype.storm.task.TopologyContext;
-import backtype.storm.topology.IRichSpout;
-import backtype.storm.topology.OutputFieldsDeclarer;
-import backtype.storm.tuple.Fields;
-import backtype.storm.tuple.Values;
-
-import java.util.Map;
-
-public class FiniteTestSpout implements IRichSpout {
-	private static final long serialVersionUID = 7992419478267824279L;
-
-	private int numberOfOutputTuples;
-	private SpoutOutputCollector collector;
-
-	public FiniteTestSpout(final int numberOfOutputTuples) {
-		this.numberOfOutputTuples = numberOfOutputTuples;
-	}
-
-	@SuppressWarnings("rawtypes")
-	@Override
-	public void open(final Map conf, final TopologyContext context, final SpoutOutputCollector collector) {
-		this.collector = collector;
-	}
-
-	@Override
-	public void close() {/* nothing to do */}
-
-	@Override
-	public void activate() {/* nothing to do */}
-
-	@Override
-	public void deactivate() {/* nothing to do */}
-
-	@Override
-	public void nextTuple() {
-		if (--this.numberOfOutputTuples >= 0) {
-			this.collector.emit(new Values(new Integer(this.numberOfOutputTuples)));
-		}
-	}
-
-	@Override
-	public void ack(final Object msgId) {/* nothing to do */}
-
-	@Override
-	public void fail(final Object msgId) {/* nothing to do */}
-
-	@Override
-	public void declareOutputFields(final OutputFieldsDeclarer declarer) {
-		declarer.declare(new Fields("dummy"));
-	}
-
-	@Override
-	public Map<String, Object> getComponentConfiguration() {
-		return null;
-	}
-
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/4cb96708/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/util/FlinkOutputFieldsDeclarerTest.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/util/FlinkOutputFieldsDeclarerTest.java b/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/util/FlinkOutputFieldsDeclarerTest.java
deleted file mode 100644
index 8e63563..0000000
--- a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/util/FlinkOutputFieldsDeclarerTest.java
+++ /dev/null
@@ -1,193 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.stormcompatibility.util;
-
-import 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.stormcompatibility.util.FlinkOutputFieldsDeclarer;
-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-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/util/FlinkStormStreamSelectorTest.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/util/FlinkStormStreamSelectorTest.java b/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/util/FlinkStormStreamSelectorTest.java
deleted file mode 100644
index c3cb7d7..0000000
--- a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/util/FlinkStormStreamSelectorTest.java
+++ /dev/null
@@ -1,51 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.stormcompatibility.util;
-
-import java.util.Iterator;
-
-import org.apache.flink.stormcompatibility.util.FlinkStormStreamSelector;
-import org.apache.flink.stormcompatibility.util.SplitStreamType;
-import org.junit.Assert;
-import org.junit.Test;
-
-public class FlinkStormStreamSelectorTest {
-
-	@Test
-	public void testSelector() {
-		FlinkStormStreamSelector<Object> selector = new FlinkStormStreamSelector<Object>();
-		SplitStreamType<Object> tuple = new SplitStreamType<Object>();
-		Iterator<String> result;
-
-		tuple.streamId = "stream1";
-		result = selector.select(tuple).iterator();
-		Assert.assertEquals("stream1", result.next());
-		Assert.assertFalse(result.hasNext());
-
-		tuple.streamId = "stream2";
-		result = selector.select(tuple).iterator();
-		Assert.assertEquals("stream2", result.next());
-		Assert.assertFalse(result.hasNext());
-
-		tuple.streamId = "stream1";
-		result = selector.select(tuple).iterator();
-		Assert.assertEquals("stream1", result.next());
-		Assert.assertFalse(result.hasNext());
-	}
-
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/4cb96708/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/util/FlinkTopologyContextTest.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/util/FlinkTopologyContextTest.java b/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/util/FlinkTopologyContextTest.java
deleted file mode 100644
index bd9ea3f..0000000
--- a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/util/FlinkTopologyContextTest.java
+++ /dev/null
@@ -1,114 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.stormcompatibility.util;
-
-import java.util.HashMap;
-
-import backtype.storm.generated.Bolt;
-import backtype.storm.generated.SpoutSpec;
-import backtype.storm.generated.StateSpoutSpec;
-import backtype.storm.generated.StormTopology;
-import backtype.storm.metric.api.ICombiner;
-import backtype.storm.metric.api.IMetric;
-import backtype.storm.metric.api.IReducer;
-
-import org.apache.flink.stormcompatibility.util.FlinkTopologyContext;
-import org.junit.Test;
-
-
-/*
- * FlinkTopologyContext.getSources(componentId) and FlinkTopologyContext.getTargets(componentId) are not tested here,
- * because those are tested in StormWrapperSetupHelperTest.
- */
-public class FlinkTopologyContextTest extends AbstractTest {
-
-	@Test(expected = UnsupportedOperationException.class)
-	public void testAddTaskHook() {
-		new FlinkTopologyContext(new StormTopology(new HashMap<String, SpoutSpec>(),
-				new HashMap<String, Bolt>(), new HashMap<String, StateSpoutSpec>()), null, null,
-				null, null, null, null, null, null, null, null, null, null, null, null, null)
-		.addTaskHook(null);
-	}
-
-	@Test(expected = UnsupportedOperationException.class)
-	public void testGetHooks() {
-		new FlinkTopologyContext(new StormTopology(new HashMap<String, SpoutSpec>(),
-				new HashMap<String, Bolt>(), new HashMap<String, StateSpoutSpec>()), null, null,
-				null, null, null, null, null, null, null, null, null, null, null, null, null)
-		.getHooks();
-	}
-
-	@SuppressWarnings("rawtypes")
-	@Test(expected = UnsupportedOperationException.class)
-	public void testRegisteredMetric1() {
-		new FlinkTopologyContext(new StormTopology(new HashMap<String, SpoutSpec>(),
-				new HashMap<String, Bolt>(), new HashMap<String, StateSpoutSpec>()), null, null,
-				null, null, null, null, null, null, null, null, null, null, null, null, null)
-		.registerMetric(null, (ICombiner) null, 0);
-	}
-
-	@SuppressWarnings("rawtypes")
-	@Test(expected = UnsupportedOperationException.class)
-	public void testRegisteredMetric2() {
-		new FlinkTopologyContext(new StormTopology(new HashMap<String, SpoutSpec>(),
-				new HashMap<String, Bolt>(), new HashMap<String, StateSpoutSpec>()), null, null,
-				null, null, null, null, null, null, null, null, null, null, null, null, null)
-		.registerMetric(null, (IReducer) null, 0);
-	}
-
-	@Test(expected = UnsupportedOperationException.class)
-	public void testRegisteredMetric3() {
-		new FlinkTopologyContext(new StormTopology(new HashMap<String, SpoutSpec>(),
-				new HashMap<String, Bolt>(), new HashMap<String, StateSpoutSpec>()), null, null,
-				null, null, null, null, null, null, null, null, null, null, null, null, null)
-		.registerMetric(null, (IMetric) null, 0);
-	}
-
-	@Test(expected = UnsupportedOperationException.class)
-	public void testGetRegisteredMetricByName() {
-		new FlinkTopologyContext(new StormTopology(new HashMap<String, SpoutSpec>(),
-				new HashMap<String, Bolt>(), new HashMap<String, StateSpoutSpec>()), null, null,
-				null, null, null, null, null, null, null, null, null, null, null, null, null)
-		.getRegisteredMetricByName(null);
-	}
-
-	@Test(expected = UnsupportedOperationException.class)
-	public void testSetAllSubscribedState() {
-		new FlinkTopologyContext(new StormTopology(new HashMap<String, SpoutSpec>(),
-				new HashMap<String, Bolt>(), new HashMap<String, StateSpoutSpec>()), null, null,
-				null, null, null, null, null, null, null, null, null, null, null, null, null)
-		.setAllSubscribedState(null);
-	}
-
-	@Test(expected = UnsupportedOperationException.class)
-	public void testSetSubscribedState1() {
-		new FlinkTopologyContext(new StormTopology(new HashMap<String, SpoutSpec>(),
-				new HashMap<String, Bolt>(), new HashMap<String, StateSpoutSpec>()), null, null,
-				null, null, null, null, null, null, null, null, null, null, null, null, null)
-		.setSubscribedState(null, null);
-	}
-
-	@Test(expected = UnsupportedOperationException.class)
-	public void testSetSubscribedState2() {
-		new FlinkTopologyContext(new StormTopology(new HashMap<String, SpoutSpec>(),
-				new HashMap<String, Bolt>(), new HashMap<String, StateSpoutSpec>()), null, null,
-				null, null, null, null, null, null, null, null, null, null, null, null, null)
-		.setSubscribedState(null, null, null);
-	}
-
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/4cb96708/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/util/TestDummyBolt.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/util/TestDummyBolt.java b/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/util/TestDummyBolt.java
deleted file mode 100644
index b499373..0000000
--- a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/util/TestDummyBolt.java
+++ /dev/null
@@ -1,71 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.stormcompatibility.util;
-
-import java.util.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.Fields;
-import backtype.storm.tuple.Tuple;
-import backtype.storm.tuple.Values;
-
-public class TestDummyBolt implements IRichBolt {
-	private static final long serialVersionUID = 6893611247443121322L;
-
-	public final static String shuffleStreamId = "shuffleStream";
-	public final static String groupingStreamId = "groupingStream";
-
-	private boolean emit = true;
-	private TopologyContext context;
-	private OutputCollector collector;
-
-	@SuppressWarnings("rawtypes")
-	@Override
-	public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
-		this.context = context;
-		this.collector = collector;
-	}
-
-	@Override
-	public void execute(Tuple input) {
-		if (this.context.getThisTaskIndex() == 0) {
-			this.collector.emit(shuffleStreamId, input.getValues());
-		}
-		if (this.emit) {
-			this.collector.emit(groupingStreamId, new Values("bolt", this.context));
-			this.emit = false;
-		}
-	}
-
-	@Override
-	public void cleanup() {}
-
-	@Override
-	public void declareOutputFields(OutputFieldsDeclarer declarer) {
-		declarer.declareStream(shuffleStreamId, new Fields("data"));
-		declarer.declareStream(groupingStreamId, new Fields("id", "data"));
-	}
-
-	@Override
-	public Map<String, Object> getComponentConfiguration() {
-		return null;
-	}
-
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/4cb96708/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/util/TestDummySpout.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/util/TestDummySpout.java b/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/util/TestDummySpout.java
deleted file mode 100644
index 345ca12..0000000
--- a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/util/TestDummySpout.java
+++ /dev/null
@@ -1,79 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.stormcompatibility.util;
-
-import java.util.Map;
-
-import backtype.storm.spout.SpoutOutputCollector;
-import backtype.storm.task.TopologyContext;
-import backtype.storm.topology.IRichSpout;
-import backtype.storm.topology.OutputFieldsDeclarer;
-import backtype.storm.tuple.Fields;
-import backtype.storm.tuple.Values;
-import backtype.storm.utils.Utils;
-
-public class TestDummySpout implements IRichSpout {
-	private static final long serialVersionUID = -5190945609124603118L;
-
-	public final static String spoutStreamId = "spout-stream";
-
-	private boolean emit = true;
-	private TopologyContext context;
-	private SpoutOutputCollector collector;
-
-	@SuppressWarnings("rawtypes")
-	@Override
-	public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
-		this.context = context;
-		this.collector = collector;
-	}
-
-	@Override
-	public void close() {}
-
-	@Override
-	public void activate() {}
-
-	@Override
-	public void deactivate() {}
-
-	@Override
-	public void nextTuple() {
-		if (this.emit) {
-			this.collector.emit(new Values(this.context));
-			this.emit = false;
-		}
-	}
-
-	@Override
-	public void ack(Object msgId) {}
-
-	@Override
-	public void fail(Object msgId) {}
-
-	@Override
-	public void declareOutputFields(OutputFieldsDeclarer declarer) {
-		declarer.declareStream(Utils.DEFAULT_STREAM_ID, new Fields("data"));
-		declarer.declareStream(spoutStreamId, new Fields("id", "data"));
-	}
-
-	@Override
-	public Map<String, Object> getComponentConfiguration() {
-		return null;
-	}
-
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/4cb96708/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/util/TestSink.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/util/TestSink.java b/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/util/TestSink.java
deleted file mode 100644
index c8e5584..0000000
--- a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/util/TestSink.java
+++ /dev/null
@@ -1,60 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.stormcompatibility.util;
-
-import java.util.LinkedList;
-import java.util.List;
-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 TestSink implements IRichBolt {
-	private static final long serialVersionUID = 4314871456719370877L;
-
-	public final static List<TopologyContext> result = new LinkedList<TopologyContext>();
-
-	@SuppressWarnings("rawtypes")
-	@Override
-	public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
-		result.add(context);
-	}
-
-	@Override
-	public void execute(Tuple input) {
-		if (input.size() == 1) {
-			result.add((TopologyContext) input.getValue(0));
-		} else {
-			result.add((TopologyContext) input.getValue(1));
-		}
-	}
-
-	@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-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/wrappers/FiniteStormSpoutWrapperTest.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/wrappers/FiniteStormSpoutWrapperTest.java b/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/wrappers/FiniteStormSpoutWrapperTest.java
deleted file mode 100644
index b44e8a1..0000000
--- a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/wrappers/FiniteStormSpoutWrapperTest.java
+++ /dev/null
@@ -1,76 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.stormcompatibility.wrappers;
-
-import org.apache.flink.api.common.ExecutionConfig;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.stormcompatibility.util.FiniteStormSpout;
-import org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext;
-import org.apache.flink.streaming.runtime.tasks.StreamingRuntimeContext;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.powermock.core.classloader.annotations.PrepareForTest;
-import org.powermock.modules.junit4.PowerMockRunner;
-
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.never;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-
-@RunWith(PowerMockRunner.class)
-@PrepareForTest(StormWrapperSetupHelper.class)
-public class FiniteStormSpoutWrapperTest {
-
-	@SuppressWarnings("unchecked")
-	@Test
-	public void runAndExecuteTest1() throws Exception {
-		final FiniteStormSpout stormSpout = mock(FiniteStormSpout.class);
-		when(stormSpout.reachedEnd()).thenReturn(false, false, false, true, false, false, true);
-
-		final StreamingRuntimeContext taskContext = mock(StreamingRuntimeContext.class);
-		when(taskContext.getExecutionConfig()).thenReturn(new ExecutionConfig());
-		when(taskContext.getTaskStubParameters()).thenReturn(new Configuration());
-		when(taskContext.getTaskName()).thenReturn("name");
-
-		final FiniteStormSpoutWrapper<?> wrapper = new FiniteStormSpoutWrapper<Object>(stormSpout);
-		wrapper.setRuntimeContext(taskContext);
-
-		wrapper.run(mock(SourceContext.class));
-		verify(stormSpout, times(3)).nextTuple();
-	}
-
-	@SuppressWarnings("unchecked")
-	@Test
-	public void runAndExecuteTest2() throws Exception {
-		final FiniteStormSpout stormSpout = mock(FiniteStormSpout.class);
-		when(stormSpout.reachedEnd()).thenReturn(true, false, true, false, true, false, true);
-
-		final StreamingRuntimeContext taskContext = mock(StreamingRuntimeContext.class);
-		when(taskContext.getExecutionConfig()).thenReturn(new ExecutionConfig());
-		when(taskContext.getTaskStubParameters()).thenReturn(new Configuration());
-		when(taskContext.getTaskName()).thenReturn("name");
-
-		final FiniteStormSpoutWrapper<?> wrapper = new FiniteStormSpoutWrapper<Object>(stormSpout);
-		wrapper.setRuntimeContext(taskContext);
-
-		wrapper.run(mock(SourceContext.class));
-		verify(stormSpout, never()).nextTuple();
-	}
-
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/4cb96708/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/wrappers/SetupOutputFieldsDeclarerTest.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/wrappers/SetupOutputFieldsDeclarerTest.java b/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/wrappers/SetupOutputFieldsDeclarerTest.java
deleted file mode 100644
index 738eb1e..0000000
--- a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/wrappers/SetupOutputFieldsDeclarerTest.java
+++ /dev/null
@@ -1,91 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.stormcompatibility.wrappers;
-
-import backtype.storm.tuple.Fields;
-import backtype.storm.utils.Utils;
-
-import org.apache.flink.stormcompatibility.util.AbstractTest;
-import org.junit.Assert;
-import org.junit.Test;
-
-import java.util.ArrayList;
-
-public class SetupOutputFieldsDeclarerTest extends AbstractTest {
-
-	@Test
-	public void testDeclare() {
-		final SetupOutputFieldsDeclarer declarer = new SetupOutputFieldsDeclarer();
-
-		int numberOfAttributes = this.r.nextInt(26);
-		declarer.declare(createSchema(numberOfAttributes));
-		Assert.assertEquals(1, declarer.outputSchemas.size());
-		Assert.assertEquals(numberOfAttributes, declarer.outputSchemas.get(Utils.DEFAULT_STREAM_ID)
-				.intValue());
-
-		final String sid = "streamId";
-		numberOfAttributes = this.r.nextInt(26);
-		declarer.declareStream(sid, createSchema(numberOfAttributes));
-		Assert.assertEquals(2, declarer.outputSchemas.size());
-		Assert.assertEquals(numberOfAttributes, declarer.outputSchemas.get(sid).intValue());
-	}
-
-	private Fields createSchema(final int numberOfAttributes) {
-		final ArrayList<String> schema = new ArrayList<String>(numberOfAttributes);
-		for (int i = 0; i < numberOfAttributes; ++i) {
-			schema.add("a" + i);
-		}
-		return new Fields(schema);
-	}
-
-	@Test
-	public void testDeclareDirect() {
-		new SetupOutputFieldsDeclarer().declare(false, new Fields());
-	}
-
-	@Test(expected = UnsupportedOperationException.class)
-	public void testDeclareDirectFail() {
-		new SetupOutputFieldsDeclarer().declare(true, new Fields());
-	}
-
-	@Test
-	public void testDeclareStream() {
-		new SetupOutputFieldsDeclarer().declareStream(Utils.DEFAULT_STREAM_ID, new Fields());
-	}
-
-	@Test(expected = IllegalArgumentException.class)
-	public void testDeclareStreamFail() {
-		new SetupOutputFieldsDeclarer().declareStream(null, new Fields());
-	}
-
-	@Test
-	public void testDeclareFullStream() {
-		new SetupOutputFieldsDeclarer().declareStream(Utils.DEFAULT_STREAM_ID, false, new Fields());
-	}
-
-	@Test(expected = IllegalArgumentException.class)
-	public void testDeclareFullStreamFailNonDefaultStream() {
-		new SetupOutputFieldsDeclarer().declareStream(null, false, new Fields());
-	}
-
-	@Test(expected = UnsupportedOperationException.class)
-	public void testDeclareFullStreamFailDirect() {
-		new SetupOutputFieldsDeclarer().declareStream(Utils.DEFAULT_STREAM_ID, true, new Fields());
-	}
-
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/4cb96708/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/wrappers/StormBoltCollectorTest.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/wrappers/StormBoltCollectorTest.java b/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/wrappers/StormBoltCollectorTest.java
deleted file mode 100644
index d01c3e0..0000000
--- a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/wrappers/StormBoltCollectorTest.java
+++ /dev/null
@@ -1,105 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.stormcompatibility.wrappers;
-
-import backtype.storm.tuple.Values;
-
-import org.apache.flink.api.java.tuple.Tuple;
-import org.apache.flink.stormcompatibility.util.AbstractTest;
-import org.apache.flink.streaming.api.operators.Output;
-import org.junit.Assert;
-import org.junit.Test;
-
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.List;
-
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.verify;
-
-public class StormBoltCollectorTest extends AbstractTest {
-
-	@SuppressWarnings({ "rawtypes", "unchecked" })
-	@Test
-	public void testBoltStormCollector() throws InstantiationException, IllegalAccessException {
-		for (int numberOfAttributes = -1; numberOfAttributes < 26; ++numberOfAttributes) {
-			final Output flinkCollector = mock(Output.class);
-			Tuple flinkTuple = null;
-			final Values tuple = new Values();
-
-			StormBoltCollector<?> collector;
-
-			final String streamId = "streamId";
-			HashMap<String, Integer> attributes = new HashMap<String, Integer>();
-			attributes.put(streamId, numberOfAttributes);
-
-			if (numberOfAttributes == -1) {
-				collector = new StormBoltCollector(attributes, flinkCollector);
-				tuple.add(new Integer(this.r.nextInt()));
-
-			} else {
-				collector = new StormBoltCollector(attributes, flinkCollector);
-				flinkTuple = Tuple.getTupleClass(numberOfAttributes).newInstance();
-
-				for (int i = 0; i < numberOfAttributes; ++i) {
-					tuple.add(new Integer(this.r.nextInt()));
-					flinkTuple.setField(tuple.get(i), i);
-				}
-			}
-
-			final Collection anchors = mock(Collection.class);
-			final List<Integer> taskIds;
-			taskIds = collector.emit(streamId, anchors, tuple);
-
-			Assert.assertNull(taskIds);
-
-			if (numberOfAttributes == -1) {
-				verify(flinkCollector).collect(tuple.get(0));
-			} else {
-				verify(flinkCollector).collect(flinkTuple);
-			}
-		}
-	}
-
-
-	@SuppressWarnings("unchecked")
-	@Test(expected = UnsupportedOperationException.class)
-	public void testReportError() {
-		new StormBoltCollector<Object>(mock(HashMap.class), mock(Output.class)).reportError(null);
-	}
-
-	@SuppressWarnings("unchecked")
-	@Test(expected = UnsupportedOperationException.class)
-	public void testEmitDirect() {
-		new StormBoltCollector<Object>(mock(HashMap.class), mock(Output.class)).emitDirect(0, null,
-				null, null);
-	}
-
-	@SuppressWarnings("unchecked")
-	@Test(expected = UnsupportedOperationException.class)
-	public void testAck() {
-		new StormBoltCollector<Object>(mock(HashMap.class), mock(Output.class)).ack(null);
-	}
-
-	@SuppressWarnings("unchecked")
-	@Test(expected = UnsupportedOperationException.class)
-	public void testFail() {
-		new StormBoltCollector<Object>(mock(HashMap.class), mock(Output.class)).fail(null);
-	}
-
-}


Mime
View raw message