test_class.file: flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/wrappers/BoltWrapperTest.java
test_class.identifier: BoltWrapperTest
test_class.superclass: extends AbstractTest
test_case.body: @SuppressWarnings("unchecked")
  	@Test
  	public void testOpen() throws Exception {
  
  		// utility mocks
  		final StormConfig stormConfig = new StormConfig();
  		final Configuration flinkConfig = new Configuration();
  
  		final ExecutionConfig taskConfig = mock(ExecutionConfig.class);
  		when(taskConfig.getGlobalJobParameters()).thenReturn(null).thenReturn(stormConfig)
  		.thenReturn(flinkConfig);
  
  		final StreamingRuntimeContext taskContext = mock(StreamingRuntimeContext.class);
  		when(taskContext.getExecutionConfig()).thenReturn(taskConfig);
  		when(taskContext.getTaskName()).thenReturn("name");
  
  		final SetupOutputFieldsDeclarer declarer = new SetupOutputFieldsDeclarer();
  		declarer.declare(new Fields("dummy"));
  		PowerMockito.whenNew(SetupOutputFieldsDeclarer.class).withNoArguments().thenReturn(declarer);
  
  		// (1) open with no configuration
  		{
  			ExecutionConfig execConfig = mock(ExecutionConfig.class);
  			when(execConfig.getGlobalJobParameters()).thenReturn(null);
  
  			final IRichBolt bolt = mock(IRichBolt.class);
  			BoltWrapper<Object, Object> wrapper = new BoltWrapper<Object, Object>(bolt);
  			wrapper.setup(createMockStreamTask(execConfig), new StreamConfig(new Configuration()), mock(Output.class));
  
  			wrapper.open();
  			verify(bolt).prepare(any(Map.class), any(TopologyContext.class), any(OutputCollector.class));
  		}
  
  		// (2) open with a storm specific configuration
  		{
  			ExecutionConfig execConfig = mock(ExecutionConfig.class);
  			when(execConfig.getGlobalJobParameters()).thenReturn(stormConfig);
  
  			final IRichBolt bolt = mock(IRichBolt.class);
  			BoltWrapper<Object, Object> wrapper = new BoltWrapper<Object, Object>(bolt);
  			wrapper.setup(createMockStreamTask(execConfig), new StreamConfig(new Configuration()), mock(Output.class));
  
  			wrapper.open();
  			verify(bolt).prepare(same(stormConfig), any(TopologyContext.class), any(OutputCollector.class));
  		}
  
  		// (3) open with a flink config
  		{
  			final Configuration cfg = new Configuration();
  			cfg.setString("foo", "bar");
  			cfg.setInteger("the end (the int)", Integer.MAX_VALUE);
  
  			ExecutionConfig execConfig = mock(ExecutionConfig.class);
  			when(execConfig.getGlobalJobParameters()).thenReturn(new UnmodifiableConfiguration(cfg));
  
  			TestDummyBolt testBolt = new TestDummyBolt();
  			BoltWrapper<Object, Object> wrapper = new BoltWrapper<Object, Object>(testBolt);
  			wrapper.setup(createMockStreamTask(execConfig), new StreamConfig(new Configuration()), mock(Output.class));
  
  			wrapper.open();
  			for (Entry<String, String> entry : cfg.toMap().entrySet()) {
  				Assert.assertEquals(entry.getValue(), testBolt.config.get(entry.getKey()));
  			}
  		}
  	}
test_case.class_method_signature: BoltWrapperTest.testOpen()
test_case.full_signature: @SuppressWarnings("unchecked") @Test public void testOpen()
test_case.identifier: testOpen
test_case.invocations: mock thenReturn thenReturn thenReturn when getGlobalJobParameters mock thenReturn when getExecutionConfig thenReturn when getTaskName declare thenReturn withNoArguments whenNew mock thenReturn when getGlobalJobParameters mock setup createMockStreamTask mock open prepare verify any any any mock thenReturn when getGlobalJobParameters mock setup createMockStreamTask mock open prepare verify same any any setString setInteger mock thenReturn when getGlobalJobParameters setup createMockStreamTask mock open entrySet toMap assertEquals getValue get getKey
test_case.modifiers: @SuppressWarnings("unchecked") @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testOpen()
focal_class.declarator: serialVersionUID = -4788589118464155835L
focal_class.modifier: private static final
focal_class.original_string: private static final long serialVersionUID = -4788589118464155835L;
focal_class.type: long
focal_class.var_name: serialVersionUID
focal_class.declarator: DEFAULT_ID = "default ID"
focal_class.modifier: public final static
focal_class.original_string: public final static String DEFAULT_ID = "default ID";
focal_class.type: String
focal_class.var_name: DEFAULT_ID
focal_class.declarator: DEFUALT_BOLT_NAME = "Unnamed Bolt"
focal_class.modifier: public final static
focal_class.original_string: public final static String DEFUALT_BOLT_NAME = "Unnamed Bolt";
focal_class.type: String
focal_class.var_name: DEFUALT_BOLT_NAME
focal_class.declarator: bolt
focal_class.modifier: protected final
focal_class.original_string: protected final IRichBolt bolt;
focal_class.type: IRichBolt
focal_class.var_name: bolt
focal_class.declarator: name
focal_class.modifier: private final
focal_class.original_string: private final String name;
focal_class.type: String
focal_class.var_name: name
focal_class.declarator: numberOfAttributes
focal_class.modifier: private final
focal_class.original_string: private final HashMap<String, Integer> numberOfAttributes;
focal_class.type: HashMap<String, Integer>
focal_class.var_name: numberOfAttributes
focal_class.declarator: stormTopology
focal_class.modifier: private
focal_class.original_string: private StormTopology stormTopology;
focal_class.type: StormTopology
focal_class.var_name: stormTopology
focal_class.declarator: topologyContext
focal_class.modifier: private transient
focal_class.original_string: private transient TopologyContext topologyContext;
focal_class.type: TopologyContext
focal_class.var_name: topologyContext
focal_class.declarator: inputStreamIds = new HashMap<Integer, String>()
focal_class.modifier: private final
focal_class.original_string: private final HashMap<Integer, String> inputStreamIds = new HashMap<Integer, String>();
focal_class.type: HashMap<Integer, String>
focal_class.var_name: inputStreamIds
focal_class.declarator: inputComponentIds = new HashMap<Integer, String>()
focal_class.modifier: private final
focal_class.original_string: private final HashMap<Integer, String> inputComponentIds = new HashMap<Integer, String>();
focal_class.type: HashMap<Integer, String>
focal_class.var_name: inputComponentIds
focal_class.declarator: inputSchemas = new HashMap<Integer, Fields>()
focal_class.modifier: private final
focal_class.original_string: private final HashMap<Integer, Fields> inputSchemas = new HashMap<Integer, Fields>();
focal_class.type: HashMap<Integer, Fields>
focal_class.var_name: inputSchemas
focal_class.declarator: flinkCollector
focal_class.modifier: protected transient
focal_class.original_string: protected transient TimestampedCollector<OUT> flinkCollector;
focal_class.type: TimestampedCollector<OUT>
focal_class.var_name: flinkCollector
focal_class.file: flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/BoltWrapper.java
focal_class.identifier: BoltWrapper
focal_class.interfaces: implements OneInputStreamOperator<IN, OUT>
focal_class.class_method_signature: BoltWrapper.BoltWrapper(final IRichBolt bolt)
focal_class.full_signature: public  BoltWrapper(final IRichBolt bolt)
focal_class.identifier: BoltWrapper
focal_class.modifiers: public
focal_class.parameters: (final IRichBolt bolt)
focal_class.signature:  BoltWrapper(final IRichBolt bolt)
focal_class.class_method_signature: BoltWrapper.BoltWrapper(final IRichBolt bolt, final Fields inputSchema)
focal_class.full_signature: public  BoltWrapper(final IRichBolt bolt, final Fields inputSchema)
focal_class.identifier: BoltWrapper
focal_class.modifiers: public
focal_class.parameters: (final IRichBolt bolt, final Fields inputSchema)
focal_class.signature:  BoltWrapper(final IRichBolt bolt, final Fields inputSchema)
focal_class.class_method_signature: BoltWrapper.BoltWrapper(final IRichBolt bolt, final String[] rawOutputs)
focal_class.full_signature: public  BoltWrapper(final IRichBolt bolt, final String[] rawOutputs)
focal_class.identifier: BoltWrapper
focal_class.modifiers: public
focal_class.parameters: (final IRichBolt bolt, final String[] rawOutputs)
focal_class.signature:  BoltWrapper(final IRichBolt bolt, final String[] rawOutputs)
focal_class.class_method_signature: BoltWrapper.BoltWrapper(final IRichBolt bolt, final Collection<String> rawOutputs)
focal_class.full_signature: public  BoltWrapper(final IRichBolt bolt, final Collection<String> rawOutputs)
focal_class.identifier: BoltWrapper
focal_class.modifiers: public
focal_class.parameters: (final IRichBolt bolt, final Collection<String> rawOutputs)
focal_class.signature:  BoltWrapper(final IRichBolt bolt, final Collection<String> rawOutputs)
focal_class.class_method_signature: BoltWrapper.BoltWrapper(final IRichBolt bolt, final Fields inputSchema, final String[] rawOutputs)
focal_class.full_signature: public  BoltWrapper(final IRichBolt bolt, final Fields inputSchema, final String[] rawOutputs)
focal_class.identifier: BoltWrapper
focal_class.modifiers: public
focal_class.parameters: (final IRichBolt bolt, final Fields inputSchema, final String[] rawOutputs)
focal_class.signature:  BoltWrapper(final IRichBolt bolt, final Fields inputSchema, final String[] rawOutputs)
focal_class.class_method_signature: BoltWrapper.BoltWrapper(final IRichBolt bolt, final Fields inputSchema,
  			final Collection<String> rawOutputs)
focal_class.full_signature: public  BoltWrapper(final IRichBolt bolt, final Fields inputSchema,
  			final Collection<String> rawOutputs)
focal_class.identifier: BoltWrapper
focal_class.modifiers: public
focal_class.parameters: (final IRichBolt bolt, final Fields inputSchema,
  			final Collection<String> rawOutputs)
focal_class.signature:  BoltWrapper(final IRichBolt bolt, final Fields inputSchema,
  			final Collection<String> rawOutputs)
focal_class.class_method_signature: BoltWrapper.BoltWrapper(final IRichBolt bolt, final String name, final String inputStreamId,
  			final String inputComponentId, final Fields inputSchema,
  			final Collection<String> rawOutputs)
focal_class.full_signature: public  BoltWrapper(final IRichBolt bolt, final String name, final String inputStreamId,
  			final String inputComponentId, final Fields inputSchema,
  			final Collection<String> rawOutputs)
focal_class.identifier: BoltWrapper
focal_class.modifiers: public
focal_class.parameters: (final IRichBolt bolt, final String name, final String inputStreamId,
  			final String inputComponentId, final Fields inputSchema,
  			final Collection<String> rawOutputs)
focal_class.signature:  BoltWrapper(final IRichBolt bolt, final String name, final String inputStreamId,
  			final String inputComponentId, final Fields inputSchema,
  			final Collection<String> rawOutputs)
focal_class.class_method_signature: BoltWrapper.setStormTopology(StormTopology stormTopology)
focal_class.full_signature: public void setStormTopology(StormTopology stormTopology)
focal_class.identifier: setStormTopology
focal_class.modifiers: public
focal_class.parameters: (StormTopology stormTopology)
focal_class.return: void
focal_class.signature: void setStormTopology(StormTopology stormTopology)
focal_class.class_method_signature: BoltWrapper.open()
focal_class.full_signature: @Override public void open()
focal_class.identifier: open
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void open()
focal_class.class_method_signature: BoltWrapper.dispose()
focal_class.full_signature: @Override public void dispose()
focal_class.identifier: dispose
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void dispose()
focal_class.class_method_signature: BoltWrapper.processElement(final StreamRecord<IN> element)
focal_class.full_signature: @Override public void processElement(final StreamRecord<IN> element)
focal_class.identifier: processElement
focal_class.modifiers: @Override public
focal_class.parameters: (final StreamRecord<IN> element)
focal_class.return: void
focal_class.signature: void processElement(final StreamRecord<IN> element)
focal_class.class_method_signature: BoltWrapper.processWatermark(Watermark mark)
focal_class.full_signature: @Override public void processWatermark(Watermark mark)
focal_class.identifier: processWatermark
focal_class.modifiers: @Override public
focal_class.parameters: (Watermark mark)
focal_class.return: void
focal_class.signature: void processWatermark(Watermark mark)
focal_class.superclass: extends AbstractStreamOperator<OUT>
focal_method.body: @Override
  	public void open() throws Exception {
  		super.open();
  
  		this.flinkCollector = new TimestampedCollector<>(this.output);
  
  		GlobalJobParameters config = getExecutionConfig().getGlobalJobParameters();
  		StormConfig stormConfig = new StormConfig();
  
  		if (config != null) {
  			if (config instanceof StormConfig) {
  				stormConfig = (StormConfig) config;
  			} else {
  				stormConfig.putAll(config.toMap());
  			}
  		}
  
  		this.topologyContext = WrapperSetupHelper.createTopologyContext(
  				getRuntimeContext(), this.bolt, this.name, this.stormTopology, stormConfig);
  
  		final OutputCollector stormCollector = new OutputCollector(new BoltCollector<OUT>(
  				this.numberOfAttributes, this.topologyContext.getThisTaskId(), this.flinkCollector));
  
  		if (this.stormTopology != null) {
  			Map<GlobalStreamId, Grouping> inputs = this.topologyContext.getThisSources();
  
  			for (GlobalStreamId inputStream : inputs.keySet()) {
  				for (Integer tid : this.topologyContext.getComponentTasks(inputStream
  						.get_componentId())) {
  					this.inputComponentIds.put(tid, inputStream.get_componentId());
  					this.inputStreamIds.put(tid, inputStream.get_streamId());
  					this.inputSchemas.put(tid,
  							this.topologyContext.getComponentOutputFields(inputStream));
  				}
  			}
  		}
  
  		this.bolt.prepare(stormConfig, this.topologyContext, stormCollector);
  	}
focal_method.class_method_signature: BoltWrapper.open()
focal_method.full_signature: @Override public void open()
focal_method.identifier: open
focal_method.invocations: open getGlobalJobParameters getExecutionConfig putAll toMap createTopologyContext getRuntimeContext getThisTaskId getThisSources keySet getComponentTasks get_componentId put get_componentId put get_streamId put getComponentOutputFields prepare
focal_method.modifiers: @Override public
focal_method.parameters: ()
focal_method.return: void
focal_method.signature: void open()