KREST-182: (Correctly) fences access to internal consumer buffer.

Concurrent cosume requests on the same consumer could race in between calls to `KafkaConsumerState.hasNext()` and `KafkaConsumerState.next()`. I use double-checked locking in `KafkaConsumerReadTask` to allow multiple consume requests on the same consumer to make progress in parallel (instead of one request completing at a time, per consumer).

Author: rigelbm

kafka-rest/src/main/java/io/confluent/kafkarest/v2/KafkaConsumerReadTask.java

@@ -39,7 +39,7 @@ class KafkaConsumerReadTask<KafkaKeyT, KafkaValueT, ClientKeyT, ClientValueT> {
 
   private static final Logger log = LoggerFactory.getLogger(KafkaConsumerReadTask.class);
 
-  private KafkaConsumerState<KafkaKeyT, KafkaValueT, ClientKeyT, ClientValueT> parent;
+  private final KafkaConsumerState<KafkaKeyT, KafkaValueT, ClientKeyT, ClientValueT> parent;
   private final long requestTimeoutMs;
   // the minimum bytes the task should accumulate
   // before returning a response (or hitting the timeout)
@@ -138,11 +138,18 @@ public boolean isDone() {
    */
   private void addRecords() {
     while (!exceededMinResponseBytes && !exceededMaxResponseBytes && parent.hasNext()) {
-      maybeAddRecord();
+      synchronized (parent) {
+        if (parent.hasNext()) {
+          maybeAddRecord();
+        }
+      }
     }
     while (!exceededMaxResponseBytes && parent.hasNextCached()) {
-      // will not call poll() anymore. Continue draining loaded records
-      maybeAddRecord();
+      synchronized (parent) {
+        if (parent.hasNextCached()) {
+          maybeAddRecord();
+        }
+      }
     }
   }