BoltWrapper extends AbstractStreamOperator<OUT> implements OneInputStreamOperator<IN, OUT> { @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); } BoltWrapper(final IRichBolt bolt);  BoltWrapper(final IRichBolt bolt, final Fields inputSchema);  BoltWrapper(final IRichBolt bolt, final String[] rawOutputs);  BoltWrapper(final IRichBolt bolt, final Collection<String> rawOutputs);  BoltWrapper(
			final IRichBolt bolt,
			final Fields inputSchema,
			final String[] rawOutputs);  BoltWrapper(final IRichBolt bolt, final Fields inputSchema,
			final Collection<String> rawOutputs);  BoltWrapper(final IRichBolt bolt, final String name, final String inputStreamId,
			final String inputComponentId, final Fields inputSchema,
			final Collection<String> rawOutputs); void setStormTopology(StormTopology stormTopology); @Override void open(); @Override void dispose(); @Override void processElement(final StreamRecord<IN> element); static final String DEFAULT_ID; static final String DEFUALT_BOLT_NAME; }
@SuppressWarnings("unchecked") @Test public void testOpen() throws Exception { 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"); when(taskContext.getMetricGroup()).thenReturn(new UnregisteredMetricsGroup()); final SetupOutputFieldsDeclarer declarer = new SetupOutputFieldsDeclarer(); declarer.declare(new Fields("dummy")); PowerMockito.whenNew(SetupOutputFieldsDeclarer.class).withNoArguments().thenReturn(declarer); { 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)); } { 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)); } { 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())); } } }
@SuppressWarnings("unchecked") @Test public void testOpenSink() throws Exception { 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"); when(taskContext.getMetricGroup()).thenReturn(new UnregisteredMetricsGroup()); final IRichBolt bolt = mock(IRichBolt.class); BoltWrapper<Object, Object> wrapper = new BoltWrapper<Object, Object>(bolt); wrapper.setup(createMockStreamTask(), new StreamConfig(new Configuration()), mock(Output.class)); wrapper.open(); verify(bolt).prepare(any(Map.class), any(TopologyContext.class), isNotNull(OutputCollector.class)); }