Kafka record list (#162)

* kafka source init

* kafka consumer wrapper and listener

* unit test and checkstyle

* exception handling

* unit tests for kafka source

* interfaces and wrappers for kafka source

* removed kafka dep

* integration tests for kafka source

* config deserialization

* Fixed deserializer

* Fixed deserializer

* styling

* kafka docker

* pipeline config example

* style

* error checking

* error checking

* integration test kafka source from config

* style

* added parser to kafka source

* example pipeline

* Fail integration test on send fail to kafka server

* requested changes

* requested changes

* configurable backoff time for kafka source

* fixed conf deserializer

* concurrent parsing workers

* multi worker unit test

* queue holds record values instead of records

* style

* instrument init

* todo

* mock observer for multithreaded testing

* configurable buffer queue size

* moved fill queue integration test to unit test

* style

* ensure queue fills in queue filled test

* refactor kafka source constructors

* style

* fix injector in integration tests

* instrumentation testing init

* unit tests for instrumentation counter

* unit test gauge metric

* more instrumentation metrics

* remove prinln

* new metric names

* metrics unit tests

* requested changes

* nonnull annotate

* Make kafka source parser output list of Records

Author: joeyjackson

src/main/java/com/arpnetworking/metrics/common/sources/KafkaSource.java

@@ -23,6 +23,7 @@
 import com.arpnetworking.metrics.common.parsers.Parser;
 import com.arpnetworking.metrics.common.parsers.exceptions.ParsingException;
 import com.arpnetworking.metrics.incubator.PeriodicMetrics;
+import com.arpnetworking.metrics.mad.model.Record;
 import com.arpnetworking.steno.LogValueMapFactory;
 import com.arpnetworking.steno.Logger;
 import com.arpnetworking.steno.LoggerFactory;
@@ -37,6 +38,7 @@
 import org.apache.kafka.common.KafkaException;
 
 import java.time.Duration;
+import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
@@ -49,19 +51,18 @@
  * Produce instances of {@link com.arpnetworking.metrics.mad.model.Record} from the values of entries
  * from a Kafka topic. The key from the entries gets discarded
  *
- * @param <T> the type of data created by the source
  * @param <V> the type of data of value in kafka {@code ConsumerRecords}
  *
  * @author Joey Jackson (jjackson at dropbox dot com)
  */
-public final class KafkaSource<T, V> extends BaseSource {
+public final class KafkaSource<V> extends BaseSource {
     private static final Logger LOGGER = LoggerFactory.getLogger(KafkaSource.class);
 
     private final Consumer<?, V> _consumer;
     private final RunnableConsumer _runnableConsumer;
     private final ExecutorService _consumerExecutor;
     private final ExecutorService _parserExecutor;
-    private final Parser<T, V> _parser;
+    private final Parser<List<Record>, V> _parser;
     private final Logger _logger;
     private final Duration _shutdownAwaitTime;
     private final Duration _backoffTime;
@@ -135,21 +136,21 @@ public String toString() {
     }
 
     @SuppressWarnings("unused")
-    private KafkaSource(final Builder<T, V> builder) {
+    private KafkaSource(final Builder<V> builder) {
         this(builder, LOGGER, new ArrayBlockingQueue<>(builder._bufferSize));
     }
 
     // NOTE: Package private for testing
-    /* package private */ KafkaSource(final Builder<T, V> builder, final Logger logger) {
+    /* package private */ KafkaSource(final Builder<V> builder, final Logger logger) {
         this(builder, logger, new ArrayBlockingQueue<>(builder._bufferSize));
     }
 
     // NOTE: Package private for testing
-    /* package private */ KafkaSource(final Builder<T, V> builder, final BlockingQueue<V> buffer) {
+    /* package private */ KafkaSource(final Builder<V> builder, final BlockingQueue<V> buffer) {
         this(builder, LOGGER, buffer);
     }
 
-    private KafkaSource(final Builder<T, V> builder, final Logger logger, final BlockingQueue<V> buffer) {
+    private KafkaSource(final Builder<V> builder, final Logger logger, final BlockingQueue<V> buffer) {
         super(builder);
         _consumer = builder._consumer;
         _parser = builder._parser;
@@ -183,10 +184,10 @@ public void run() {
                 final V value = _buffer.poll();
                 _periodicMetrics.recordGauge(_queueSizeGaugeMetricName, _buffer.size());
                 if (value != null) {
-                    final T record;
+                    final List<Record> records;
                     try {
                         final Stopwatch parsingTimer = Stopwatch.createStarted();
-                        record = _parser.parse(value);
+                        records = _parser.parse(value);
                         parsingTimer.stop();
                         _periodicMetrics.recordTimer(_parsingTimeMetricName,
                                 parsingTimer.elapsed(TimeUnit.NANOSECONDS), Optional.of(Units.NANOSECOND));
@@ -198,8 +199,10 @@ record = _parser.parse(value);
                                 .log();
                         continue;
                     }
-                    KafkaSource.this.notify(record);
-                    _currentRecordsProcessedCount.getAndIncrement();
+                    for (final Record record : records) {
+                        KafkaSource.this.notify(record);
+                        _currentRecordsProcessedCount.getAndIncrement();
+                    }
                 } else {
                     // Queue is empty
                     try {
@@ -289,12 +292,11 @@ private void backoff(final Throwable throwable) {
     /**
      * Builder pattern class for {@link KafkaSource}.
      *
-     * @param <T> the type of data created by the source
      * @param <V> the type of data of value in kafka {@code ConsumerRecords}
      *
      * @author Joey Jackson (jjackson at dropbox dot com)
      */
-    public static final class Builder<T, V> extends BaseSource.Builder<Builder<T, V>, KafkaSource<T, V>> {
+    public static final class Builder<V> extends BaseSource.Builder<Builder<V>, KafkaSource<V>> {
 
         /**
          * Public constructor.
@@ -309,7 +311,7 @@ public Builder() {
          * @param consumer The {@code Consumer}.
          * @return This instance of {@link KafkaSource.Builder}.
          */
-        public Builder<T, V> setConsumer(final Consumer<?, V> consumer) {
+        public Builder<V> setConsumer(final Consumer<?, V> consumer) {
             _consumer = consumer;
             return this;
         }
@@ -320,7 +322,7 @@ public Builder<T, V> setConsumer(final Consumer<?, V> consumer) {
          * @param value The {@link Parser}.
          * @return This instance of {@link KafkaSource.Builder}.
          */
-        public Builder<T, V> setParser(final Parser<T, V> value) {
+        public Builder<V> setParser(final Parser<List<Record>, V> value) {
             _parser = value;
             return this;
         }
@@ -331,7 +333,7 @@ public Builder<T, V> setParser(final Parser<T, V> value) {
          * @param pollTime The {@code Duration} of each poll call made by the {@code KafkaConsumer}.
          * @return This instance of {@link KafkaSource.Builder}.
          */
-        public Builder<T, V> setPollTime(final Duration pollTime) {
+        public Builder<V> setPollTime(final Duration pollTime) {
             _pollTime = pollTime;
             return this;
         }
@@ -344,7 +346,7 @@ public Builder<T, V> setPollTime(final Duration pollTime) {
          *                          the {@link RunnableConsumer} thread.
          * @return This instance of {@link KafkaSource.Builder}.
          */
-        public Builder<T, V> setShutdownAwaitTime(final Duration shutdownAwaitTime) {
+        public Builder<V> setShutdownAwaitTime(final Duration shutdownAwaitTime) {
             _shutdownAwaitTime = shutdownAwaitTime;
             return this;
         }
@@ -357,7 +359,7 @@ public Builder<T, V> setShutdownAwaitTime(final Duration shutdownAwaitTime) {
          *                          an operation on exception.
          * @return This instance of {@link KafkaSource.Builder}.
          */
-        public Builder<T, V> setBackoffTime(final Duration backoffTime) {
+        public Builder<V> setBackoffTime(final Duration backoffTime) {
             _backoffTime = backoffTime;
             return this;
         }
@@ -369,7 +371,7 @@ public Builder<T, V> setBackoffTime(final Duration backoffTime) {
          * @param numWorkerThreads The number of parsing worker threads.
          * @return This instance of {@link KafkaSource.Builder}.
          */
-        public Builder<T, V> setNumWorkerThreads(final Integer numWorkerThreads) {
+        public Builder<V> setNumWorkerThreads(final Integer numWorkerThreads) {
             _numWorkerThreads = numWorkerThreads;
             return this;
         }
@@ -381,7 +383,7 @@ public Builder<T, V> setNumWorkerThreads(final Integer numWorkerThreads) {
          * @param bufferSize The size of the buffer.
          * @return This instance of {@link KafkaSource.Builder}.
          */
-        public Builder<T, V> setBufferSize(final Integer bufferSize) {
+        public Builder<V> setBufferSize(final Integer bufferSize) {
             _bufferSize = bufferSize;
             return this;
         }
@@ -392,20 +394,20 @@ public Builder<T, V> setBufferSize(final Integer bufferSize) {
          * @param periodicMetrics The {@code PeriodicMetrics} for the {@link KafkaSource}.
          * @return This instance of {@link KafkaSource.Builder}.
          */
-        public Builder<T, V> setPeriodicMetrics(final PeriodicMetrics periodicMetrics) {
+        public Builder<V> setPeriodicMetrics(final PeriodicMetrics periodicMetrics) {
             _periodicMetrics = periodicMetrics;
             return this;
         }
 
         @Override
-        protected Builder<T, V> self() {
+        protected Builder<V> self() {
             return this;
         }
 
         @NotNull
         private Consumer<?, V> _consumer;
         @NotNull
-        private Parser<T, V> _parser;
+        private Parser<List<Record>, V> _parser;
         @NotNull
         @CheckWith(value = PositiveDuration.class, message = "Poll duration must be positive.")
         private Duration _pollTime;

src/test/java/com/arpnetworking/metrics/common/integration/KafkaIT.java

@@ -20,7 +20,7 @@
 import com.arpnetworking.metrics.common.kafka.ConsumerDeserializer;
 import com.arpnetworking.metrics.common.sources.KafkaSource;
 import com.arpnetworking.metrics.incubator.PeriodicMetrics;
-import com.arpnetworking.test.StringParser;
+import com.arpnetworking.test.StringToRecordParser;
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.introspect.AnnotationIntrospectorPair;
@@ -84,7 +84,7 @@ public class KafkaIT {
     private Map<String, Object> _consumerProps;
     private KafkaConsumer<Integer, String> _consumer;
     private String _topicName;
-    private KafkaSource<String, String> _source;
+    private KafkaSource<String> _source;
     private List<ProducerRecord<Integer, String>> _producerRecords;
     private PeriodicMetrics _periodicMetrics;
 
@@ -109,9 +109,9 @@ public void testKafkaSourceSingleObserver() {
         // Create Kafka Source
         _consumer = new KafkaConsumer<>(_consumerProps);
         _consumer.subscribe(Collections.singletonList(_topicName));
-        _source = new KafkaSource.Builder<String, String>()
+        _source = new KafkaSource.Builder<String>()
                 .setName("KafkaSource")
-                .setParser(new StringParser())
+                .setParser(new StringToRecordParser())
                 .setConsumer(_consumer)
                 .setPollTime(POLL_DURATION)
                 .setPeriodicMetrics(_periodicMetrics)
@@ -122,7 +122,8 @@ public void testKafkaSourceSingleObserver() {
         _source.attach(observer);
         _source.start();
         for (ProducerRecord<Integer, String> expected : _producerRecords) {
-            Mockito.verify(observer, Mockito.timeout(TIMEOUT)).notify(_source, expected.value());
+            Mockito.verify(observer, Mockito.timeout(TIMEOUT)).notify(_source,
+                    StringToRecordParser.recordWithID(expected.value()));
         }
     }
 
@@ -131,9 +132,9 @@ public void testKafkaSourceMultipleObservers() {
         // Create Kafka Source
         _consumer = new KafkaConsumer<>(_consumerProps);
         _consumer.subscribe(Collections.singletonList(_topicName));
-        _source = new KafkaSource.Builder<String, String>()
+        _source = new KafkaSource.Builder<String>()
                 .setName("KafkaSource")
-                .setParser(new StringParser())
+                .setParser(new StringToRecordParser())
                 .setConsumer(_consumer)
                 .setPollTime(POLL_DURATION)
                 .setPeriodicMetrics(_periodicMetrics)
@@ -146,8 +147,10 @@ public void testKafkaSourceMultipleObservers() {
         _source.attach(observer2);
         _source.start();
         for (ProducerRecord<Integer, String> expected : _producerRecords) {
-            Mockito.verify(observer1, Mockito.timeout(TIMEOUT)).notify(_source, expected.value());
-            Mockito.verify(observer2, Mockito.timeout(TIMEOUT)).notify(_source, expected.value());
+            Mockito.verify(observer1, Mockito.timeout(TIMEOUT)).notify(_source,
+                    StringToRecordParser.recordWithID(expected.value()));
+            Mockito.verify(observer2, Mockito.timeout(TIMEOUT)).notify(_source,
+                    StringToRecordParser.recordWithID(expected.value()));
         }
     }
 
@@ -171,7 +174,7 @@ public void testKafkaSourceFromConfig() throws IOException {
                 + "\n    }"
                 + "\n  },"
                 + "\n  \"parser\":{"
-                + "\n    \"type\":\"com.arpnetworking.test.StringParser\""
+                + "\n    \"type\":\"com.arpnetworking.test.StringToRecordParser\""
                 + "\n  },"
                 + "\n  \"backoffTime\":\"PT1S\""
                 + "\n}";
@@ -204,7 +207,8 @@ protected void configure() {
         _source.attach(observer);
         _source.start();
         for (ProducerRecord<Integer, String> expected : _producerRecords) {
-            Mockito.verify(observer, Mockito.timeout(TIMEOUT)).notify(_source, expected.value());
+            Mockito.verify(observer, Mockito.timeout(TIMEOUT)).notify(_source,
+                    StringToRecordParser.recordWithID(expected.value()));
         }
     }
 
@@ -213,9 +217,9 @@ public void testKafkaSourceMultipleWorkerThreads() {
         // Create Kafka Source
         _consumer = new KafkaConsumer<>(_consumerProps);
         _consumer.subscribe(Collections.singletonList(_topicName));
-        _source = new KafkaSource.Builder<String, String>()
+        _source = new KafkaSource.Builder<String>()
                 .setName("KafkaSource")
-                .setParser(new StringParser())
+                .setParser(new StringToRecordParser())
                 .setConsumer(_consumer)
                 .setPollTime(POLL_DURATION)
                 .setNumWorkerThreads(4)
@@ -231,7 +235,7 @@ public void testKafkaSourceMultipleWorkerThreads() {
         Mockito.verify(observer, Mockito.timeout(TIMEOUT).times(NUM_RECORDS)).notify(Mockito.any(), captor.capture());
         Assert.assertEquals(
                 _producerRecords.stream().map(ProducerRecord::value).sorted().collect(Collectors.toList()),
-                captor.getAllValues().stream().sorted().collect(Collectors.toList())
+                captor.getAllValues().stream().map(StringToRecordParser::recordID).sorted().collect(Collectors.toList())
         );
     }
 
@@ -318,5 +322,5 @@ private void setupKafka() {
      *
      * @author Joey Jackson (jjackson at dropbox dot com)
      */
-    private static class KafkaSourceStringType extends TypeReference<KafkaSource<String, String>> {}
+    private static class KafkaSourceStringType extends TypeReference<KafkaSource<String>> {}
 }