Fix compiler warning & thread was not terminated (#137)

* Remove unused input parameter

Kafka Consumer Configs didn't have configurations related to topic.

* Fix thread was not terminated correctly.

Author: jiminhsieh

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

@@ -54,7 +54,7 @@ class KafkaConsumerActorSpec(system_ : ActorSystem) extends KafkaIntSpec(system_
         """.stripMargin)
     )
 
-  def configuredActor(topic: String): Config =
+  def configuredActor: Config =
     ConfigFactory.parseString(
       s"""
          | bootstrap.servers = "localhost:$kafkaPort",
@@ -98,7 +98,7 @@ class KafkaConsumerActorSpec(system_ : ActorSystem) extends KafkaIntSpec(system_
     producer.flush()
 
     // Consumer and actor config in same config file
-    val consumer = system.actorOf(KafkaConsumerActor.props(configuredActor(topic), new StringDeserializer(), new StringDeserializer(), testActor))
+    val consumer = system.actorOf(KafkaConsumerActor.props(configuredActor, new StringDeserializer(), new StringDeserializer(), testActor))
     consumer ! Subscribe.AutoPartition(List(topic))
 
     val rs = expectMsgClass(30.seconds, classOf[ConsumerRecords[String, String]])
@@ -120,7 +120,7 @@ class KafkaConsumerActorSpec(system_ : ActorSystem) extends KafkaIntSpec(system_
     val downstreamActor = TestProbe().ref
 
     // Consumer and actor config in same config file
-    val consumer = system.actorOf(KafkaConsumerActor.props(configuredActor(topic), new StringDeserializer(), new StringDeserializer(), downstreamActor))
+    val consumer = system.actorOf(KafkaConsumerActor.props(configuredActor, new StringDeserializer(), new StringDeserializer(), downstreamActor))
     consumer ! Subscribe.AutoPartition(List(topic))
 
     // Initiate DeathWatch

client/src/test/scala/cakesolutions/kafka/ConfigureSerializationSpec.scala

@@ -46,7 +46,8 @@ class ConfigureSerializationSpec extends KafkaIntSpec{
          """.stripMargin
       ), keySerializer, valueSerializer)
 
-    val _ = KafkaProducer(conf)
+    val producer = KafkaProducer(conf)
+    producer.close
 
     keySerializer.configuration shouldEqual "mock_value"
     keySerializer.isKeySerializer shouldEqual true
@@ -66,7 +67,8 @@ class ConfigureSerializationSpec extends KafkaIntSpec{
          """.stripMargin
       ), keyDeserializer, valueDeserializer)
 
-    val _ = KafkaConsumer(conf)
+    val consumer = KafkaConsumer(conf)
+    consumer.close
 
     keyDeserializer.configuration shouldEqual "mock_value"
     keyDeserializer.isKeyDeserializer shouldEqual true