ProfileSplitterBolt extends ConfiguredProfilerBolt { @Override public void execute(Tuple input) { try { LOG.debug("Received message; topic={}, partition={}, offset={}, kafkaTimestamp={}", input.contains(TOPIC.getFieldName()) ? input.getStringByField(TOPIC.getFieldName()): "unknown", input.contains(PARTITION.getFieldName()) ? input.getIntegerByField(PARTITION.getFieldName()): "unknown", input.contains(OFFSET.getFieldName()) ? input.getLongByField(OFFSET.getFieldName()): "unknown", input.contains(TIMESTAMP.getFieldName()) ? input.getLongByField(TIMESTAMP.getFieldName()): "unknown"); doExecute(input); } catch (Throwable t) { LOG.error("Unexpected error", t); collector.reportError(t); } finally { collector.ack(input); } } ProfileSplitterBolt(String zookeeperUrl); @Override void prepare(Map stormConf, TopologyContext context, OutputCollector collector); Context getStellarContext(); @Override void execute(Tuple input); @Override void declareOutputFields(OutputFieldsDeclarer declarer); void setRouter(MessageRouter router);  }
@Test public void testEmitTupleWithOneProfile() throws Exception { ProfilerConfig config = toProfilerConfig(profileWithOnlyIfTrue); ProfileSplitterBolt bolt = createBolt(config); bolt.execute(tuple); String expectedEntity = "10.0.0.1"; ProfileConfig expectedConfig = config.getProfiles().get(0); Values expected = new Values(message, timestamp, expectedEntity, expectedConfig); verify(outputCollector, times(1)) .emit(eq(tuple), eq(expected)); verify(outputCollector, times(1)) .ack(eq(tuple)); }
@Test public void testEmitTupleWithTwoProfiles() throws Exception { ProfilerConfig config = toProfilerConfig(twoProfilesDefined); ProfileSplitterBolt bolt = createBolt(config); bolt.execute(tuple); final String expectedEntity = "global"; { ProfileConfig profile1 = config.getProfiles().get(0); Values expected = new Values(message, timestamp, expectedEntity, profile1); verify(outputCollector, times(1)) .emit(eq(tuple), eq(expected)); } { ProfileConfig profile2 = config.getProfiles().get(1); Values expected = new Values(message, timestamp, expectedEntity, profile2); verify(outputCollector, times(1)) .emit(eq(tuple), eq(expected)); } verify(outputCollector, times(1)) .ack(eq(tuple)); }
@Test public void testNoProfilesDefined() throws Exception { ProfilerConfig config = toProfilerConfig(noProfilesDefined); ProfileSplitterBolt bolt = createBolt(config); bolt.execute(tuple); verify(outputCollector, times(0)) .emit(any(Tuple.class), any()); verify(outputCollector, times(1)) .ack(eq(tuple)); }
@Test public void testOnlyIfTrue() throws Exception { ProfilerConfig config = toProfilerConfig(profileWithOnlyIfTrue); ProfileSplitterBolt bolt = createBolt(config); bolt.execute(tuple); verify(outputCollector, times(1)) .emit(eq(tuple), any(Values.class)); verify(outputCollector, times(1)) .ack(eq(tuple)); }
@Test public void testOnlyIfMissing() throws Exception { ProfilerConfig config = toProfilerConfig(profileWithOnlyIfMissing); ProfileSplitterBolt bolt = createBolt(config); bolt.execute(tuple); verify(outputCollector, times(1)) .emit(eq(tuple), any(Values.class)); verify(outputCollector, times(1)) .ack(eq(tuple)); }
@Test public void testOnlyIfFalse() throws Exception { ProfilerConfig config = toProfilerConfig(profileWithOnlyIfFalse); ProfileSplitterBolt bolt = createBolt(config); bolt.execute(tuple); verify(outputCollector, times(0)) .emit(any()); verify(outputCollector, times(1)) .ack(eq(tuple)); }
@Test public void testResolveEntityName() throws Exception { ProfilerConfig config = toProfilerConfig(profileWithOnlyIfTrue); ProfileSplitterBolt bolt = createBolt(config); bolt.execute(tuple); String expectedEntity = "10.0.0.1"; ProfileConfig expectedConfig = config.getProfiles().get(0); Values expected = new Values(message, timestamp, expectedEntity, expectedConfig); verify(outputCollector, times(1)) .emit(eq(tuple), eq(expected)); verify(outputCollector, times(1)) .ack(eq(tuple)); }
@Test public void testOnlyIfInvalid() throws Exception { ProfilerConfig config = toProfilerConfig(profileWithOnlyIfInvalid); ProfileSplitterBolt bolt = createBolt(config); bolt.execute(tuple); verify(outputCollector, times(0)) .emit(any(Values.class)); }
@Test public void testWithNullMessage() throws Exception { when(tuple.getBinary(0)).thenReturn(null); ProfilerConfig config = toProfilerConfig(profileWithOnlyIfInvalid); ProfileSplitterBolt bolt = createBolt(config); bolt.execute(tuple); verify(outputCollector, times(0)) .emit(any(Values.class)); }