diff options
Diffstat (limited to 'plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-kafka/src')
4 files changed, 772 insertions, 0 deletions
diff --git a/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-kafka/src/main/java/org/onap/policy/apex/plugins/event/carrier/kafka/ApexKafkaConsumer.java b/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-kafka/src/main/java/org/onap/policy/apex/plugins/event/carrier/kafka/ApexKafkaConsumer.java new file mode 100644 index 000000000..3351a58e9 --- /dev/null +++ b/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-kafka/src/main/java/org/onap/policy/apex/plugins/event/carrier/kafka/ApexKafkaConsumer.java @@ -0,0 +1,198 @@ +/*- + * ============LICENSE_START======================================================= + * Copyright (C) 2016-2018 Ericsson. All rights reserved. + * ================================================================================ + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + * SPDX-License-Identifier: Apache-2.0 + * ============LICENSE_END========================================================= + */ + +package org.onap.policy.apex.plugins.event.carrier.kafka; + +import java.util.EnumMap; +import java.util.Map; + +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.onap.policy.apex.core.infrastructure.threading.ApplicationThreadFactory; +import org.onap.policy.apex.core.infrastructure.threading.ThreadUtilities; +import org.onap.policy.apex.service.engine.event.ApexEventConsumer; +import org.onap.policy.apex.service.engine.event.ApexEventException; +import org.onap.policy.apex.service.engine.event.ApexEventReceiver; +import org.onap.policy.apex.service.engine.event.PeeredReference; +import org.onap.policy.apex.service.parameters.eventhandler.EventHandlerParameters; +import org.onap.policy.apex.service.parameters.eventhandler.EventHandlerPeeredMode; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * This class implements an Apex event consumer that receives events using Kafka. + * + * @author Liam Fallon (liam.fallon@ericsson.com) + */ +public class ApexKafkaConsumer implements ApexEventConsumer, Runnable { + // Get a reference to the logger + private static final Logger LOGGER = LoggerFactory.getLogger(ApexKafkaConsumer.class); + + // The Kafka parameters read from the parameter service + private KAFKACarrierTechnologyParameters kafkaConsumerProperties; + + // The event receiver that will receive events from this consumer + private ApexEventReceiver eventReceiver; + + // The Kafka consumer used to receive events using Kafka + private KafkaConsumer<String, String> kafkaConsumer; + + // The name for this consumer + private String name = null; + + // The peer references for this event handler + private Map<EventHandlerPeeredMode, PeeredReference> peerReferenceMap = new EnumMap<>(EventHandlerPeeredMode.class); + + // The consumer thread and stopping flag + private Thread consumerThread; + private boolean stopOrderedFlag = false; + + /* + * (non-Javadoc) + * + * @see org.onap.policy.apex.service.engine.event.ApexEventConsumer#init(java.lang.String, + * org.onap.policy.apex.service.parameters.eventhandler.EventHandlerParameters, + * org.onap.policy.apex.service.engine.event.ApexEventReceiver) + */ + @Override + public void init(final String consumerName, final EventHandlerParameters consumerParameters, + final ApexEventReceiver incomingEventReceiver) throws ApexEventException { + this.eventReceiver = incomingEventReceiver; + this.name = consumerName; + + // Check and get the Kafka Properties + if (!(consumerParameters.getCarrierTechnologyParameters() instanceof KAFKACarrierTechnologyParameters)) { + LOGGER.warn("specified consumer properties of type \"" + + consumerParameters.getCarrierTechnologyParameters().getClass().getCanonicalName() + + "\" are not applicable to a Kafka consumer"); + throw new ApexEventException("specified consumer properties of type \"" + + consumerParameters.getCarrierTechnologyParameters().getClass().getCanonicalName() + + "\" are not applicable to a Kafka consumer"); + } + kafkaConsumerProperties = + (KAFKACarrierTechnologyParameters) consumerParameters.getCarrierTechnologyParameters(); + + // Kick off the Kafka consumer + kafkaConsumer = new KafkaConsumer<String, String>(kafkaConsumerProperties.getKafkaConsumerProperties()); + kafkaConsumer.subscribe(kafkaConsumerProperties.getConsumerTopicList()); + if (LOGGER.isDebugEnabled()) { + LOGGER.debug("event receiver for " + this.getClass().getName() + ":" + this.name + " subscribed to topics: " + + kafkaConsumerProperties.getConsumerTopicList()); + } + } + + /* + * (non-Javadoc) + * + * @see org.onap.policy.apex.service.engine.event.ApexEventConsumer#start() + */ + @Override + public void start() { + // Configure and start the event reception thread + final String threadName = this.getClass().getName() + ":" + this.name; + consumerThread = new ApplicationThreadFactory(threadName).newThread(this); + consumerThread.setDaemon(true); + consumerThread.start(); + } + + /* + * (non-Javadoc) + * + * @see org.onap.policy.apex.service.engine.event.ApexEventConsumer#getName() + */ + @Override + public String getName() { + return name; + } + + /* + * (non-Javadoc) + * + * @see org.onap.policy.apex.service.engine.event.ApexEventConsumer#getPeeredReference(org.onap.policy.apex.service. + * parameters.eventhandler.EventHandlerPeeredMode) + */ + @Override + public PeeredReference getPeeredReference(final EventHandlerPeeredMode peeredMode) { + return peerReferenceMap.get(peeredMode); + } + + /* + * (non-Javadoc) + * + * @see org.onap.policy.apex.service.engine.event.ApexEventConsumer#setPeeredReference(org.onap.policy.apex.service. + * parameters.eventhandler.EventHandlerPeeredMode, org.onap.policy.apex.service.engine.event.PeeredReference) + */ + @Override + public void setPeeredReference(final EventHandlerPeeredMode peeredMode, final PeeredReference peeredReference) { + peerReferenceMap.put(peeredMode, peeredReference); + } + + /* + * (non-Javadoc) + * + * @see java.lang.Runnable#run() + */ + @Override + public void run() { + // Kick off the Kafka consumer + kafkaConsumer = new KafkaConsumer<String, String>(kafkaConsumerProperties.getKafkaConsumerProperties()); + kafkaConsumer.subscribe(kafkaConsumerProperties.getConsumerTopicList()); + if (LOGGER.isDebugEnabled()) { + LOGGER.debug("event receiver for " + this.getClass().getName() + ":" + this.name + " subscribed to topics: " + + kafkaConsumerProperties.getConsumerTopicList()); + } + + // The endless loop that receives events over Kafka + while (consumerThread.isAlive() && !stopOrderedFlag) { + try { + final ConsumerRecords<String, String> records = + kafkaConsumer.poll(kafkaConsumerProperties.getConsumerPollTime()); + for (final ConsumerRecord<String, String> record : records) { + if (LOGGER.isTraceEnabled()) { + LOGGER.trace("event received for {} for forwarding to Apex engine : {} {}", + this.getClass().getName() + ":" + this.name, record.key(), record.value()); + } + eventReceiver.receiveEvent(record.value()); + } + } catch (final Exception e) { + LOGGER.warn("error receiving events on thread {}", consumerThread.getName(), e); + } + } + + if (!consumerThread.isInterrupted()) { + kafkaConsumer.close(); + } + } + + /* + * (non-Javadoc) + * + * @see org.onap.policy.apex.service.engine.event.ApexEventConsumer#stop() + */ + @Override + public void stop() { + stopOrderedFlag = true; + + while (consumerThread.isAlive()) { + ThreadUtilities.sleep(kafkaConsumerProperties.getConsumerPollTime()); + } + } +} diff --git a/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-kafka/src/main/java/org/onap/policy/apex/plugins/event/carrier/kafka/ApexKafkaProducer.java b/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-kafka/src/main/java/org/onap/policy/apex/plugins/event/carrier/kafka/ApexKafkaProducer.java new file mode 100644 index 000000000..fb851bc70 --- /dev/null +++ b/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-kafka/src/main/java/org/onap/policy/apex/plugins/event/carrier/kafka/ApexKafkaProducer.java @@ -0,0 +1,151 @@ +/*- + * ============LICENSE_START======================================================= + * Copyright (C) 2016-2018 Ericsson. All rights reserved. + * ================================================================================ + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + * SPDX-License-Identifier: Apache-2.0 + * ============LICENSE_END========================================================= + */ + +package org.onap.policy.apex.plugins.event.carrier.kafka; + +import java.util.EnumMap; +import java.util.Map; + +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.onap.policy.apex.service.engine.event.ApexEventException; +import org.onap.policy.apex.service.engine.event.ApexEventProducer; +import org.onap.policy.apex.service.engine.event.PeeredReference; +import org.onap.policy.apex.service.engine.event.SynchronousEventCache; +import org.onap.policy.apex.service.parameters.eventhandler.EventHandlerParameters; +import org.onap.policy.apex.service.parameters.eventhandler.EventHandlerPeeredMode; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Concrete implementation of an Apex event producer that sends events using Kafka. + * + * @author Liam Fallon (liam.fallon@ericsson.com) + */ +public class ApexKafkaProducer implements ApexEventProducer { + + // Get a reference to the logger + private static final Logger LOGGER = LoggerFactory.getLogger(ApexKafkaProducer.class); + + // The Kafka parameters read from the parameter service + private KAFKACarrierTechnologyParameters kafkaProducerProperties; + + // The Kafka Producer used to send events using Kafka + private Producer<String, Object> kafkaProducer; + + // The name for this producer + private String name = null; + + // The peer references for this event handler + private Map<EventHandlerPeeredMode, PeeredReference> peerReferenceMap = new EnumMap<>(EventHandlerPeeredMode.class); + + @Override + public void init(final String producerName, final EventHandlerParameters producerParameters) + throws ApexEventException { + this.name = producerName; + + // Check and get the Kafka Properties + if (!(producerParameters.getCarrierTechnologyParameters() instanceof KAFKACarrierTechnologyParameters)) { + LOGGER.warn("specified producer properties are not applicable to a Kafka producer (" + this.name + ")"); + throw new ApexEventException( + "specified producer properties are not applicable to a Kafka producer (" + this.name + ")"); + } + kafkaProducerProperties = + (KAFKACarrierTechnologyParameters) producerParameters.getCarrierTechnologyParameters(); + } + + /* + * (non-Javadoc) + * + * @see org.onap.policy.apex.service.engine.event.ApexEventProducer#getName() + */ + @Override + public String getName() { + return name; + } + + /* + * (non-Javadoc) + * + * @see org.onap.policy.apex.service.engine.event.ApexEventProducer#getPeeredReference(org.onap.policy.apex.service. + * parameters.eventhandler.EventHandlerPeeredMode) + */ + @Override + public PeeredReference getPeeredReference(final EventHandlerPeeredMode peeredMode) { + return peerReferenceMap.get(peeredMode); + } + + /* + * (non-Javadoc) + * + * @see org.onap.policy.apex.service.engine.event.ApexEventProducer#setPeeredReference(org.onap.policy.apex.service. + * parameters.eventhandler.EventHandlerPeeredMode, org.onap.policy.apex.service.engine.event.PeeredReference) + */ + @Override + public void setPeeredReference(final EventHandlerPeeredMode peeredMode, final PeeredReference peeredReference) { + peerReferenceMap.put(peeredMode, peeredReference); + } + + /* + * (non-Javadoc) + * + * @see org.onap.policy.apex.apps.uservice.producer.ApexEventProducer#sendEvent(long, java.lang.String, + * java.lang.Object) + */ + @Override + public void sendEvent(final long executionId, final String eventName, final Object event) { + // Check if this is a synchronized event, if so we have received a reply + final SynchronousEventCache synchronousEventCache = + (SynchronousEventCache) peerReferenceMap.get(EventHandlerPeeredMode.SYNCHRONOUS); + if (synchronousEventCache != null) { + synchronousEventCache.removeCachedEventToApexIfExists(executionId); + } + + // Kafka producer must be started in the same thread as it is stopped, so we must start it here + if (kafkaProducer == null) { + // Kick off the Kafka producer + kafkaProducer = new KafkaProducer<String, Object>(kafkaProducerProperties.getKafkaProducerProperties()); + if (LOGGER.isDebugEnabled()) { + LOGGER.debug("event producer " + this.name + " is ready to send to topics: " + + kafkaProducerProperties.getProducerTopic()); + } + } + + kafkaProducer.send(new ProducerRecord<String, Object>(kafkaProducerProperties.getProducerTopic(), name, event)); + if (LOGGER.isTraceEnabled()) { + LOGGER.trace("event sent from engine using {} to topic {} : {} ", this.name, + kafkaProducerProperties.getProducerTopic(), event); + } + } + + /* + * (non-Javadoc) + * + * @see org.onap.policy.apex.apps.uservice.producer.ApexEventProducer#stop() + */ + @Override + public void stop() { + if (kafkaProducer != null) { + kafkaProducer.flush(); + kafkaProducer.close(); + } + } +} diff --git a/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-kafka/src/main/java/org/onap/policy/apex/plugins/event/carrier/kafka/KAFKACarrierTechnologyParameters.java b/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-kafka/src/main/java/org/onap/policy/apex/plugins/event/carrier/kafka/KAFKACarrierTechnologyParameters.java new file mode 100644 index 000000000..2357b1807 --- /dev/null +++ b/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-kafka/src/main/java/org/onap/policy/apex/plugins/event/carrier/kafka/KAFKACarrierTechnologyParameters.java @@ -0,0 +1,396 @@ +/*- + * ============LICENSE_START======================================================= + * Copyright (C) 2016-2018 Ericsson. All rights reserved. + * ================================================================================ + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + * SPDX-License-Identifier: Apache-2.0 + * ============LICENSE_END========================================================= + */ + +package org.onap.policy.apex.plugins.event.carrier.kafka; + +import java.util.Arrays; +import java.util.Collection; +import java.util.Properties; + +import org.onap.policy.apex.service.parameters.carriertechnology.CarrierTechnologyParameters; + +/** + * Apex parameters for Kafka as an event carrier technology. + * + * @author Liam Fallon (liam.fallon@ericsson.com) + */ +public class KAFKACarrierTechnologyParameters extends CarrierTechnologyParameters { + // @formatter:off + /** The label of this carrier technology. */ + public static final String KAFKA_CARRIER_TECHNOLOGY_LABEL = "KAFKA"; + + /** The producer plugin class for the Kafka carrier technology. */ + public static final String KAFKA_EVENT_PRODUCER_PLUGIN_CLASS = ApexKafkaProducer.class.getCanonicalName(); + + /** The consumer plugin class for the Kafka carrier technology. */ + public static final String KAFKA_EVENT_CONSUMER_PLUGIN_CLASS = ApexKafkaConsumer.class.getCanonicalName(); + + // Default parameter values + private static final String DEFAULT_ACKS = "all"; + private static final String DEFAULT_BOOTSTRAP_SERVERS = "localhost:9092"; + private static final int DEFAULT_RETRIES = 0; + private static final int DEFAULT_BATCH_SIZE = 16384; + private static final int DEFAULT_LINGER_TIME = 1; + private static final long DEFAULT_BUFFER_MEMORY = 33554432; + private static final String DEFAULT_GROUP_ID = "default-group-id"; + private static final boolean DEFAULT_ENABLE_AUTO_COMMIT = true; + private static final int DEFAULT_AUTO_COMMIT_TIME = 1000; + private static final int DEFAULT_SESSION_TIMEOUT = 30000; + private static final String DEFAULT_PRODUCER_TOPIC = "apex-out"; + private static final int DEFAULT_CONSUMER_POLL_TIME = 100; + private static final String[] DEFAULT_CONSUMER_TOPIC_LIST = {"apex-in"}; + private static final String DEFAULT_KEY_SERIALIZER = "org.apache.kafka.common.serialization.StringSerializer"; + private static final String DEFAULT_VALUE_SERIALIZER = "org.apache.kafka.common.serialization.StringSerializer"; + private static final String DEFAULT_KEY_DESERIALIZER = "org.apache.kafka.common.serialization.StringDeserializer"; + private static final String DEFAULT_VALUE_DESERIALIZER = "org.apache.kafka.common.serialization.StringDeserializer"; + + // Parameter property map tokens + private static final String PROPERTY_BOOTSTRAP_SERVERS = "bootstrap.servers"; + private static final String PROPERTY_ACKS = "acks"; + private static final String PROPERTY_RETRIES = "retries"; + private static final String PROPERTY_BATCH_SIZE = "batch.size"; + private static final String PROPERTY_LINGER_TIME = "linger.ms"; + private static final String PROPERTY_BUFFER_MEMORY = "buffer.memory"; + private static final String PROPERTY_GROUP_ID = "group.id"; + private static final String PROPERTY_ENABLE_AUTO_COMMIT = "enable.auto.commit"; + private static final String PROPERTY_AUTO_COMMIT_TIME = "auto.commit.interval.ms"; + private static final String PROPERTY_SESSION_TIMEOUT = "session.timeout.ms"; + private static final String PROPERTY_KEY_SERIALIZER = "key.serializer"; + private static final String PROPERTY_VALUE_SERIALIZER = "value.serializer"; + private static final String PROPERTY_KEY_DESERIALIZER = "key.deserializer"; + private static final String PROPERTY_VALUE_DESERIALIZER = "value.deserializer"; + + // kafka carrier parameters + private String bootstrapServers = DEFAULT_BOOTSTRAP_SERVERS; + private String acks = DEFAULT_ACKS; + private int retries = DEFAULT_RETRIES; + private int batchSize = DEFAULT_BATCH_SIZE; + private int lingerTime = DEFAULT_LINGER_TIME; + private long bufferMemory = DEFAULT_BUFFER_MEMORY; + private String groupId = DEFAULT_GROUP_ID; + private boolean enableAutoCommit = DEFAULT_ENABLE_AUTO_COMMIT; + private int autoCommitTime = DEFAULT_AUTO_COMMIT_TIME; + private int sessionTimeout = DEFAULT_SESSION_TIMEOUT; + private String producerTopic = DEFAULT_PRODUCER_TOPIC; + private int consumerPollTime = DEFAULT_CONSUMER_POLL_TIME; + private String[] consumerTopicList = DEFAULT_CONSUMER_TOPIC_LIST; + private String keySerializer = DEFAULT_KEY_SERIALIZER; + private String valueSerializer = DEFAULT_VALUE_SERIALIZER; + private String keyDeserializer = DEFAULT_KEY_DESERIALIZER; + private String valueDeserializer = DEFAULT_VALUE_DESERIALIZER; + // @formatter:on + + /** + * Constructor to create a kafka carrier technology parameters instance and register the instance with the parameter + * service. + */ + public KAFKACarrierTechnologyParameters() { + super(KAFKACarrierTechnologyParameters.class.getCanonicalName()); + + // Set the carrier technology properties for the kafka carrier technology + this.setLabel(KAFKA_CARRIER_TECHNOLOGY_LABEL); + this.setEventProducerPluginClass(KAFKA_EVENT_PRODUCER_PLUGIN_CLASS); + this.setEventConsumerPluginClass(KAFKA_EVENT_CONSUMER_PLUGIN_CLASS); + } + + /** + * Gets the kafka producer properties. + * + * @return the kafka producer properties + */ + public Properties getKafkaProducerProperties() { + final Properties kafkaProperties = new Properties(); + + kafkaProperties.put(PROPERTY_BOOTSTRAP_SERVERS, bootstrapServers); + kafkaProperties.put(PROPERTY_ACKS, acks); + kafkaProperties.put(PROPERTY_RETRIES, retries); + kafkaProperties.put(PROPERTY_BATCH_SIZE, batchSize); + kafkaProperties.put(PROPERTY_LINGER_TIME, lingerTime); + kafkaProperties.put(PROPERTY_BUFFER_MEMORY, bufferMemory); + kafkaProperties.put(PROPERTY_KEY_SERIALIZER, keySerializer); + kafkaProperties.put(PROPERTY_VALUE_SERIALIZER, valueSerializer); + + return kafkaProperties; + } + + /** + * Gets the kafka consumer properties. + * + * @return the kafka consumer properties + */ + public Properties getKafkaConsumerProperties() { + final Properties kafkaProperties = new Properties(); + + kafkaProperties.put(PROPERTY_BOOTSTRAP_SERVERS, bootstrapServers); + kafkaProperties.put(PROPERTY_GROUP_ID, groupId); + kafkaProperties.put(PROPERTY_ENABLE_AUTO_COMMIT, enableAutoCommit); + kafkaProperties.put(PROPERTY_AUTO_COMMIT_TIME, autoCommitTime); + kafkaProperties.put(PROPERTY_SESSION_TIMEOUT, sessionTimeout); + kafkaProperties.put(PROPERTY_KEY_DESERIALIZER, keyDeserializer); + kafkaProperties.put(PROPERTY_VALUE_DESERIALIZER, valueDeserializer); + + return kafkaProperties; + } + + /** + * Gets the bootstrap servers. + * + * @return the bootstrap servers + */ + public String getBootstrapServers() { + return bootstrapServers; + } + + /** + * Gets the acks. + * + * @return the acks + */ + public String getAcks() { + return acks; + } + + /** + * Gets the retries. + * + * @return the retries + */ + public int getRetries() { + return retries; + } + + /** + * Gets the batch size. + * + * @return the batch size + */ + public int getBatchSize() { + return batchSize; + } + + /** + * Gets the linger time. + * + * @return the linger time + */ + public int getLingerTime() { + return lingerTime; + } + + /** + * Gets the buffer memory. + * + * @return the buffer memory + */ + public long getBufferMemory() { + return bufferMemory; + } + + /** + * Gets the group id. + * + * @return the group id + */ + public String getGroupId() { + return groupId; + } + + /** + * Checks if is enable auto commit. + * + * @return true, if checks if is enable auto commit + */ + public boolean isEnableAutoCommit() { + return enableAutoCommit; + } + + /** + * Gets the auto commit time. + * + * @return the auto commit time + */ + public int getAutoCommitTime() { + return autoCommitTime; + } + + /** + * Gets the session timeout. + * + * @return the session timeout + */ + public int getSessionTimeout() { + return sessionTimeout; + } + + /** + * Gets the producer topic. + * + * @return the producer topic + */ + public String getProducerTopic() { + return producerTopic; + } + + /** + * Gets the consumer poll time. + * + * @return the consumer poll time + */ + public long getConsumerPollTime() { + return consumerPollTime; + } + + /** + * Gets the consumer topic list. + * + * @return the consumer topic list + */ + public Collection<String> getConsumerTopicList() { + return Arrays.asList(consumerTopicList); + } + + /** + * Gets the key serializer. + * + * @return the key serializer + */ + public String getKeySerializer() { + return keySerializer; + } + + /** + * Gets the value serializer. + * + * @return the value serializer + */ + public String getValueSerializer() { + return valueSerializer; + } + + /** + * Gets the key deserializer. + * + * @return the key deserializer + */ + public String getKeyDeserializer() { + return keyDeserializer; + } + + /** + * Gets the value deserializer. + * + * @return the value deserializer + */ + public String getValueDeserializer() { + return valueDeserializer; + } + + /* + * (non-Javadoc) + * + * @see org.onap.policy.apex.apps.uservice.parameters.ApexParameterValidator#validate() + */ + @Override + public String validate() { + final StringBuilder errorMessageBuilder = new StringBuilder(); + + errorMessageBuilder.append(super.validate()); + + if (bootstrapServers == null || bootstrapServers.trim().length() == 0) { + errorMessageBuilder + .append(" bootstrapServers not specified, must be specified as a string of form host:port\n"); + } + + if (acks == null || acks.trim().length() == 0) { + errorMessageBuilder.append(" acks not specified, must be specified as a string with values [0|1|all]\n"); + } + + if (retries < 0) { + errorMessageBuilder.append(" retries [" + retries + "] invalid, must be specified as retries >= 0\n"); + } + + if (batchSize < 0) { + errorMessageBuilder + .append(" batchSize [" + batchSize + "] invalid, must be specified as batchSize >= 0\n"); + } + + if (lingerTime < 0) { + errorMessageBuilder + .append(" lingerTime [" + lingerTime + "] invalid, must be specified as lingerTime >= 0\n"); + } + + if (bufferMemory < 0) { + errorMessageBuilder + .append(" bufferMemory [" + bufferMemory + "] invalid, must be specified as bufferMemory >= 0\n"); + } + + if (groupId == null || groupId.trim().length() == 0) { + errorMessageBuilder.append(" groupId not specified, must be specified as a string\n"); + } + + if (autoCommitTime < 0) { + errorMessageBuilder.append( + " autoCommitTime [" + autoCommitTime + "] invalid, must be specified as autoCommitTime >= 0\n"); + } + + if (sessionTimeout < 0) { + errorMessageBuilder.append( + " sessionTimeout [" + sessionTimeout + "] invalid, must be specified as sessionTimeout >= 0\n"); + } + + if (producerTopic == null || producerTopic.trim().length() == 0) { + errorMessageBuilder.append(" producerTopic not specified, must be specified as a string\n"); + } + + if (consumerPollTime < 0) { + errorMessageBuilder.append(" consumerPollTime [" + consumerPollTime + + "] invalid, must be specified as consumerPollTime >= 0\n"); + } + + if (consumerTopicList == null || consumerTopicList.length == 0) { + errorMessageBuilder.append(" consumerTopicList not specified, must be specified as a list of strings\n"); + } + + for (final String consumerTopic : consumerTopicList) { + if (consumerTopic == null || consumerTopic.trim().length() == 0) { + errorMessageBuilder.append(" invalid consumer topic \"" + consumerTopic + + "\" specified on consumerTopicList, consumer topics must be specified as strings\n"); + } + } + + if (keySerializer == null || keySerializer.trim().length() == 0) { + errorMessageBuilder.append(" keySerializer not specified, must be specified as a string\n"); + } + + if (valueSerializer == null || valueSerializer.trim().length() == 0) { + errorMessageBuilder.append(" valueSerializer not specified, must be specified as a string\n"); + } + + if (keyDeserializer == null || keyDeserializer.trim().length() == 0) { + errorMessageBuilder.append(" keyDeserializer not specified, must be specified as a string\n"); + } + + if (valueDeserializer == null || valueDeserializer.trim().length() == 0) { + errorMessageBuilder.append(" valueDeserializer not specified, must be specified as a string\n"); + } + + return errorMessageBuilder.toString(); + } +} diff --git a/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-kafka/src/main/java/org/onap/policy/apex/plugins/event/carrier/kafka/package-info.java b/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-kafka/src/main/java/org/onap/policy/apex/plugins/event/carrier/kafka/package-info.java new file mode 100644 index 000000000..f04aaceef --- /dev/null +++ b/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-kafka/src/main/java/org/onap/policy/apex/plugins/event/carrier/kafka/package-info.java @@ -0,0 +1,27 @@ +/*- + * ============LICENSE_START======================================================= + * Copyright (C) 2016-2018 Ericsson. All rights reserved. + * ================================================================================ + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + * SPDX-License-Identifier: Apache-2.0 + * ============LICENSE_END========================================================= + */ + +/** + * Implements the APEX event carrier technology plugin for Kafka. + * + * @author Liam Fallon (liam.fallon@ericsson.com) + */ + +package org.onap.policy.apex.plugins.event.carrier.kafka; |