Update streaming consumer to commit state (#179)

* Commit Consumer State

* Code cleanup

* Update changelog

Author: Crim

CHANGELOG.md

@@ -6,6 +6,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 #### Bug Fixes
 - [ISSUE-175](https://github.com/SourceLabOrg/kafka-webview/issues/175) Update multi-threaded consumers with unique consumerId [PR](https://github.com/SourceLabOrg/kafka-webview/pull/176).
 - [PR-178](https://github.com/SourceLabOrg/kafka-webview/pull/178) @[lucrito](https://github.com/lucrito) fixed shebang in start.sh script.  Thanks!
+- [PR-179](https://github.com/SourceLabOrg/kafka-webview/pull/179) Streaming consumer now persists consumer state.
 
 ## 2.3.0 (06/19/2019)
 #### New Features

kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/manager/kafka/SocketKafkaConsumer.java

@@ -61,7 +61,7 @@ public class SocketKafkaConsumer implements Runnable {
     // get a large flood of messages, which maybe is a good thing? Or bad? Unsure.
     private static final int maxQueueCapacity = 25;
 
-    private final KafkaConsumer kafkaConsumer;
+    private final KafkaConsumer<?,?> kafkaConsumer;
     private final ClientConfig clientConfig;
     private final Duration pollTimeoutDuration = Duration.ofMillis(POLL_TIMEOUT_MS);
     private final BlockingQueue<KafkaResult> outputQueue;
@@ -74,7 +74,7 @@ public class SocketKafkaConsumer implements Runnable {
      * @param clientConfig The client's configuration.
      */
     public SocketKafkaConsumer(
-        final KafkaConsumer kafkaConsumer,
+        final KafkaConsumer<?,?> kafkaConsumer,
         final ClientConfig clientConfig) {
 
         this.kafkaConsumer = kafkaConsumer;
@@ -109,7 +109,7 @@ public void run() {
 
         do {
             // Start trying to consume messages from kafka
-            final ConsumerRecords consumerRecords = kafkaConsumer.poll(pollTimeoutDuration);
+            final ConsumerRecords<?,?> consumerRecords = kafkaConsumer.poll(pollTimeoutDuration);
 
             // If no records found
             if (consumerRecords.isEmpty()) {
@@ -121,7 +121,7 @@ public void run() {
             }
 
             // Push messages onto output queue
-            for (final ConsumerRecord consumerRecord : (Iterable<ConsumerRecord>) consumerRecords) {
+            for (final ConsumerRecord consumerRecord : consumerRecords) {
                 // Translate record
                 final KafkaResult kafkaResult = new KafkaResult(
                     consumerRecord.partition(),
@@ -141,6 +141,9 @@ public void run() {
                 }
             }
 
+            // Commit state async.
+            kafkaConsumer.commitAsync();
+
             // Sleep for a bit
             sleep(DWELL_TIME_MS);
         }
@@ -231,7 +234,6 @@ private void seekToTimestamp(final long timestamp) {
     /**
      * Seek to the specified offsets.
      * @param partitionOffsetMap Map of PartitionId to Offset to seek to.
-     * @return ConsumerState representing the consumer's positions.
      */
     private void seek(final Map<TopicPartition, Long> partitionOffsetMap) {
         for (final Map.Entry<TopicPartition, Long> entry: partitionOffsetMap.entrySet()) {