Experimental kafa simple consumer based firehose

common/src/main/java/io/druid/common/utils/SerializerUtils.java

@@ -1,25 +1,28 @@
 /*
- * Druid - a distributed column store.
- * Copyright 2012 - 2015 Metamarkets Group Inc.
- *
- * 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.
- */
+* Licensed to Metamarkets Group Inc. (Metamarkets) under one
+* or more contributor license agreements. See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership. Metamarkets licenses this file
+* to you 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 io.druid.common.utils;
 
 import com.google.common.io.ByteStreams;
 import com.google.common.io.OutputSupplier;
 import com.google.common.primitives.Ints;
+import com.metamx.common.StringUtils;
 import io.druid.collections.IntList;
 
 import java.io.IOException;
@@ -61,15 +64,20 @@ public String readString(InputStream in) throws IOException
     final int length = readInt(in);
     byte[] stringBytes = new byte[length];
     ByteStreams.readFully(in, stringBytes);
-    return new String(stringBytes, UTF8);
+    return StringUtils.fromUtf8(stringBytes);
   }
 
   public String readString(ByteBuffer in) throws IOException
   {
     final int length = in.getInt();
-    byte[] stringBytes = new byte[length];
-    in.get(stringBytes);
-    return new String(stringBytes, UTF8);
+    return StringUtils.fromUtf8(readBytes(in, length));
+  }
+
+  public byte[] readBytes(ByteBuffer in, int length) throws IOException
+  {
+    byte[] bytes = new byte[length];
+    in.get(bytes);
+    return bytes;
   }
 
   public void writeStrings(OutputStream out, String[] names) throws IOException

docs/content/development/experimental.md

@@ -11,4 +11,4 @@ To enable experimental features, include their artifacts in the configuration ru
 druid.extensions.coordinates=["io.druid.extensions:druid-histogram:{VERSION}"]
 ```
 
-The configuration for all the indexer and query nodes need to be updated with this.
+The configuration for all the indexer and query nodes need to be updated with this.

docs/content/development/kafka-simple-consumer-firehose.md

@@ -0,0 +1,30 @@
+---
+layout: doc_page
+---
+# KafkaSimpleConsumerFirehose
+This is an experimental firehose to ingest data from kafka using kafka simple consumer api. Currently, this firehose would only work inside standalone realtime nodes.
+The configuration for KafkaSimpleConsumerFirehose is similar to the KafkaFirehose [Kafka firehose example](realtime-ingestion.html#realtime-specfile), except `firehose` should be replaced with `firehoseV2` like this:
+```json
+"firehoseV2": {
+"type" : "kafka-0.8-v2",
+"brokerList" :  ["localhost:4443"],
+"queueBufferLength":10001,
+"resetOffsetToEarliest":"true",
+"partitionIdList" : ["0"],
+"clientId" : "localclient",
+"feed": "wikipedia"
+}
+```
+
+|property|description|required?|
+|--------|-----------|---------|
+|type|kafka-0.8-v2|yes|
+|brokerList|list of the kafka brokers|yes|
+|queueBufferLength|the buffer length for kafka message queue|no default(20000)|
+|resetOffsetToEarliest|in case of kafkaOffsetOutOfRange error happens, consumer should starts from the earliest or latest message available|true|
+|partitionIdList|list of kafka partition ids|yes|
+|clientId|the clientId for kafka SimpleConsumer|yes|
+|feed|kafka topic|yes|
+
+For using this firehose at scale and possibly in production, it is recommended to set replication factor to at least three, which means at least three Kafka brokers in the `brokerList`. For a 1*10^4 events per second kafka topic, keeping one partition can work properly, but more partitions could be added if higher throughput is required.
+

docs/content/ingestion/firehose.md

@@ -260,7 +260,6 @@ When using this firehose, events can be sent by submitting a POST request to the
 |serviceName|name used to announce the event receiver service endpoint|yes|
 |bufferSize| size of buffer used by firehose to store events|no default(100000)|
 
-
 #### TimedShutoffFirehose
 
 This can be used to start a firehose that will shut down at a specified time.
@@ -283,3 +282,5 @@ An example is shown below:
 |type|This should be "timed"|yes|
 |shutoffTime|time at which the firehose should shut down, in ISO8601 format|yes|
 |delegate|firehose to use|yes|
+=======
+

docs/content/toc.textile

@@ -90,6 +90,7 @@ h2. Development
 ** "Select Query":../development/select-query.html
 ** "Approximate Histograms and Quantiles":../development/approximate-histograms.html
 ** "Router node":../development/router.html
+** "New Kafka Firehose":../development/kafka-simple-consumer-firehose.html
 
 h2. Misc
 * "Papers & Talks":../misc/papers-and-talks.html

extensions/kafka-eight-simpleConsumer/pom.xml

@@ -0,0 +1,81 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+~ Druid - a distributed column store.
+~ Copyright 2012 - 2015 Metamarkets Group Inc.
+~
+~ 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.
+-->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+  <groupId>io.druid.extensions</groupId>
+  <artifactId>druid-kafka-eight-simple-consumer</artifactId>
+  <name>druid-kafka-eight-simple-consumer</name>
+  <description>druid-kafka-eight-simple-consumer</description>
+
+  <parent>
+    <groupId>io.druid</groupId>
+    <artifactId>druid</artifactId>
+    <version>0.8.2-SNAPSHOT</version>
+    <relativePath>../../pom.xml</relativePath>
+  </parent>
+
+  <dependencies>
+    <dependency>
+      <groupId>io.druid</groupId>
+      <artifactId>druid-api</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>com.metamx</groupId>
+      <artifactId>emitter</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.kafka</groupId>
+      <artifactId>kafka_2.10</artifactId>
+      <version>0.8.2.1</version>
+      <exclusions>
+        <exclusion>
+          <groupId>log4j</groupId>
+          <artifactId>log4j</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>org.apache.zookeeper</groupId>
+          <artifactId>zookeeper</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+    <!-- Tests -->
+    <dependency>
+      <groupId>junit</groupId>
+      <artifactId>junit</artifactId>
+      <scope>test</scope>
+    </dependency>
+  </dependencies>
+
+  <build>
+    <plugins>
+      <plugin>
+        <artifactId>maven-jar-plugin</artifactId>
+        <configuration>
+          <archive>
+            <manifest>
+              <addDefaultImplementationEntries>true</addDefaultImplementationEntries>
+              <addDefaultSpecificationEntries>true</addDefaultSpecificationEntries>
+            </manifest>
+          </archive>
+        </configuration>
+      </plugin>
+    </plugins>
+  </build>
+</project>

extensions/kafka-eight-simpleConsumer/src/main/java/io/druid/firehose/kafka/KafkaEightSimpleConsumerDruidModule.java

@@ -0,0 +1,48 @@
+/*
+* Licensed to Metamarkets Group Inc. (Metamarkets) under one
+* or more contributor license agreements. See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership. Metamarkets licenses this file
+* to you 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 io.druid.firehose.kafka;
+
+import com.fasterxml.jackson.databind.Module;
+import com.fasterxml.jackson.databind.jsontype.NamedType;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+import com.google.common.collect.ImmutableList;
+import com.google.inject.Binder;
+import io.druid.initialization.DruidModule;
+
+import java.util.List;
+
+public class KafkaEightSimpleConsumerDruidModule implements DruidModule
+{
+  @Override
+  public List<? extends Module> getJacksonModules()
+  {
+    return ImmutableList.of(
+        new SimpleModule("KafkaEightSimpleConsumerFirehoseModule").registerSubtypes(
+            new NamedType(KafkaEightSimpleConsumerFirehoseFactory.class, "kafka-0.8-v2")
+        )
+    );
+  }
+
+  @Override
+  public void configure(Binder binder)
+  {
+
+  }
+}