Hi Bharath, The DMaaP topic is unauthenticated.VES_MEASUREMENT_OUTPUT and the PNDA topic is ves.avro. The logstash configuration would look like this:
input { kafka { topics => ["unauthenticated.VES_MEASUREMENT_OUTPUT"] bootstrap_servers => "dmaap-host:9092" } } filter { mutate { add_field => { "src" => "ves" "host_ip" => "onap.demon" } rename => { "message" => "rawdata"} } ruby { # Convert the Logstash timestamp to a milliseconds timestamp integer # You can use whatever makes sense to you as long as the timestamp is a valid timestamp integer in milliseconds code => "event.set('timestamp', (event.get('@timestamp').to_f * 1000).to_i)" } } output { kafka { topic_id => "ves.avro" bootstrap_servers => "pnda-kafka-host:9094" value_serializer => "org.apache.kafka.common.serialization.ByteArraySerializer" codec => pnda-avro{ schema_uri => "dataplatform-raw.avsc" } } stdout { codec => "json" } } Note that this will also display the messages to stdout, which is helpful for debugging. You can remove the stdout section if you don't want that. Cheers, Donald. -=-=-=-=-=-=-=-=-=-=-=- Links: You receive all messages sent to this group. View/Reply Online (#14050): https://lists.onap.org/g/onap-discuss/message/14050 Mute This Topic: https://lists.onap.org/mt/28292264/21656 Group Owner: onap-discuss+ow...@lists.onap.org Unsubscribe: https://lists.onap.org/g/onap-discuss/unsub [arch...@mail-archive.com] -=-=-=-=-=-=-=-=-=-=-=-