Reverted back the tests and made changes according to the comments

Author: shiv4nsh

project/build.properties

@@ -0,0 +1 @@
+sbt.version=1.0.3

src/main/scala/com/lightbend/kafka/scala/streams/StreamsBuilderS.scala

@@ -1,81 +1,74 @@
 /**
- * Copyright (C) 2018 Lightbend Inc. <https://www.lightbend.com>
- */
+  * Copyright (C) 2018 Lightbend Inc. <https://www.lightbend.com>
+  */
 
 package com.lightbend.kafka.scala.streams
 
 import java.util.regex.Pattern
 
-import ImplicitConversions._
-import org.apache.kafka.streams.kstream.{ GlobalKTable, Materialized }
+import com.lightbend.kafka.scala.streams.ImplicitConversions._
+import org.apache.kafka.common.utils.Bytes
+import org.apache.kafka.streams.kstream.{GlobalKTable, Materialized}
 import org.apache.kafka.streams.processor.{ProcessorSupplier, StateStore}
-import org.apache.kafka.streams.state.{ StoreBuilder, KeyValueStore }
+import org.apache.kafka.streams.state.{KeyValueStore, StoreBuilder}
 import org.apache.kafka.streams.{Consumed, StreamsBuilder, Topology}
-import org.apache.kafka.common.utils.Bytes
 
 import scala.collection.JavaConverters._
 
 /**
- * Wraps the Java class StreamsBuilder and delegates method calls to the underlying Java object.
- */
-class StreamsBuilderS(inner:StreamsBuilder) {
+  * Wraps the Java class StreamsBuilder and delegates method calls to the underlying Java object.
+  */
+class StreamsBuilderS(inner: StreamsBuilder = new StreamsBuilder) {
 
-  def stream[K, V](topic: String) : KStreamS[K, V] =
-     inner.stream[K, V](topic)
+  def stream[K, V](topic: String): KStreamS[K, V] =
+    inner.stream[K, V](topic)
 
-  def stream[K, V](topic: String, consumed: Consumed[K, V]) : KStreamS[K, V] =
-     inner.stream[K, V](topic, consumed)
+  def stream[K, V](topic: String, consumed: Consumed[K, V]): KStreamS[K, V] =
+    inner.stream[K, V](topic, consumed)
 
   def stream[K, V](topics: List[String]): KStreamS[K, V] =
-     inner.stream[K, V](topics.asJava)
+    inner.stream[K, V](topics.asJava)
 
   def stream[K, V](topics: List[String], consumed: Consumed[K, V]): KStreamS[K, V] =
-     inner.stream[K, V](topics.asJava, consumed)
+    inner.stream[K, V](topics.asJava, consumed)
 
-  def stream[K, V](topicPattern: Pattern) : KStreamS[K, V] =
+  def stream[K, V](topicPattern: Pattern): KStreamS[K, V] =
     inner.stream[K, V](topicPattern)
 
-  def stream[K, V](topicPattern: Pattern, consumed: Consumed[K, V]) : KStreamS[K, V] =
+  def stream[K, V](topicPattern: Pattern, consumed: Consumed[K, V]): KStreamS[K, V] =
     inner.stream[K, V](topicPattern, consumed)
 
-  def table[K, V](topic: String) : KTableS[K, V] = inner.table[K, V](topic)
+  def table[K, V](topic: String): KTableS[K, V] = inner.table[K, V](topic)
 
-  def table[K, V](topic: String, consumed: Consumed[K, V]) : KTableS[K, V] =
+  def table[K, V](topic: String, consumed: Consumed[K, V]): KTableS[K, V] =
     inner.table[K, V](topic, consumed)
 
   def table[K, V](topic: String, consumed: Consumed[K, V],
-    materialized: Materialized[K, V, KeyValueStore[Bytes, Array[Byte]]]): KTableS[K, V] =
-    inner.table[K, V](topic,  consumed, materialized)
+                  materialized: Materialized[K, V, KeyValueStore[Bytes, Array[Byte]]]): KTableS[K, V] =
+    inner.table[K, V](topic, consumed, materialized)
 
   def table[K, V](topic: String,
-    materialized: Materialized[K, V, KeyValueStore[Bytes, Array[Byte]]]): KTableS[K, V] =
+                  materialized: Materialized[K, V, KeyValueStore[Bytes, Array[Byte]]]): KTableS[K, V] =
     inner.table[K, V](topic, materialized)
 
   def globalTable[K, V](topic: String): GlobalKTable[K, V] =
     inner.globalTable(topic)
 
-  def globalTable[K, V](topic: String, consumed: Consumed[K, V]) : GlobalKTable[K, V] =
-     inner.globalTable(topic, consumed)
+  def globalTable[K, V](topic: String, consumed: Consumed[K, V]): GlobalKTable[K, V] =
+    inner.globalTable(topic, consumed)
 
   def globalTable[K, V](topic: String, consumed: Consumed[K, V],
-    materialized: Materialized[K, V, KeyValueStore[Bytes, Array[Byte]]]): GlobalKTable[K, V] =
-    inner.globalTable(topic,  consumed, materialized)
+                        materialized: Materialized[K, V, KeyValueStore[Bytes, Array[Byte]]]): GlobalKTable[K, V] =
+    inner.globalTable(topic, consumed, materialized)
 
   def globalTable[K, V](topic: String,
-    materialized: Materialized[K, V, KeyValueStore[Bytes, Array[Byte]]]): GlobalKTable[K, V] =
-    inner.globalTable(topic,  materialized)
+                        materialized: Materialized[K, V, KeyValueStore[Bytes, Array[Byte]]]): GlobalKTable[K, V] =
+    inner.globalTable(topic, materialized)
 
   def addStateStore(builder: StoreBuilder[_ <: StateStore]): StreamsBuilder = inner.addStateStore(builder)
 
   def addGlobalStore(storeBuilder: StoreBuilder[_ <: StateStore], topic: String, sourceName: String, consumed: Consumed[_, _], processorName: String, stateUpdateSupplier: ProcessorSupplier[_, _]): StreamsBuilder =
-    inner.addGlobalStore(storeBuilder,topic,sourceName,consumed,processorName,stateUpdateSupplier)
-
-  def build() : Topology = inner.build()
-}
-
-object StreamsBuilderS {
-
-  def apply() = new StreamsBuilderS(new StreamsBuilder)
+    inner.addGlobalStore(storeBuilder, topic, sourceName, consumed, processorName, stateUpdateSupplier)
 
-  def apply(streamsBuilder: StreamsBuilder)= new StreamsBuilderS(streamsBuilder)
+  def build(): Topology = inner.build()
 }

src/test/scala/com/lightbend/kafka/scala/streams/KafkaStreamsTest.scala

@@ -46,7 +46,7 @@ object KafkaStreamsTest extends TestSuite[KafkaLocalServer] with WordCountTestDa
     streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName())
     streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG, localStateDir)
 
-    val builder = StreamsBuilderS()
+    val builder = new StreamsBuilderS()
 
     val textLines = builder.stream[String, String](inputTopic)
 

src/test/scala/com/lightbend/kafka/scala/streams/ProbabilisticCountingScalaIntegrationTest.scala

@@ -97,7 +97,7 @@ object ProbabilisticCountingScalaIntegrationTest extends TestSuite[KafkaLocalSer
       p
     }
 
-    val builder = StreamsBuilderS()
+    val builder = new StreamsBuilderS()
 
     val cmsStoreName = "cms-store"
     val cmsStoreBuilder = {

src/test/scala/com/lightbend/kafka/scala/streams/StreamToTableJoinScalaIntegrationTest.scala

@@ -81,7 +81,7 @@ object StreamToTableJoinScalaIntegrationTest extends TestSuite[KafkaLocalServer]
       p
     }
 
-    val builder = StreamsBuilderS()
+    val builder = new StreamsBuilderS()
 
     val userClicksStream: KStreamS[String, Long] = builder.stream(userClicksTopic, Consumed.`with`(stringSerde, longSerde))