SFM Publication Interface

Analysis

EPS specifies following interfaces for event publication.

1. PublicationIFace
2. TopicIFace

We have to implement publishEvent method in TopicIFace interface to publish messages using Apache Kafka

    public String publishEvent(Message event);

Need to provide implementation of above method which will act as an interface between eps and kafka. So we have to write a code which connects to kafka server and publishes the messages into particular topic.

Following is the code snippet which publishes message to kafka topic

1) Start Zookeeper

bin/zookeeper-server-start.sh config/zookeeper.properties

2) Start Broker/Kafka server

bin/kafka-server-start.sh config/server.properties

3) Create Topic or Kafka supports dynamic creation of topic as well

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

4) Start Producer

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

5) Start Consumer

bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning

        Properties properties = new Properties();

        properties.put("metadata.broker.list","localhost:9092");

        properties.put("serializer.class","kafka.serializer.StringEncoder");

        ProducerConfig producerConfig = new ProducerConfig(properties);

        kafka.javaapi.producer.Producer<String,String> producer = new kafka.javaapi.producer.Producer<String, String>(producerConfig);

        KeyedMessage<String, String> message =new KeyedMessage<String, String>("testtopic","Test message from java program client");

        producer.send(message);

        producer.close();