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

   1 //
   2 // ============LICENSE_START=======================================================
   3 //  Copyright (C) 2016-2018 Ericsson. All rights reserved.
   4 //  Modifications Copyright (C) 2019 Nordix Foundation.
   5 // ================================================================================
   6 // This file is licensed under the CREATIVE COMMONS ATTRIBUTION 4.0 INTERNATIONAL LICENSE
   7 // Full license text at https://creativecommons.org/licenses/by/4.0/legalcode
   8 //
   9 // SPDX-License-Identifier: CC-BY-4.0
  10 // ============LICENSE_END=========================================================
  11 //
  12 // @author Sven van der Meer (sven.van.der.meer@ericsson.com)
  13 //
  14
  15 == Kafka IO
  16
  17 Kafka IO is supported by the APEX Kafka plugin.
  18 The configurations below are examples.
  19 APEX will take any configuration inside the parameter object and forward it to Kafka.
  20 More information on Kafka specific configuration parameters can be found in the Kafka documentation:
  21
  22 * link:https://kafka.apache.org/090/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html[Kafka Consumer Class]
  23 * link:https://kafka.apache.org/090/javadoc/org/apache/kafka/clients/producer/KafkaProducer.html[Kafka Producer Class]
  24
  25
  26
  27 === Kafka Input
  28
  29 APEX will receive events from the Apache Kafka messaging system.
  30 The input is uni-directional, an engine will only receive events from the input but not send any event to the input.
  31
  32 [source%nowrap,json]
  33 ----
  34 "carrierTechnologyParameters" : {
  35   "carrierTechnology" : "KAFKA", <1>
  36   "parameterClassName" :
  37     "org.onap.policy.apex.plugins.event.carrier.kafka.KAFKACarrierTechnologyParameters",
  38   "parameters" : {
  39     "bootstrapServers"  : "localhost:49092", <2>
  40     "groupId"           : "apex-group-id", <3>
  41     "enableAutoCommit"  : true, <4>
  42     "autoCommitTime"    : 1000, <5>
  43     "sessionTimeout"    : 30000, <6>
  44     "consumerPollTime"  : 100, <7>
  45     "consumerTopicList" : ["apex-in-0", "apex-in-1"], <8>
  46     "keyDeserializer"   :
  47         "org.apache.kafka.common.serialization.StringDeserializer", <9>
  48     "valueDeserializer" :
  49         "org.apache.kafka.common.serialization.StringDeserializer", <10>
  50     "kafkaProperties"   : [] <11>
  51   }
  52 }
  53 ----
  54
  55 <1> set Kafka as carrier technology
  56 <2> bootstrap server and port
  57 <3> a group identifier
  58 <4> flag for auto-commit
  59 <5> auto-commit timeout in milliseconds
  60 <6> session timeout in milliseconds
  61 <7> consumer poll time in milliseconds
  62 <8> consumer topic list
  63 <9> key for the Kafka de-serializer
  64 <10> value for the Kafka de-serializer
  65 <11> an optional list of name value pairs of properties to be passed transparently to Kafka.
  66 This field need not be specified, can be set to null, or to an empty list as here.
  67
  68
  69 === Kafka Output
  70
  71 APEX will send events to the Apache Kafka messaging system.
  72 The output is uni-directional, an engine will send events to the output but not receive any event from the output.
  73
  74
  75 [source%nowrap,json]
  76 ----
  77 "carrierTechnologyParameters" : {
  78   "carrierTechnology" : "KAFKA", <1>
  79   "parameterClassName" :
  80     "org.onap.policy.apex.plugins.event.carrier.kafka.KAFKACarrierTechnologyParameters",
  81   "parameters" : {
  82     "bootstrapServers"  : "localhost:49092", <2>
  83     "acks"              : "all", <3>
  84     "retries"           : 0, <4>
  85     "batchSize"         : 16384, <5>
  86     "lingerTime"        : 1, <6>
  87     "bufferMemory"      : 33554432, <7>
  88     "producerTopic"     : "apex-out", <8>
  89     "keySerializer"     :
  90         "org.apache.kafka.common.serialization.StringSerializer", <9>
  91     "valueSerializer"   :
  92         "org.apache.kafka.common.serialization.StringSerializer", <10>
  93     "kafkaProperties": [ <11>
  94         ["message.max.bytes", 1000000],
  95         ["compression.codec", "none"]
  96     ]
  97   }
  98 }
  99 ----
 100
 101 <1> set Kafka as carrier technology
 102 <2> bootstrap server and port
 103 <3> acknowledgement strategy
 104 <4> number of retries
 105 <5> batch size
 106 <6> time to linger in milliseconds
 107 <7> buffer memory in byte
 108 <8> producer topic
 109 <9> key for the Kafka serializer
 110 <10> value for the Kafka serializer
 111 <11> an optional list of name value pairs of properties to be passed transparently to Kafka. If a property appears in
 112 the _kafkaProperties_ field and is also explicitly specified to a non-default value (such as _lingerTime_
 113 and _linger.ms_) the explictly specified value of the property is used rather than the value specified in the
 114 _kafkaProperties_ list.