gerrit.onap Code Review - policy/apex-pdp.git/blob

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  *  Copyright (C) 2016-2018 Ericsson. All rights reserved.
   4  * ================================================================================
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *      http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  *
  17  * SPDX-License-Identifier: Apache-2.0
  18  * ============LICENSE_END=========================================================
  19  */
  20
  21 package org.onap.policy.apex.apps.uservice.test.adapt.kafka;
  22
  23 import java.util.Arrays;
  24 import java.util.Properties;
  25
  26 import org.apache.kafka.clients.consumer.ConsumerRecord;
  27 import org.apache.kafka.clients.consumer.ConsumerRecords;
  28 import org.apache.kafka.clients.consumer.KafkaConsumer;
  29 import org.onap.policy.apex.core.infrastructure.messaging.MessagingException;
  30 import org.onap.policy.apex.core.infrastructure.threading.ThreadUtilities;
  31
  32 /**
  33  * @author Liam Fallon (liam.fallon@ericsson.com)
  34  */
  35 public class KafkaEventSubscriber implements Runnable {
  36     private final String topic;
  37     private final String kafkaServerAddress;
  38     private long eventsReceivedCount = 0;
  39
  40     KafkaConsumer<String, String> consumer;
  41
  42     Thread subscriberThread;
  43
  44     public KafkaEventSubscriber(final String topic, final String kafkaServerAddress) throws MessagingException {
  45         this.topic = topic;
  46         this.kafkaServerAddress = kafkaServerAddress;
  47
  48         final Properties props = new Properties();
  49         props.put("bootstrap.servers", kafkaServerAddress);
  50         props.put("group.id", "test");
  51         props.put("enable.auto.commit", "true");
  52         props.put("auto.commit.interval.ms", "1000");
  53         props.put("session.timeout.ms", "30000");
  54         props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  55         props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  56
  57         consumer = new KafkaConsumer<String, String>(props);
  58         consumer.subscribe(Arrays.asList(topic));
  59
  60         subscriberThread = new Thread(this);
  61         subscriberThread.start();
  62     }
  63
  64     @Override
  65     public void run() {
  66         System.out.println(KafkaEventSubscriber.class.getCanonicalName() + ": receiving events from Kafka server at "
  67                 + kafkaServerAddress + " on topic " + topic);
  68
  69         while (subscriberThread.isAlive() && !subscriberThread.isInterrupted()) {
  70             try {
  71                 final ConsumerRecords<String, String> records = consumer.poll(100);
  72                 for (final ConsumerRecord<String, String> record : records) {
  73                     System.out.println("******");
  74                     System.out.println("offset=" + record.offset());
  75                     System.out.println("key=" + record.key());
  76                     System.out.println("name=" + record.value());
  77                     eventsReceivedCount++;
  78                 }
  79             } catch (final Exception e) {
  80                 // Thread interrupted
  81                 break;
  82             }
  83         }
  84
  85         System.out.println(KafkaEventSubscriber.class.getCanonicalName() + ": event reception completed");
  86     }
  87
  88     public long getEventsReceivedCount() {
  89         return eventsReceivedCount;
  90     }
  91
  92     public void shutdown() {
  93         subscriberThread.interrupt();
  94
  95         while (subscriberThread.isAlive()) {
  96             ThreadUtilities.sleep(10);
  97         }
  98
  99         consumer.close();
 100         System.out.println(KafkaEventSubscriber.class.getCanonicalName() + ": stopped");
 101     }
 102
 103
 104     public static void main(final String[] args) throws MessagingException {
 105         if (args.length != 2) {
 106             System.err.println("usage KafkaEventSubscriber topic kafkaServerAddress");
 107             return;
 108         }
 109         new KafkaEventSubscriber(args[0], args[1]);
 110     }
 111 }