- kafka 2.6

- more logging

build.gradle

@@ -38,8 +38,8 @@ allprojects {
     compile 'net.savantly:graphite-client:1.1.0-RELEASE'
     compile 'com.timgroup:java-statsd-client:3.0.1'
     compile 'com.signalfx.public:signalfx-codahale:0.0.47'
-    compile group: 'org.apache.kafka', name: 'kafka_2.12', version: '2.4.0'
-    compile group: 'org.apache.kafka', name: 'kafka-clients', version: '2.3.1'
+    compile group: 'org.apache.kafka', name: 'kafka_2.12', version: '2.6.1'
+    compile group: 'org.apache.kafka', name: 'kafka-clients', version: '2.6.1'
     testCompile 'org.mockito:mockito-core:2.24.0'
     testCompile 'org.testng:testng:6.8.8'
   }

config/log4j2.properties

@@ -28,20 +28,20 @@ appender.kafka.policies.type = Policies
 appender.kafka.policies.time.type = TimeBasedTriggeringPolicy
 
 # Modify this as needed when working on dev box. Trace -> Debug -> Info -> Warn -> Error -> Fatal
-rootLogger.level = info
+rootLogger.level = debug
 rootLogger.appenderRef.console.ref = STDOUT
 
-logger.kmf.name = com.linkedin.kmf.core.KafkaMonitor
-logger.kmf.level = info
+logger.kmf.name = com.linkedin.xinfra.monitor
+logger.kmf.level = debug
 logger.kmf.additivity = false
 logger.kmf.appenderRef.console.ref = STDOUT
 
 logger.kafkaClient.name = org.apache.kafka
 logger.kafkaClient.level = warn
 logger.kafkaClient.additivity = false
-logger.kafkaClient.appenderRef.kafka.ref = KAFKA
+logger.kafkaClient.appenderRef.kafka.ref = STDOUT
 
 logger.kafka.name = kafka
 logger.kafka.level = warn
 logger.kafka.additivity = false
-logger.kafka.appenderRef.kafka.ref = KAFKA
+logger.kafka.appenderRef.kafka.ref = STDOUT

src/main/java/com/linkedin/xinfra/monitor/services/OffsetCommitService.java

@@ -107,14 +107,14 @@ public class OffsetCommitService implements Service {
     List<String> bootstrapServers = config.getList(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG);
     List<InetSocketAddress> addresses =
         ClientUtils.parseAndValidateAddresses(bootstrapServers, ClientDnsLookup.DEFAULT);
-    ChannelBuilder channelBuilder = ClientUtils.createChannelBuilder(config, _time);
+    ChannelBuilder channelBuilder = ClientUtils.createChannelBuilder(config, _time, logContext);
 
     LOGGER.info("Bootstrap servers config: {} | broker addresses: {}", bootstrapServers, addresses);
 
     Metadata metadata = new Metadata(retryBackoffMs, config.getLong(ConsumerConfig.METADATA_MAX_AGE_CONFIG), logContext,
         new ClusterResourceListeners());
 
-    metadata.bootstrap(addresses, _time.milliseconds());
+    metadata.bootstrap(addresses);
 
     Selector selector =
         new Selector(config.getLong(ConsumerConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG), new Metrics(), _time,