ConfigurableIngestTopology { void configureStreamFieldsGrouping(String prevComponent, String streamId, Configuration boltConf, BoltDeclarer declarer) throws ConfigurationException { String streamTypeArgs = boltConf.getString(STREAM_GROUPING_CONF_ARGS); if (StringUtils.isBlank(streamTypeArgs)) { logger.error("{} stream grouping requires {} configuration", STREAM_GROUPING_FIELDS, STREAM_GROUPING_CONF_ARGS); throw new ConfigurationException("Missing stream grouping configuration"); } else { String[] fields = streamTypeArgs.split(","); declarer.fieldsGrouping(prevComponent, streamId, new Fields(fields)); } }  static void main(String[] args);  }
@Test public void testConfigureStreamFieldsGrouping( @Injectable String prevComponent, @Injectable String streamId, @Injectable Configuration boltConf, @Injectable BoltDeclarer declarer, @Injectable Fields fields) throws Exception { new Expectations(Fields.class) {{ boltConf.getString(STREAM_GROUPING_CONF_ARGS); result = "a,b"; new Fields(new String[]{"a", "b"}); result = fields; declarer.fieldsGrouping(prevComponent, streamId, fields); }}; topology.configureStreamFieldsGrouping(prevComponent, streamId, boltConf, declarer); }
@Test public void configureStreamFieldsGroupingFail( @Injectable String prevComponent, @Injectable String streamId, @Injectable Configuration boltConf, @Injectable BoltDeclarer declarer) throws Exception { new Expectations() {{ boltConf.getString(STREAM_GROUPING_CONF_ARGS); result = ""; }}; thrown.expect(ConfigurationException.class); topology.configureStreamFieldsGrouping(prevComponent, streamId, boltConf, declarer); }