- @Override
- public void init() {
- Properties dmaapConsumerProperties = applicationConfig.getDmaapConsumerConfig();
- Properties dmaapPublisherProperties = applicationConfig.getDmaapPublisherConfig();
- // No need to start if there is no configuration.
- if (dmaapConsumerProperties == null || dmaapPublisherProperties == null || dmaapConsumerProperties.size() == 0
- || dmaapPublisherProperties.size() == 0) {
- logger.error("DMaaP properties Failed to Load");
- return;
- }
- try {
- logger.debug("Creating DMAAP Client");
- logger.debug("dmaapConsumerProperties---> {}", dmaapConsumerProperties.getProperty("topic"));
- logger.debug("dmaapPublisherProperties---> {}", dmaapPublisherProperties.getProperty("topic"));
- consumer = MRClientFactory.createConsumer(dmaapConsumerProperties);
- this.alive = true;
- } catch (IOException e) {
- logger.error("Exception occurred while creating Dmaap Consumer", e);
- }
- }
-
- @Override
- public void processMsg(String msg) throws Exception {