You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streams.apache.org by sb...@apache.org on 2014/02/11 23:18:00 UTC
svn commit: r1567420 - in
/incubator/streams/trunk/streams-contrib/streams-persist-kafka/src/main/java/org/apache/streams/kafka:
KafkaPersistReader.java KafkaPersistReaderTask.java
Author: sblackmon
Date: Tue Feb 11 22:17:59 2014
New Revision: 1567420
URL: http://svn.apache.org/r1567420
Log:
added kafka reader capability - not yet tested!
Added:
incubator/streams/trunk/streams-contrib/streams-persist-kafka/src/main/java/org/apache/streams/kafka/KafkaPersistReader.java
incubator/streams/trunk/streams-contrib/streams-persist-kafka/src/main/java/org/apache/streams/kafka/KafkaPersistReaderTask.java
Added: incubator/streams/trunk/streams-contrib/streams-persist-kafka/src/main/java/org/apache/streams/kafka/KafkaPersistReader.java
URL: http://svn.apache.org/viewvc/incubator/streams/trunk/streams-contrib/streams-persist-kafka/src/main/java/org/apache/streams/kafka/KafkaPersistReader.java?rev=1567420&view=auto
==============================================================================
--- incubator/streams/trunk/streams-contrib/streams-persist-kafka/src/main/java/org/apache/streams/kafka/KafkaPersistReader.java (added)
+++ incubator/streams/trunk/streams-contrib/streams-persist-kafka/src/main/java/org/apache/streams/kafka/KafkaPersistReader.java Tue Feb 11 22:17:59 2014
@@ -0,0 +1,141 @@
+package org.apache.streams.kafka;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.typesafe.config.Config;
+import kafka.consumer.Consumer;
+import kafka.consumer.ConsumerConfig;
+import kafka.consumer.KafkaStream;
+import kafka.consumer.Whitelist;
+import kafka.javaapi.consumer.ConsumerConnector;
+import kafka.serializer.StringDecoder;
+import kafka.utils.VerifiableProperties;
+import org.apache.streams.config.StreamsConfigurator;
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.core.StreamsPersistReader;
+import org.apache.streams.core.StreamsResultSet;
+import org.joda.time.DateTime;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Serializable;
+import java.math.BigInteger;
+import java.util.List;
+import java.util.Properties;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+public class KafkaPersistReader implements StreamsPersistReader, Serializable, Runnable {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(KafkaPersistReader.class);
+
+ protected volatile Queue<StreamsDatum> persistQueue;
+
+ private ObjectMapper mapper = new ObjectMapper();
+
+ private KafkaConfiguration config;
+
+ private ConsumerConfig consumerConfig;
+ private ConsumerConnector consumerConnector;
+
+ public List<KafkaStream<String, String>> inStreams;
+
+ private ExecutorService executor = Executors.newSingleThreadExecutor();
+
+ public KafkaPersistReader() {
+ Config config = StreamsConfigurator.config.getConfig("kafka");
+ this.config = KafkaConfigurator.detectConfiguration(config);
+ this.persistQueue = new ConcurrentLinkedQueue<StreamsDatum>();
+ }
+
+ public KafkaPersistReader(Queue<StreamsDatum> persistQueue) {
+ Config config = StreamsConfigurator.config.getConfig("kafka");
+ this.config = KafkaConfigurator.detectConfiguration(config);
+ this.persistQueue = persistQueue;
+ }
+
+ public KafkaPersistReader(KafkaConfiguration config) {
+ this.config = config;
+ this.persistQueue = new ConcurrentLinkedQueue<StreamsDatum>();
+ }
+
+ public KafkaPersistReader(KafkaConfiguration config, Queue<StreamsDatum> persistQueue) {
+ this.config = config;
+ this.persistQueue = persistQueue;
+ }
+
+ @Override
+ public void start() {
+ Properties props = new Properties();
+ props.setProperty("serializer.encoding", "UTF8");
+
+ consumerConfig = new ConsumerConfig(props);
+
+ consumerConnector = Consumer.createJavaConsumerConnector(consumerConfig);
+
+ Whitelist topics = new Whitelist(config.getTopic());
+ VerifiableProperties vprops = new VerifiableProperties(props);
+
+ inStreams = consumerConnector.createMessageStreamsByFilter(topics, 1, new StringDecoder(vprops), new StringDecoder(vprops));
+
+ for (final KafkaStream stream : inStreams) {
+ executor.submit(new KafkaPersistReaderTask(this, stream));
+ }
+
+ }
+
+ @Override
+ public void stop() {
+ consumerConnector.shutdown();
+ while( !executor.isTerminated()) {
+ try {
+ executor.awaitTermination(5, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {}
+ }
+ }
+
+ @Override
+ public void setPersistQueue(Queue<StreamsDatum> persistQueue) {
+ this.persistQueue = persistQueue;
+ }
+
+ @Override
+ public Queue<StreamsDatum> getPersistQueue() {
+ return this.persistQueue;
+ }
+
+ @Override
+ public StreamsResultSet readAll() {
+ return null;
+ }
+
+ @Override
+ public StreamsResultSet readNew(BigInteger bigInteger) {
+ return null;
+ }
+
+ @Override
+ public StreamsResultSet readRange(DateTime dateTime, DateTime dateTime2) {
+ return null;
+ }
+
+ private static ConsumerConfig createConsumerConfig(String a_zookeeper, String a_groupId) {
+ Properties props = new Properties();
+ props.put("zookeeper.connect", a_zookeeper);
+ props.put("group.id", a_groupId);
+ props.put("zookeeper.session.timeout.ms", "400");
+ props.put("zookeeper.sync.time.ms", "200");
+ props.put("auto.commit.interval.ms", "1000");
+ return new ConsumerConfig(props);
+ }
+
+ @Override
+ public void run() {
+ start();
+
+ // once this class can be told when to shutdown by streams, it will run stop
+ // stop();
+ }
+}
Added: incubator/streams/trunk/streams-contrib/streams-persist-kafka/src/main/java/org/apache/streams/kafka/KafkaPersistReaderTask.java
URL: http://svn.apache.org/viewvc/incubator/streams/trunk/streams-contrib/streams-persist-kafka/src/main/java/org/apache/streams/kafka/KafkaPersistReaderTask.java?rev=1567420&view=auto
==============================================================================
--- incubator/streams/trunk/streams-contrib/streams-persist-kafka/src/main/java/org/apache/streams/kafka/KafkaPersistReaderTask.java (added)
+++ incubator/streams/trunk/streams-contrib/streams-persist-kafka/src/main/java/org/apache/streams/kafka/KafkaPersistReaderTask.java Tue Feb 11 22:17:59 2014
@@ -0,0 +1,44 @@
+package org.apache.streams.kafka;
+
+import kafka.consumer.ConsumerIterator;
+import kafka.consumer.KafkaStream;
+import kafka.message.MessageAndMetadata;
+import org.apache.streams.core.StreamsDatum;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Random;
+
+public class KafkaPersistReaderTask implements Runnable {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(KafkaPersistReaderTask.class);
+
+ private KafkaPersistReader reader;
+ private KafkaStream<String,String> stream;
+
+ public KafkaPersistReaderTask(KafkaPersistReader reader, KafkaStream<String,String> stream) {
+ this.reader = reader;
+ this.stream = stream;
+ }
+
+
+
+ @Override
+ public void run() {
+
+ MessageAndMetadata<String,String> item;
+ while(true) {
+
+ ConsumerIterator<String, String> it = stream.iterator();
+ while (it.hasNext()) {
+ item = it.next();
+ reader.persistQueue.add(new StreamsDatum(item.message()));
+ }
+ try {
+ Thread.sleep(new Random().nextInt(100));
+ } catch (InterruptedException e) {}
+ }
+
+ }
+
+}