Support of Kafka 0.11.0.0 (#117)

* Initial support for 0.11.0.0 * Added new transaction related properties to config

Author: simonsouter

akka/src/it/scala/cakesolutions/kafka/akka/KafkaConsumerActorPerfSpec.scala

@@ -3,8 +3,7 @@ package cakesolutions.kafka.akka
 import akka.actor.{ActorRef, ActorSystem}
 import akka.testkit.TestActor.AutoPilot
 import akka.testkit.{ImplicitSender, TestActor, TestKit, TestProbe}
-import cakesolutions.kafka.akka.KafkaConsumerActor.Confirm
-import cakesolutions.kafka.akka.KafkaConsumerActor.Subscribe.AutoPartition
+import cakesolutions.kafka.akka.KafkaConsumerActor.{Confirm, Subscribe}
 import cakesolutions.kafka.{KafkaConsumer, KafkaProducer, KafkaProducerRecord}
 import com.typesafe.config.ConfigFactory
 import org.apache.kafka.clients.consumer.ConsumerConfig
@@ -73,7 +72,7 @@ class KafkaConsumerActorPerfSpec(system_ : ActorSystem)
  producer.flush()
  log.info("Delivered {} messages to topic {}", totalMessages, topic)
 
- consumer.subscribe(AutoPartition(Seq(topic)))
+ consumer.subscribe(Subscribe.AutoPartition(Seq(topic)))
 
  whenReady(pilot.future) { case (totalTime, messagesPerSec) =>
  log.info("Total Time millis : {}", totalTime)

akka/src/it/scala/cakesolutions/kafka/akka/KafkaE2EActorPerfSpec.scala

@@ -3,8 +3,7 @@ package cakesolutions.kafka.akka
 import akka.actor.{ActorRef, ActorSystem}
 import akka.testkit.TestActor.AutoPilot
 import akka.testkit.{ImplicitSender, TestActor, TestKit, TestProbe}
-import cakesolutions.kafka.akka.KafkaConsumerActor.Confirm
-import cakesolutions.kafka.akka.KafkaConsumerActor.Subscribe.AutoPartition
+import cakesolutions.kafka.akka.KafkaConsumerActor.{Confirm, Subscribe}
 import cakesolutions.kafka.{KafkaConsumer, KafkaProducer, KafkaProducerRecord}
 import com.typesafe.config.ConfigFactory
 import org.apache.kafka.common.serialization.{StringDeserializer, StringSerializer}
@@ -78,7 +77,7 @@ class KafkaE2EActorPerfSpec(system_ : ActorSystem)
  testProducer.flush()
  log.info("Delivered {} messages to topic {}", totalMessages, sourceTopic)
 
- consumer.subscribe(AutoPartition(Seq(sourceTopic)))
+ consumer.subscribe(Subscribe.AutoPartition(Seq(sourceTopic)))
 
  whenReady(pilot.future) { case (totalTime, messagesPerSec) =>
  log.info("Total Time millis : {}", totalTime)

akka/src/test/scala/cakesolutions/kafka/akka/KafkaConsumerActorRecoverySpec.scala

@@ -1,7 +1,6 @@
 package cakesolutions.kafka.akka
 
 import akka.actor.{Actor, ActorLogging, ActorRef, ActorSystem, OneForOneStrategy, Props, SupervisorStrategy}
-import cakesolutions.kafka.akka.KafkaConsumerActor.Subscribe.{ManualOffset, ManualPartition}
 import cakesolutions.kafka.akka.KafkaConsumerActor.{Confirm, Subscribe, TriggerConsumerFailure, Unsubscribe}
 import cakesolutions.kafka.{KafkaConsumer, KafkaProducerRecord, KafkaTopicPartition}
 import org.apache.kafka.clients.consumer.OffsetResetStrategy
@@ -119,7 +118,7 @@ class KafkaConsumerActorRecoverySpec(_system: ActorSystem) extends KafkaIntSpec(
  producer.flush()
 
  val consumer = KafkaConsumerActor(consumerConf, KafkaConsumerActor.Conf(), testActor)
- consumer.subscribe(ManualPartition(List(topicPartition)))
+ consumer.subscribe(Subscribe.ManualPartition(List(topicPartition)))
 
  val rec1 = expectMsgClass(30.seconds, classOf[ConsumerRecords[String, String]])
  rec1.offsets.get(topicPartition) shouldBe Some(1)
@@ -141,7 +140,7 @@ class KafkaConsumerActorRecoverySpec(_system: ActorSystem) extends KafkaIntSpec(
 
  // Reset subscription
  consumer.unsubscribe()
- consumer.subscribe(ManualOffset(offsets))
+ consumer.subscribe(Subscribe.ManualOffset(offsets))
 
  // New subscription starts from specified offset
  val rec3 = expectMsgClass(30.seconds, classOf[ConsumerRecords[String, String]])

akka/src/test/scala/cakesolutions/kafka/akka/KafkaConsumerActorSpec.scala

@@ -2,7 +2,6 @@ package cakesolutions.kafka.akka
 
 import akka.actor.{ActorSystem, PoisonPill}
 import akka.testkit.TestProbe
-import cakesolutions.kafka.akka.KafkaConsumerActor.Subscribe.AutoPartition
 import cakesolutions.kafka.akka.KafkaConsumerActor.{Confirm, Subscribe, Unsubscribe}
 import cakesolutions.kafka.{KafkaConsumer, KafkaProducer, KafkaProducerRecord, KafkaTopicPartition}
 import com.typesafe.config.{Config, ConfigFactory}
@@ -80,7 +79,7 @@ class KafkaConsumerActorSpec(system_ : ActorSystem) extends KafkaIntSpec(system_
  producer.flush()
 
  val consumer = KafkaConsumerActor(consumerConfig, actorConf, testActor)
- consumer.subscribe(AutoPartition(Seq(topic)))
+ consumer.subscribe(Subscribe.AutoPartition(Seq(topic)))
 
  val rs = expectMsgClass(30.seconds, classOf[ConsumerRecords[String, String]])
  consumer.confirm(rs.offsets)

client/src/it/scala/cakesolutions/kafka/KafkaConsumerPerfSpec.scala

@@ -6,7 +6,7 @@ import org.apache.kafka.common.serialization.{StringDeserializer, StringSerializ
 import org.scalatest.{BeforeAndAfterAll, FlatSpecLike, Matchers}
 import org.slf4j.LoggerFactory
 
-import scala.collection.JavaConversions._
+import scala.collection.JavaConverters._
 import scala.util.Random
 
 /**
@@ -42,7 +42,7 @@ class KafkaConsumerPerfSpec extends FlatSpecLike
  producer.flush()
  log.info("Delivered 100000 msg to topic {}", topic)
 
- consumer.subscribe(List(topic))
+ consumer.subscribe(List(topic).asJava)
 
  var start = 0l
 

client/src/main/scala/cakesolutions/kafka/KafkaConsumer.scala

@@ -4,7 +4,9 @@ import cakesolutions.kafka.TypesafeConfigExtensions._
 import com.typesafe.config.Config
 import org.apache.kafka.clients.consumer.{ConsumerConfig, OffsetResetStrategy, KafkaConsumer => JKafkaConsumer}
 import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.requests.IsolationLevel
 import org.apache.kafka.common.serialization.Deserializer
+
 import scala.collection.JavaConverters._
 import scala.language.implicitConversions
 
@@ -60,7 +62,8 @@ object KafkaConsumer {
  maxPollRecords: Int = 500,
  maxPollInterval: Int = 300000,
  maxMetaDataAge : Long = 300000, 
- autoOffsetReset: OffsetResetStrategy = OffsetResetStrategy.LATEST
+ autoOffsetReset: OffsetResetStrategy = OffsetResetStrategy.LATEST,
+ isolationLevel: IsolationLevel = IsolationLevel.READ_UNCOMMITTED
  ): Conf[K, V] = {
 
  val configMap = Map[String, AnyRef](
@@ -73,7 +76,8 @@ object KafkaConsumer {
  ConsumerConfig.MAX_POLL_RECORDS_CONFIG -> maxPollRecords.toString,
  ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG -> maxPollInterval.toString,
  ConsumerConfig.METADATA_MAX_AGE_CONFIG ->maxMetaDataAge.toString,
- ConsumerConfig.AUTO_OFFSET_RESET_CONFIG -> autoOffsetReset.toString.toLowerCase
+ ConsumerConfig.AUTO_OFFSET_RESET_CONFIG -> autoOffsetReset.toString.toLowerCase,
+ ConsumerConfig.ISOLATION_LEVEL_CONFIG -> isolationLevel.toString.toLowerCase()
  )
 
  apply(configMap, keyDeserializer, valueDeserializer)