add logo

README.md

@@ -12,13 +12,13 @@ Lenses offers SQL (for data browsing and Kafka Streams), Kafka Connect connector
 You can find more on [landoop.com!](http://www.landoop.com/kafka-lenses/)
 
 <p align="center">
-  <a href="http://www.landoop.com/kafka-lenses/" target="_blank"><img src="http://www.landoop.com/docs/lenses/_images/lenses-data-management-platform.png"/></a>
+  <a href="http://www.landoop.com/kafka-lenses/" target="_blank"><img src="https://www.landoop.com/images/new/bg-12.gif"/></a>
 </p>
 
 # Stream Reactor
-Streaming reference architecture built around Kafka. 
 
-![Alt text](https://datamountaineer.com/wp-content/uploads/2016/01/stream-reactor-1.jpg)
+
+![Alt text](images/streamreactor-logo.png)
 
 A collection of components to build a real time ingestion pipeline.
 

kafka-connect-pulsar/src/main/scala/com/datamountaineer/streamreactor/connect/pulsar/sink/PulsarWriter.scala

@@ -39,17 +39,14 @@ class PulsarWriter(client: PulsarClient, settings: PulsarSinkSettings) extends S
   // Enable compression
   //TODO: pick it up from settings
   conf.setCompressionType(CompressionType.LZ4)
-  private var producersMap = scala.collection.mutable.Map.empty[String, Producer]
+  private val producersMap = scala.collection.mutable.Map.empty[String, Producer]
 
   //initialize error tracker
   initialize(settings.maxRetries, settings.errorPolicy)
   private val mappings: Map[String, Set[Kcql]] = settings.kcql.groupBy(k => k.getSource)
-  private val kcql = settings.kcql
 
   def write(records: Iterable[SinkRecord]) = {
 
-    val grouped = records.groupBy(r => r.topic())
-
     val t = Try {
       records.foreach { record =>
         val topic = record.topic()
@@ -86,6 +83,7 @@ class PulsarWriter(client: PulsarClient, settings: PulsarSinkSettings) extends S
   def flush = {}
 
   def close = {
+    logger.info("Closing client")
     client.close()
   }
 }