Kafka source poll timeout instead of backoff (#164)

* 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

* Better kafka source worker polling

Author: joeyjackson

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

@@ -181,36 +181,33 @@ private class ParsingWorker implements Runnable {
         @Override
         public void run() {
             while (_isRunning || !_buffer.isEmpty()) { // Empty the queue before stopping the workers
-                final V value = _buffer.poll();
-                _periodicMetrics.recordGauge(_queueSizeGaugeMetricName, _buffer.size());
-                if (value != null) {
-                    final List<Record> records;
-                    try {
-                        final Stopwatch parsingTimer = Stopwatch.createStarted();
-                        records = _parser.parse(value);
-                        parsingTimer.stop();
-                        _periodicMetrics.recordTimer(_parsingTimeMetricName,
-                                parsingTimer.elapsed(TimeUnit.NANOSECONDS), Optional.of(Units.NANOSECOND));
-                    } catch (final ParsingException e) {
-                        _periodicMetrics.recordCounter(_parsingExceptionCountMetricName, 1);
-                        _logger.error()
-                                .setMessage("Failed to parse data")
-                                .setThrowable(e)
-                                .log();
-                        continue;
-                    }
-                    for (final Record record : records) {
-                        KafkaSource.this.notify(record);
-                        _currentRecordsProcessedCount.getAndIncrement();
-                    }
-                } else {
-                    // Queue is empty
-                    try {
-                        Thread.sleep(_backoffTime.toMillis());
-                    } catch (final InterruptedException e) {
-                        Thread.currentThread().interrupt();
-                        stop();
+                try {
+                    final V value = _buffer.poll(_backoffTime.toMillis(), TimeUnit.MILLISECONDS);
+                    _periodicMetrics.recordGauge(_queueSizeGaugeMetricName, _buffer.size());
+                    if (value != null) {
+                        final List<Record> records;
+                        try {
+                            final Stopwatch parsingTimer = Stopwatch.createStarted();
+                            records = _parser.parse(value);
+                            parsingTimer.stop();
+                            _periodicMetrics.recordTimer(_parsingTimeMetricName,
+                                    parsingTimer.elapsed(TimeUnit.NANOSECONDS), Optional.of(Units.NANOSECOND));
+                        } catch (final ParsingException e) {
+                            _periodicMetrics.recordCounter(_parsingExceptionCountMetricName, 1);
+                            _logger.error()
+                                    .setMessage("Failed to parse data")
+                                    .setThrowable(e)
+                                    .log();
+                            continue;
+                        }
+                        for (final Record record : records) {
+                            KafkaSource.this.notify(record);
+                            _currentRecordsProcessedCount.getAndIncrement();
+                        }
                     }
+                } catch (final InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                    stop();
                 }
             }
         }