KafkaTopicEmitter implements Emitter { public void initialize(Properties properties) throws RequiredInputMissingException, ComponentInitializationFailedException { if(properties == null) throw new RequiredInputMissingException("Missing required properties"); clientId = StringUtils.lowerCase(StringUtils.trim(properties.getProperty(CFG_OPT_KAFKA_CLIENT_ID))); if(StringUtils.isBlank(clientId)) throw new RequiredInputMissingException("Missing required kafka client id"); zookeeperConnect = StringUtils.lowerCase(StringUtils.trim(properties.getProperty(CFG_OPT_ZOOKEEPER_CONNECT))); if(StringUtils.isBlank(zookeeperConnect)) throw new RequiredInputMissingException("Missing required zookeeper connect"); topicId = StringUtils.lowerCase(StringUtils.trim(properties.getProperty(CFG_OPT_TOPIC_ID))); if(StringUtils.isBlank(topicId)) throw new RequiredInputMissingException("Missing required topic"); brokerList = StringUtils.lowerCase(StringUtils.trim(properties.getProperty(CFG_OPT_BROKER_LIST))); if(StringUtils.isBlank(brokerList)) throw new RequiredInputMissingException("Missing required broker list"); String charsetName = StringUtils.trim(properties.getProperty(CFG_OPT_CHARSET)); try { if(StringUtils.isNotBlank(charsetName)) charset = Charset.forName(charsetName); else charset = Charset.forName("UTF-8"); } catch(Exception e) { throw new RequiredInputMissingException("Invalid character set provided: " + charsetName + ". Error: " + e.getMessage()); } messageAcking = StringUtils.equalsIgnoreCase(StringUtils.trim(properties.getProperty(CFG_OPT_MESSAGE_ACKING)), "true"); if(logger.isDebugEnabled()) logger.debug("kafka emitter[id="+this.id+", client="+clientId+", topic="+topicId+", broker="+brokerList+", zookeeper="+zookeeperConnect+", charset="+charset.name()+", messageAck="+messageAcking+"]"); if(kafkaProducer == null) { Properties props = new Properties(); props.put(CFG_ZK_CONNECT, zookeeperConnect); props.put(CFG_BROKER_LIST, brokerList); props.put(CFG_REQUEST_REQUIRED_ACKS, (messageAcking ? "1" : "0")); props.put(CFG_CLIENT_ID, clientId); this.kafkaProducer = new Producer<>(new ProducerConfig(props)); } }  void initialize(Properties properties); boolean shutdown(); boolean onMessage(StreamingDataMessage message); long getTotalNumOfMessages(); MicroPipelineComponentType getType(); void setId(String id); String getId(); static final String CFG_OPT_KAFKA_CLIENT_ID; static final String CFG_OPT_ZOOKEEPER_CONNECT; static final String CFG_OPT_TOPIC_ID; static final String CFG_OPT_MESSAGE_ACKING; static final String CFG_OPT_BROKER_LIST; static final String CFG_OPT_CHARSET; }
@Test public void testInitialize_withNullProperties() throws Exception { try { new KafkaTopicEmitter().initialize(null); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } }
@Test public void testInitialize_withEmptyClientId() throws Exception { this.defaultProperties.put(KafkaTopicEmitter.CFG_OPT_KAFKA_CLIENT_ID, ""); try { new KafkaTopicEmitter().initialize(defaultProperties); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } }
@Test public void testInitialize_withEmptyZookeeperConnect() throws Exception { this.defaultProperties.put(KafkaTopicEmitter.CFG_OPT_ZOOKEEPER_CONNECT, ""); try { new KafkaTopicEmitter().initialize(defaultProperties); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } }
@Test public void testInitialize_withEmptyTopicId() throws Exception { this.defaultProperties.put(KafkaTopicEmitter.CFG_OPT_TOPIC_ID, ""); try { new KafkaTopicEmitter().initialize(defaultProperties); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } }
@Test public void testInitialize_withEmptyBrokerList() throws Exception { this.defaultProperties.put(KafkaTopicEmitter.CFG_OPT_BROKER_LIST, ""); try { new KafkaTopicEmitter().initialize(defaultProperties); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } }
@Test public void testInitialize_withInvalidCharset() throws Exception { this.defaultProperties.put(KafkaTopicEmitter.CFG_OPT_CHARSET, "unknown-charset"); try { new KafkaTopicEmitter().initialize(defaultProperties); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } }