Kafka Application - Updated the Application to a Distributed Application
1. Distributed Work Queue to publish events.
2. Only one instance can publish to queue and only one instance can consume from the queue.
3. KafkaPublisherAdminService which does start,stop and restart a kafka producer.
3. Other cosmectic changes (some files have been renamed)
Change-Id: I15b2015ed303eae1c9fb9f4820fb14476726161e
diff --git a/apps/kafka-integration/api/src/main/java/org/onosproject/kafkaintegration/api/KafkaEventStorageService.java b/apps/kafka-integration/api/src/main/java/org/onosproject/kafkaintegration/api/KafkaEventStorageService.java
index ae6b720..0d0cc57 100644
--- a/apps/kafka-integration/api/src/main/java/org/onosproject/kafkaintegration/api/KafkaEventStorageService.java
+++ b/apps/kafka-integration/api/src/main/java/org/onosproject/kafkaintegration/api/KafkaEventStorageService.java
@@ -23,18 +23,16 @@
public interface KafkaEventStorageService {
/**
- * Inserts the Generated event into the local cache.
+ * Inserts the Onos Event into Distributed Work Queue.
*
* @param e the ONOS Event
- * @return true if the insertion was successful
*/
- boolean insertCacheEntry(OnosEvent e);
+ void publishEvent(OnosEvent event);
/**
- * Updates the counter with the most recently published event's sequence
- * number.
+ * Removes the Onos Event from the Distributed Work Queue.
*
- * @param sequenceNumber the updated value of sequence number.
+ * @return the Onos Event
*/
- void updateLastPublishedEntry(Long sequenceNumber);
+ OnosEvent consumeEvent();
}
diff --git a/apps/kafka-integration/api/src/main/java/org/onosproject/kafkaintegration/api/KafkaPublisherAdminService.java b/apps/kafka-integration/api/src/main/java/org/onosproject/kafkaintegration/api/KafkaPublisherAdminService.java
new file mode 100644
index 0000000..181e903
--- /dev/null
+++ b/apps/kafka-integration/api/src/main/java/org/onosproject/kafkaintegration/api/KafkaPublisherAdminService.java
@@ -0,0 +1,41 @@
+/**
+ * Copyright 2016-present Open Networking Laboratory
+ *
+ * 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.
+ */
+package org.onosproject.kafkaintegration.api;
+
+import org.onosproject.kafkaintegration.api.dto.KafkaServerConfig;
+
+
+public interface KafkaPublisherAdminService {
+ /**
+ * Starts the Kafka Producer.
+ *
+ * @param config the Kafka Server Config
+ */
+ void start(KafkaServerConfig config);
+
+ /**
+ * Stops the Kafka Producer.
+ *
+ */
+ void stop();
+
+ /**
+ * Restarts the Kafka Producer.
+ *
+ * @param config the Kafka Server Config
+ */
+ void restart(KafkaServerConfig config);
+}
diff --git a/apps/kafka-integration/api/src/main/java/org/onosproject/kafkaintegration/api/KafkaPublisherService.java b/apps/kafka-integration/api/src/main/java/org/onosproject/kafkaintegration/api/KafkaPublisherService.java
index 6156134..d1b1f2f 100644
--- a/apps/kafka-integration/api/src/main/java/org/onosproject/kafkaintegration/api/KafkaPublisherService.java
+++ b/apps/kafka-integration/api/src/main/java/org/onosproject/kafkaintegration/api/KafkaPublisherService.java
@@ -16,41 +16,19 @@
package org.onosproject.kafkaintegration.api;
import java.util.concurrent.Future;
-
-import org.onosproject.kafkaintegration.api.dto.KafkaServerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
+
/**
* APIs for controlling the Kafka Producer.
*
*/
public interface KafkaPublisherService {
-
- /**
- * Starts the Kafka Producer.
- *
- * @param config the Kafka Server Config
- */
- void start(KafkaServerConfig config);
-
- /**
- * Stops the Kafka Producer.
- *
- */
- void stop();
-
- /**
- * Restarts the Kafka Producer.
- *
- * @param config the Kafka Server Config
- */
- void restart(KafkaServerConfig config);
-
/**
* Sends message to Kafka Server.
*
* @param record a message to be sent
* @return metadata for a record that as been acknowledged
*/
- public Future<RecordMetadata> send(ProducerRecord<String, byte[]> record);
+ Future<RecordMetadata> send(ProducerRecord<String, byte[]> record);
}
diff --git a/apps/kafka-integration/api/src/main/java/org/onosproject/kafkaintegration/api/dto/OnosEvent.java b/apps/kafka-integration/api/src/main/java/org/onosproject/kafkaintegration/api/dto/OnosEvent.java
index 9b28eeb..72fa028 100644
--- a/apps/kafka-integration/api/src/main/java/org/onosproject/kafkaintegration/api/dto/OnosEvent.java
+++ b/apps/kafka-integration/api/src/main/java/org/onosproject/kafkaintegration/api/dto/OnosEvent.java
@@ -16,14 +16,12 @@
import org.onosproject.event.AbstractEvent;
-import com.google.protobuf.GeneratedMessageV3;
-
/**
* Represents the converted Onos Event data into protobuf format.
*
*/
// FIXME lack of abstraction in subject type is biting us
-public class OnosEvent extends AbstractEvent<OnosEvent.Type, GeneratedMessageV3> {
+public class OnosEvent extends AbstractEvent<OnosEvent.Type, byte[]> {
/**
* Creates a new Onos Event.
@@ -31,7 +29,7 @@
* @param type The Type of Onos Event
* @param subject Protobuf message corresponding to the Onos Event
*/
- public OnosEvent(Type type, GeneratedMessageV3 subject) {
+ public OnosEvent(Type type, byte[] subject) {
super(type, subject);
}