test_class.file: flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/wrappers/WrapperSetupHelperTest.java
test_class.identifier: WrapperSetupHelperTest
test_class.superclass: extends AbstractTest
test_case.body: @Test
  	public void testCreateTopologyContext() {
  		HashMap<String, Integer> dops = new HashMap<String, Integer>();
  		dops.put("spout1", 1);
  		dops.put("spout2", 3);
  		dops.put("bolt1", 1);
  		dops.put("bolt2", 2);
  		dops.put("sink", 1);
  
  		HashMap<String, Integer> taskCounter = new HashMap<String, Integer>();
  		taskCounter.put("spout1", 0);
  		taskCounter.put("spout2", 0);
  		taskCounter.put("bolt1", 0);
  		taskCounter.put("bolt2", 0);
  		taskCounter.put("sink", 0);
  
  		HashMap<String, IComponent> operators = new HashMap<String, IComponent>();
  		operators.put("spout1", new TestDummySpout());
  		operators.put("spout2", new TestDummySpout());
  		operators.put("bolt1", new TestDummyBolt());
  		operators.put("bolt2", new TestDummyBolt());
  		operators.put("sink", new TestSink());
  
  		TopologyBuilder builder = new TopologyBuilder();
  
  		builder.setSpout("spout1", (IRichSpout) operators.get("spout1"), dops.get("spout1"));
  		builder.setSpout("spout2", (IRichSpout) operators.get("spout2"), dops.get("spout2"));
  		builder.setBolt("bolt1", (IRichBolt) operators.get("bolt1"), dops.get("bolt1")).shuffleGrouping("spout1");
  		builder.setBolt("bolt2", (IRichBolt) operators.get("bolt2"), dops.get("bolt2")).allGrouping("spout2");
  		builder.setBolt("sink", (IRichBolt) operators.get("sink"), dops.get("sink"))
  				.shuffleGrouping("bolt1", TestDummyBolt.groupingStreamId)
  				.shuffleGrouping("bolt1", TestDummyBolt.shuffleStreamId)
  				.shuffleGrouping("bolt2", TestDummyBolt.groupingStreamId)
  				.shuffleGrouping("bolt2", TestDummyBolt.shuffleStreamId);
  
  		LocalCluster cluster = new LocalCluster();
  		Config c = new Config();
  		c.setNumAckers(0);
  		cluster.submitTopology("test", c, builder.createTopology());
  
  		while (TestSink.result.size() != 8) {
  			Utils.sleep(100);
  		}
  		cluster.shutdown();
  
  		final FlinkTopology flinkBuilder = FlinkTopology.createTopology(builder);
  		StormTopology stormTopology = flinkBuilder.getStormTopology();
  
  		Set<Integer> taskIds = new HashSet<Integer>();
  
  		for (TopologyContext expectedContext : TestSink.result) {
  			final String thisComponentId = expectedContext.getThisComponentId();
  			int index = taskCounter.get(thisComponentId);
  
  			StreamingRuntimeContext context = mock(StreamingRuntimeContext.class);
  			when(context.getTaskName()).thenReturn(thisComponentId);
  			when(context.getNumberOfParallelSubtasks()).thenReturn(dops.get(thisComponentId));
  			when(context.getIndexOfThisSubtask()).thenReturn(index);
  			taskCounter.put(thisComponentId, ++index);
  
  			Config stormConfig = new Config();
  			stormConfig.put(WrapperSetupHelper.TOPOLOGY_NAME, "test");
  
  			TopologyContext topologyContext = WrapperSetupHelper.createTopologyContext(context,
  					operators.get(thisComponentId), thisComponentId, stormTopology, stormConfig);
  
  			ComponentCommon expcetedCommon = expectedContext.getComponentCommon(thisComponentId);
  			ComponentCommon common = topologyContext.getComponentCommon(thisComponentId);
  
  			Assert.assertNull(topologyContext.getCodeDir());
  			Assert.assertNull(common.get_json_conf());
  			Assert.assertNull(topologyContext.getExecutorData(null));
  			Assert.assertNull(topologyContext.getPIDDir());
  			Assert.assertNull(topologyContext.getResource(null));
  			Assert.assertNull(topologyContext.getSharedExecutor());
  			Assert.assertNull(expectedContext.getTaskData(null));
  			Assert.assertNull(topologyContext.getThisWorkerPort());
  
  			Assert.assertTrue(expectedContext.getStormId().startsWith(topologyContext.getStormId()));
  
  			Assert.assertEquals(expcetedCommon.get_inputs(), common.get_inputs());
  			Assert.assertEquals(expcetedCommon.get_parallelism_hint(), common.get_parallelism_hint());
  			Assert.assertEquals(expcetedCommon.get_streams(), common.get_streams());
  			Assert.assertEquals(expectedContext.getComponentIds(), topologyContext.getComponentIds());
  			Assert.assertEquals(expectedContext.getComponentStreams(thisComponentId),
  					topologyContext.getComponentStreams(thisComponentId));
  			Assert.assertEquals(thisComponentId, topologyContext.getThisComponentId());
  			Assert.assertEquals(expectedContext.getThisSources(), topologyContext.getThisSources());
  			Assert.assertEquals(expectedContext.getThisStreams(), topologyContext.getThisStreams());
  			Assert.assertEquals(expectedContext.getThisTargets(), topologyContext.getThisTargets());
  			Assert.assertEquals(0, topologyContext.getThisWorkerTasks().size());
  
  			for (int taskId : topologyContext.getComponentTasks(thisComponentId)) {
  				Assert.assertEquals(thisComponentId, topologyContext.getComponentId(taskId));
  			}
  
  			for (String componentId : expectedContext.getComponentIds()) {
  				Assert.assertEquals(expectedContext.getSources(componentId),
  						topologyContext.getSources(componentId));
  				Assert.assertEquals(expectedContext.getTargets(componentId),
  						topologyContext.getTargets(componentId));
  
  				for (String streamId : expectedContext.getComponentStreams(componentId)) {
  					Assert.assertEquals(
  							expectedContext.getComponentOutputFields(componentId, streamId).toList(),
  							topologyContext.getComponentOutputFields(componentId, streamId).toList());
  				}
  			}
  
  			for (String streamId : expectedContext.getThisStreams()) {
  				Assert.assertEquals(expectedContext.getThisOutputFields(streamId).toList(),
  						topologyContext.getThisOutputFields(streamId).toList());
  			}
  
  			HashMap<Integer, String> taskToComponents = new HashMap<Integer, String>();
  			Set<Integer> allTaskIds = new HashSet<Integer>();
  			for (String componentId : expectedContext.getComponentIds()) {
  				List<Integer> possibleTasks = expectedContext.getComponentTasks(componentId);
  				List<Integer> tasks = topologyContext.getComponentTasks(componentId);
  
  				Iterator<Integer> p_it = possibleTasks.iterator();
  				Iterator<Integer> t_it = tasks.iterator();
  				while(p_it.hasNext()) {
  					Assert.assertTrue(t_it.hasNext());
  					Assert.assertNull(taskToComponents.put(p_it.next(), componentId));
  					Assert.assertTrue(allTaskIds.add(t_it.next()));
  				}
  				Assert.assertFalse(t_it.hasNext());
  			}
  
  			Assert.assertEquals(taskToComponents, expectedContext.getTaskToComponent());
  			Assert.assertTrue(taskIds.add(topologyContext.getThisTaskId()));
  
  			try {
  				topologyContext.getHooks();
  				Assert.fail();
  			} catch (UnsupportedOperationException e) { /* expected */ }
  
  			try {
  				topologyContext.getRegisteredMetricByName(null);
  				Assert.fail();
  			} catch (UnsupportedOperationException e) { /* expected */ }
  		}
  	}
test_case.class_method_signature: WrapperSetupHelperTest.testCreateTopologyContext()
test_case.full_signature: @Test public void testCreateTopologyContext()
test_case.identifier: testCreateTopologyContext
test_case.invocations: put put put put put put put put put put put put put put put setSpout get get setSpout get get shuffleGrouping setBolt get get allGrouping setBolt get get shuffleGrouping shuffleGrouping shuffleGrouping shuffleGrouping setBolt get get setNumAckers submitTopology createTopology size sleep shutdown createTopology getStormTopology getThisComponentId get mock thenReturn when getTaskName thenReturn when getNumberOfParallelSubtasks get thenReturn when getIndexOfThisSubtask put put createTopologyContext get getComponentCommon getComponentCommon assertNull getCodeDir assertNull get_json_conf assertNull getExecutorData assertNull getPIDDir assertNull getResource assertNull getSharedExecutor assertNull getTaskData assertNull getThisWorkerPort assertTrue startsWith getStormId getStormId assertEquals get_inputs get_inputs assertEquals get_parallelism_hint get_parallelism_hint assertEquals get_streams get_streams assertEquals getComponentIds getComponentIds assertEquals getComponentStreams getComponentStreams assertEquals getThisComponentId assertEquals getThisSources getThisSources assertEquals getThisStreams getThisStreams assertEquals getThisTargets getThisTargets assertEquals size getThisWorkerTasks getComponentTasks assertEquals getComponentId getComponentIds assertEquals getSources getSources assertEquals getTargets getTargets getComponentStreams assertEquals toList getComponentOutputFields toList getComponentOutputFields getThisStreams assertEquals toList getThisOutputFields toList getThisOutputFields getComponentIds getComponentTasks getComponentTasks iterator iterator hasNext assertTrue hasNext assertNull put next assertTrue add next assertFalse hasNext assertEquals getTaskToComponent assertTrue add getThisTaskId getHooks fail getRegisteredMetricByName fail
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testCreateTopologyContext()
focal_class.declarator: TOPOLOGY_NAME = "storm.topology.name"
focal_class.modifier: final static
focal_class.original_string: final static String TOPOLOGY_NAME = "storm.topology.name";
focal_class.type: String
focal_class.var_name: TOPOLOGY_NAME
focal_class.declarator: tid
focal_class.modifier: private static
focal_class.original_string: private static int tid;
focal_class.type: int
focal_class.var_name: tid
focal_class.file: flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/WrapperSetupHelper.java
focal_class.identifier: WrapperSetupHelper
focal_class.class_method_signature: WrapperSetupHelper.getNumberOfAttributes(final IComponent spoutOrBolt,
  			final Collection<String> rawOutputs)
focal_class.full_signature: static HashMap<String, Integer> getNumberOfAttributes(final IComponent spoutOrBolt,
  			final Collection<String> rawOutputs)
focal_class.identifier: getNumberOfAttributes
focal_class.modifiers: static
focal_class.parameters: (final IComponent spoutOrBolt,
  			final Collection<String> rawOutputs)
focal_class.return: HashMap<String, Integer>
focal_class.signature: HashMap<String, Integer> getNumberOfAttributes(final IComponent spoutOrBolt,
  			final Collection<String> rawOutputs)
focal_class.class_method_signature: WrapperSetupHelper.createTopologyContext(
  			final StreamingRuntimeContext context, final IComponent spoutOrBolt,
  			final String operatorName, StormTopology stormTopology, final Map stormConfig)
focal_class.full_signature: @SuppressWarnings({ "rawtypes", "unchecked" }) static synchronized TopologyContext createTopologyContext(
  			final StreamingRuntimeContext context, final IComponent spoutOrBolt,
  			final String operatorName, StormTopology stormTopology, final Map stormConfig)
focal_class.identifier: createTopologyContext
focal_class.modifiers: @SuppressWarnings({ "rawtypes", "unchecked" }) static synchronized
focal_class.parameters: (
  			final StreamingRuntimeContext context, final IComponent spoutOrBolt,
  			final String operatorName, StormTopology stormTopology, final Map stormConfig)
focal_class.return: TopologyContext
focal_class.signature: TopologyContext createTopologyContext(
  			final StreamingRuntimeContext context, final IComponent spoutOrBolt,
  			final String operatorName, StormTopology stormTopology, final Map stormConfig)
focal_class.class_method_signature: WrapperSetupHelper.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)
focal_class.full_signature: 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)
focal_class.identifier: processSingleOperator
focal_class.modifiers: private static
focal_class.parameters: (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)
focal_class.return: Integer
focal_class.signature: 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)
focal_method.body: @SuppressWarnings({ "rawtypes", "unchecked" })
  	static synchronized TopologyContext createTopologyContext(
  			final StreamingRuntimeContext context, final IComponent spoutOrBolt,
  			final String operatorName, StormTopology stormTopology, final Map stormConfig) {
  
  		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 = -1;
  		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>());
  
  			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 (i.e. FlinkTopology 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 = 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);
  	}
focal_method.class_method_signature: WrapperSetupHelper.createTopologyContext(
  			final StreamingRuntimeContext context, final IComponent spoutOrBolt,
  			final String operatorName, StormTopology stormTopology, final Map stormConfig)
focal_method.full_signature: @SuppressWarnings({ "rawtypes", "unchecked" }) static synchronized TopologyContext createTopologyContext(
  			final StreamingRuntimeContext context, final IComponent spoutOrBolt,
  			final String operatorName, StormTopology stormTopology, final Map stormConfig)
focal_method.identifier: createTopologyContext
focal_method.invocations: getNumberOfParallelSubtasks get set_parallelism_hint put put put add put declareOutputFields put get_spouts get_bolts get_state_spouts entrySet processSingleOperator getKey get_common getValue getIndexOfThisSubtask entrySet processSingleOperator getKey get_common getValue getIndexOfThisSubtask entrySet processSingleOperator getKey get_common getValue getIndexOfThisSubtask containsKey put
focal_method.modifiers: @SuppressWarnings({ "rawtypes", "unchecked" }) static synchronized
focal_method.parameters: (
  			final StreamingRuntimeContext context, final IComponent spoutOrBolt,
  			final String operatorName, StormTopology stormTopology, final Map stormConfig)
focal_method.return: TopologyContext
focal_method.signature: TopologyContext createTopologyContext(
  			final StreamingRuntimeContext context, final IComponent spoutOrBolt,
  			final String operatorName, StormTopology stormTopology, final Map stormConfig)