diff --git a/pom.xml b/pom.xml
index 1e73b39..973b4aa 100644
--- a/pom.xml
+++ b/pom.xml
@@ -2,14 +2,14 @@
 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
 
-   <groupId>com.homeadvisor.kafka</groupId>
+   <groupId>com.neilbuesing.kafka</groupId>
    <artifactId>kafdrop</artifactId>
    <version>2.0.6</version>
 
    <description>For when you have a Kaf(ka) cluster to monitor</description>
 
    <properties>
-      <spring.boot.version>1.3.6.RELEASE</spring.boot.version>
+      <spring.boot.version>1.5.10.RELEASE</spring.boot.version>
       <additionalparam>-Xdoclint:none</additionalparam>
       <curator.version>2.10.0</curator.version>
 
@@ -19,8 +19,8 @@
    </properties>
 
    <scm>
-       <connection>scm:git:git@github.com:HomeAdvisor/Kafdrop.git</connection>
-       <developerConnection>scm:git:git@github.com:HomeAdvisor/Kafdrop.git</developerConnection>
+       <connection>scm:git:git@github.com:nbuesing/Kafdrop.git</connection>
+       <developerConnection>scm:git:git@github.com:nbuesing/Kafdrop.git</developerConnection>
        <tag>HEAD</tag>
    </scm>
 
@@ -67,13 +67,29 @@
       <dependency>
          <groupId>org.apache.zookeeper</groupId>
          <artifactId>zookeeper</artifactId>
-         <version>3.4.8</version>
+         <version>3.4.13</version>
       </dependency>
+      <!--<dependency>-->
+         <!--<groupId>org.apache.kafka</groupId>-->
+         <!--<artifactId>kafka_2.9.2</artifactId>-->
+         <!--<version>0.8.2.2</version>-->
+      <!--</dependency>-->
       <dependency>
          <groupId>org.apache.kafka</groupId>
-         <artifactId>kafka_2.9.2</artifactId>
-         <version>0.8.2.2</version>
+         <artifactId>kafka_2.11</artifactId>
+         <version>0.11.0.3</version>
       </dependency>
+      <dependency>
+         <groupId>org.apache.kafka</groupId>
+         <artifactId>kafka-clients</artifactId>
+         <version>0.11.0.3</version>
+      </dependency>
+      <!--<dependency>-->
+         <!--<groupId>org.apache.kafka</groupId>-->
+         <!--<artifactId>kafka_2.12</artifactId>-->
+         <!--<version>0.10.2.2</version>-->
+      <!--</dependency>-->
+
       <dependency>
          <groupId>org.freemarker</groupId>
          <artifactId>freemarker</artifactId>
@@ -82,7 +98,7 @@
       <dependency>
          <groupId>org.springframework.retry</groupId>
          <artifactId>spring-retry</artifactId>
-         <version>1.1.3.RELEASE</version>
+         <version>1.2.2.RELEASE</version>
       </dependency>
       <dependency>
          <groupId>io.confluent</groupId>
@@ -92,12 +108,7 @@
       <dependency>
          <groupId>org.apache.avro</groupId>
          <artifactId>avro</artifactId>
-         <version>1.8.1</version>
-      </dependency>
-      <dependency>
-         <groupId>org.apache.kafka</groupId>
-         <artifactId>kafka-clients</artifactId>
-         <version>0.10.2.2</version>
+         <version>1.8.2</version>
       </dependency>
       <dependency>
          <groupId>com.google.code.findbugs</groupId>
@@ -121,6 +132,10 @@
             </exclusion>
          </exclusions>
       </dependency>
+      <dependency>
+         <groupId>org.springframework.boot</groupId>
+         <artifactId>spring-boot-starter-tomcat</artifactId>
+      </dependency>
       <dependency>
          <groupId>org.springframework.boot</groupId>
          <artifactId>spring-boot-starter-freemarker</artifactId>
diff --git a/src/main/java/com/homeadvisor/kafdrop/service/CuratorKafkaMonitor.java b/src/main/java/com/homeadvisor/kafdrop/service/CuratorKafkaMonitor.java
index 84df01b..9ff5b8f 100644
--- a/src/main/java/com/homeadvisor/kafdrop/service/CuratorKafkaMonitor.java
+++ b/src/main/java/com/homeadvisor/kafdrop/service/CuratorKafkaMonitor.java
@@ -24,13 +24,14 @@
 import com.homeadvisor.kafdrop.model.*;
 import com.homeadvisor.kafdrop.util.BrokerChannel;
 import com.homeadvisor.kafdrop.util.Version;
-import kafka.api.ConsumerMetadataRequest;
+import kafka.api.GroupCoordinatorRequest;
 import kafka.api.PartitionOffsetRequestInfo;
-import kafka.cluster.Broker;
+import kafka.cluster.BrokerEndPoint;
 import kafka.common.ErrorMapping;
 import kafka.common.TopicAndPartition;
 import kafka.javaapi.*;
 import kafka.network.BlockingChannel;
+import kafka.server.ConfigType;
 import kafka.utils.ZKGroupDirs;
 import kafka.utils.ZKGroupTopicDirs;
 import kafka.utils.ZkUtils;
@@ -107,9 +108,9 @@ public void start() throws Exception
       backOffPolicy.setBackOffPeriod(properties.getRetry().getBackoffMillis());
 
       final SimpleRetryPolicy retryPolicy =
-         new SimpleRetryPolicy(properties.getRetry().getMaxAttempts(),
-                               ImmutableMap.of(InterruptedException.class, false,
-                                               Exception.class, true));
+        new SimpleRetryPolicy(properties.getRetry().getMaxAttempts(),
+          ImmutableMap.of(InterruptedException.class, false,
+            Exception.class, true));
 
       retryTemplate = new RetryTemplate();
       retryTemplate.setBackOffPolicy(backOffPolicy);
@@ -128,7 +129,7 @@ public void start() throws Exception
       });
       brokerPathCache.start(StartMode.POST_INITIALIZED_EVENT);
 
-      topicConfigPathCache = new PathChildrenCache(curatorFramework, ZkUtils.TopicConfigPath(), true);
+      topicConfigPathCache = new PathChildrenCache(curatorFramework, ZkUtils.getEntityConfigPath(ConfigType.Topic()), true);
       topicConfigPathCache.getListenable().addListener((f, e) -> {
          if (e.getType() == PathChildrenCacheEvent.Type.INITIALIZED)
          {
@@ -172,25 +173,25 @@ private String clientId()
    private void updateController()
    {
       Optional.ofNullable(controllerNodeCache.getCurrentData())
-         .map(data -> {
-            try
-            {
-               Map controllerData = objectMapper.reader(Map.class).readValue(data.getData());
-               return (Integer) controllerData.get("brokerid");
-            }
-            catch (IOException e)
-            {
-               LOG.error("Unable to read controller data", e);
-               return null;
-            }
-         })
-         .ifPresent(this::updateController);
+        .map(data -> {
+           try
+           {
+              Map controllerData = objectMapper.reader(Map.class).readValue(data.getData());
+              return (Integer) controllerData.get("brokerid");
+           }
+           catch (IOException e)
+           {
+              LOG.error("Unable to read controller data", e);
+              return null;
+           }
+        })
+        .ifPresent(this::updateController);
    }
 
    private void updateController(int brokerId)
    {
       brokerCache.values()
-         .forEach(broker -> broker.setController(broker.getId() == brokerId));
+        .forEach(broker -> broker.setController(broker.getId() == brokerId));
    }
 
    private void validateInitialized()
@@ -257,7 +258,7 @@ private BrokerChannel brokerChannel(Integer brokerId)
 
       Integer finalBrokerId = brokerId;
       BrokerVO broker = getBroker(brokerId)
-         .orElseThrow(() -> new BrokerNotFoundException("Broker " + finalBrokerId + " is not available"));
+                          .orElseThrow(() -> new BrokerNotFoundException("Broker " + finalBrokerId + " is not available"));
 
       return BrokerChannel.forBroker(broker.getHost(), broker.getPort());
    }
@@ -284,33 +285,33 @@ public ClusterSummaryVO getClusterSummary()
    @Override
    public ClusterSummaryVO getClusterSummary(Collection<TopicVO> topics) {
       final ClusterSummaryVO topicSummary = topics.stream()
-              .map(topic -> {
-                 ClusterSummaryVO summary = new ClusterSummaryVO();
-                 summary.setPartitionCount(topic.getPartitions().size());
-                 summary.setUnderReplicatedCount(topic.getUnderReplicatedPartitions().size());
-                 summary.setPreferredReplicaPercent(topic.getPreferredReplicaPercent());
-                 topic.getPartitions()
-                         .forEach(partition -> {
-                            if (partition.getLeader() != null) {
-                               summary.addBrokerLeaderPartition(partition.getLeader().getId());
-                            }
-                            if (partition.getPreferredLeader() != null) {
-                               summary.addBrokerPreferredLeaderPartition(partition.getPreferredLeader().getId());
-                            }
-                            partition.getReplicas()
-                                    .forEach(replica -> summary.addExpectedBrokerId(replica.getId()));
-                         });
-                 return summary;
-              })
-              .reduce((s1, s2) -> {
-                 s1.setPartitionCount(s1.getPartitionCount() + s2.getPartitionCount());
-                 s1.setUnderReplicatedCount(s1.getUnderReplicatedCount() + s2.getUnderReplicatedCount());
-                 s1.setPreferredReplicaPercent(s1.getPreferredReplicaPercent() + s2.getPreferredReplicaPercent());
-                 s2.getBrokerLeaderPartitionCount().forEach(s1::addBrokerLeaderPartition);
-                 s2.getBrokerPreferredLeaderPartitionCount().forEach(s1::addBrokerPreferredLeaderPartition);
-                 return s1;
-              })
-              .orElseGet(ClusterSummaryVO::new);
+                                              .map(topic -> {
+                                                 ClusterSummaryVO summary = new ClusterSummaryVO();
+                                                 summary.setPartitionCount(topic.getPartitions().size());
+                                                 summary.setUnderReplicatedCount(topic.getUnderReplicatedPartitions().size());
+                                                 summary.setPreferredReplicaPercent(topic.getPreferredReplicaPercent());
+                                                 topic.getPartitions()
+                                                   .forEach(partition -> {
+                                                      if (partition.getLeader() != null) {
+                                                         summary.addBrokerLeaderPartition(partition.getLeader().getId());
+                                                      }
+                                                      if (partition.getPreferredLeader() != null) {
+                                                         summary.addBrokerPreferredLeaderPartition(partition.getPreferredLeader().getId());
+                                                      }
+                                                      partition.getReplicas()
+                                                        .forEach(replica -> summary.addExpectedBrokerId(replica.getId()));
+                                                   });
+                                                 return summary;
+                                              })
+                                              .reduce((s1, s2) -> {
+                                                 s1.setPartitionCount(s1.getPartitionCount() + s2.getPartitionCount());
+                                                 s1.setUnderReplicatedCount(s1.getUnderReplicatedCount() + s2.getUnderReplicatedCount());
+                                                 s1.setPreferredReplicaPercent(s1.getPreferredReplicaPercent() + s2.getPreferredReplicaPercent());
+                                                 s2.getBrokerLeaderPartitionCount().forEach(s1::addBrokerLeaderPartition);
+                                                 s2.getBrokerPreferredLeaderPartitionCount().forEach(s1::addBrokerPreferredLeaderPartition);
+                                                 return s1;
+                                              })
+                                              .orElseGet(ClusterSummaryVO::new);
       topicSummary.setTopicCount(topics.size());
       topicSummary.setPreferredReplicaPercent(topicSummary.getPreferredReplicaPercent() / topics.size());
       return topicSummary;
@@ -321,8 +322,8 @@ public List<TopicVO> getTopics()
    {
       validateInitialized();
       return getTopicMetadata().values().stream()
-         .sorted(Comparator.comparing(TopicVO::getName))
-         .collect(Collectors.toList());
+               .sorted(Comparator.comparing(TopicVO::getName))
+               .collect(Collectors.toList());
    }
 
    @Override
@@ -331,14 +332,14 @@ public Optional<TopicVO> getTopic(String topic)
       validateInitialized();
       final Optional<TopicVO> topicVO = Optional.ofNullable(getTopicMetadata(topic).get(topic));
       topicVO.ifPresent(
-         vo -> {
-            getTopicPartitionSizes(vo, kafka.api.OffsetRequest.LatestTime())
-               .entrySet()
-               .forEach(entry -> vo.getPartition(entry.getKey()).ifPresent(p -> p.setSize(entry.getValue())));
-            getTopicPartitionSizes(vo, kafka.api.OffsetRequest.EarliestTime())
-               .entrySet()
-               .forEach(entry -> vo.getPartition(entry.getKey()).ifPresent(p -> p.setFirstOffset(entry.getValue())));
-         }
+        vo -> {
+           getTopicPartitionSizes(vo, kafka.api.OffsetRequest.LatestTime())
+             .entrySet()
+             .forEach(entry -> vo.getPartition(entry.getKey()).ifPresent(p -> p.setSize(entry.getValue())));
+           getTopicPartitionSizes(vo, kafka.api.OffsetRequest.EarliestTime())
+             .entrySet()
+             .forEach(entry -> vo.getPartition(entry.getKey()).ifPresent(p -> p.setFirstOffset(entry.getValue())));
+        }
       );
       return topicVO;
    }
@@ -348,8 +349,8 @@ private Map<String, TopicVO> getTopicMetadata(String... topics)
       if (kafkaVersion.compareTo(new Version(0, 9, 0)) >= 0)
       {
          return retryTemplate.execute(
-            context -> brokerChannel(null)
-               .execute(channel -> getTopicMetadata(channel, topics)));
+           context -> brokerChannel(null)
+                        .execute(channel -> getTopicMetadata(channel, topics)));
       }
       else
       {
@@ -357,11 +358,11 @@ private Map<String, TopicVO> getTopicMetadata(String... topics)
          if (topics == null || topics.length == 0)
          {
             topicStream =
-               Optional.ofNullable(
-                  topicTreeCache.getCurrentChildren(ZkUtils.BrokerTopicsPath()))
-                  .map(Map::keySet)
-                  .map(Collection::stream)
-                  .orElse(Stream.empty());
+              Optional.ofNullable(
+                topicTreeCache.getCurrentChildren(ZkUtils.BrokerTopicsPath()))
+                .map(Map::keySet)
+                .map(Collection::stream)
+                .orElse(Stream.empty());
          }
          else
          {
@@ -369,17 +370,17 @@ private Map<String, TopicVO> getTopicMetadata(String... topics)
          }
 
          return topicStream
-            .map(this::getTopicZkData)
-            .filter(Objects::nonNull)
-            .collect(Collectors.toMap(TopicVO::getName, topic -> topic));
+                  .map(this::getTopicZkData)
+                  .filter(Objects::nonNull)
+                  .collect(Collectors.toMap(TopicVO::getName, topic -> topic));
       }
    }
 
    private TopicVO getTopicZkData(String topic)
    {
       return Optional.ofNullable(topicTreeCache.getCurrentData(ZkUtils.getTopicPath(topic)))
-         .map(this::parseZkTopic)
-         .orElse(null);
+               .map(this::parseZkTopic)
+               .orElse(null);
    }
 
    public TopicVO parseZkTopic(ChildData input)
@@ -389,12 +390,12 @@ public TopicVO parseZkTopic(ChildData input)
          final TopicVO topic = new TopicVO(StringUtils.substringAfterLast(input.getPath(), "/"));
 
          final TopicRegistrationVO topicRegistration =
-            objectMapper.reader(TopicRegistrationVO.class).readValue(input.getData());
+           objectMapper.reader(TopicRegistrationVO.class).readValue(input.getData());
 
          topic.setConfig(
-            Optional.ofNullable(topicConfigPathCache.getCurrentData(ZkUtils.TopicConfigPath() + "/" + topic.getName()))
-               .map(this::readTopicConfig)
-               .orElse(Collections.emptyMap()));
+           Optional.ofNullable(topicConfigPathCache.getCurrentData(ZkUtils.getEntityConfigPath(ConfigType.Topic()) + "/" + topic.getName()))
+             .map(this::readTopicConfig)
+             .orElse(Collections.emptyMap()));
 
          for (Map.Entry<Integer, List<Integer>> entry : topicRegistration.getReplicas().entrySet())
          {
@@ -406,14 +407,14 @@ public TopicVO parseZkTopic(ChildData input)
             final Optional<TopicPartitionStateVO> partitionState = partitionState(topic.getName(), partition.getId());
 
             partitionBrokerIds.stream()
-               .map(brokerId -> {
-                  TopicPartitionVO.PartitionReplica replica = new TopicPartitionVO.PartitionReplica();
-                  replica.setId(brokerId);
-                  replica.setInService(partitionState.map(ps -> ps.getIsr().contains(brokerId)).orElse(false));
-                  replica.setLeader(partitionState.map(ps -> brokerId == ps.getLeader()).orElse(false));
-                  return replica;
-               })
-               .forEach(partition::addReplica);
+              .map(brokerId -> {
+                 TopicPartitionVO.PartitionReplica replica = new TopicPartitionVO.PartitionReplica();
+                 replica.setId(brokerId);
+                 replica.setInService(partitionState.map(ps -> ps.getIsr().contains(brokerId)).orElse(false));
+                 replica.setLeader(partitionState.map(ps -> brokerId == ps.getLeader()).orElse(false));
+                 return replica;
+              })
+              .forEach(partition::addReplica);
 
             topic.addPartition(partition);
          }
@@ -431,21 +432,21 @@ public TopicVO parseZkTopic(ChildData input)
    private Map<String, TopicVO> getTopicMetadata(BlockingChannel channel, String... topics)
    {
       final TopicMetadataRequest request =
-         new TopicMetadataRequest((short) 0, 0, clientId(), Arrays.asList(topics));
+        new TopicMetadataRequest((short) 0, 0, clientId(), Arrays.asList(topics));
 
       LOG.debug("Sending topic metadata request: {}", request);
 
       channel.send(request);
       final kafka.api.TopicMetadataResponse underlyingResponse =
-         kafka.api.TopicMetadataResponse.readFrom(channel.receive().buffer());
+        kafka.api.TopicMetadataResponse.readFrom(channel.receive().payload());
 
       LOG.debug("Received topic metadata response: {}", underlyingResponse);
 
       TopicMetadataResponse response = new TopicMetadataResponse(underlyingResponse);
       return response.topicsMetadata().stream()
-         .filter(tmd -> tmd.errorCode() == ErrorMapping.NoError())
-         .map(this::processTopicMetadata)
-         .collect(Collectors.toMap(TopicVO::getName, t -> t));
+               .filter(tmd -> tmd.errorCode() == ErrorMapping.NoError())
+               .map(this::processTopicMetadata)
+               .collect(Collectors.toMap(TopicVO::getName, t -> t));
    }
 
    private TopicVO processTopicMetadata(TopicMetadata tmd)
@@ -453,14 +454,14 @@ private TopicVO processTopicMetadata(TopicMetadata tmd)
       TopicVO topic = new TopicVO(tmd.topic());
 
       topic.setConfig(
-         Optional.ofNullable(topicConfigPathCache.getCurrentData(ZkUtils.TopicConfigPath() + "/" + topic.getName()))
-            .map(this::readTopicConfig)
-            .orElse(Collections.emptyMap()));
+        Optional.ofNullable(topicConfigPathCache.getCurrentData(ZkUtils.getEntityConfigPath(ConfigType.Topic()) + "/" + topic.getName()))
+          .map(this::readTopicConfig)
+          .orElse(Collections.emptyMap()));
 
       topic.setPartitions(
-         tmd.partitionsMetadata().stream()
-            .map((pmd) -> parsePartitionMetadata(tmd.topic(), pmd))
-            .collect(Collectors.toMap(TopicPartitionVO::getId, p -> p))
+        tmd.partitionsMetadata().stream()
+          .map((pmd) -> parsePartitionMetadata(tmd.topic(), pmd))
+          .collect(Collectors.toMap(TopicPartitionVO::getId, p -> p))
       );
       return topic;
    }
@@ -475,14 +476,14 @@ private TopicPartitionVO parsePartitionMetadata(String topic, PartitionMetadata
 
       final List<Integer> isr = getIsr(topic, pmd);
       pmd.replicas().stream()
-         .map(replica -> new TopicPartitionVO.PartitionReplica(replica.id(), isr.contains(replica.id()), false))
-         .forEach(partition::addReplica);
+        .map(replica -> new TopicPartitionVO.PartitionReplica(replica.id(), isr.contains(replica.id()), false))
+        .forEach(partition::addReplica);
       return partition;
    }
 
    private List<Integer> getIsr(String topic, PartitionMetadata pmd)
    {
-      return pmd.isr().stream().map(Broker::id).collect(Collectors.toList());
+      return pmd.isr().stream().map(BrokerEndPoint::id).collect(Collectors.toList());
    }
 
    private Map<String, Object> readTopicConfig(ChildData d)
@@ -500,11 +501,11 @@ private Map<String, Object> readTopicConfig(ChildData d)
 
 
    private Optional<TopicPartitionStateVO> partitionState(String topicName, int partitionId)
-      throws IOException
+     throws IOException
    {
       final Optional<byte[]> partitionData = Optional.ofNullable(topicTreeCache.getCurrentData(
-              ZkUtils.getTopicPartitionLeaderAndIsrPath(topicName, partitionId)))
-              .map(ChildData::getData);
+        ZkUtils.getTopicPartitionLeaderAndIsrPath(topicName, partitionId)))
+                                               .map(ChildData::getData);
       if (partitionData.isPresent())
       {
          return Optional.ofNullable(objectMapper.reader(TopicPartitionStateVO.class).readValue(partitionData.get()));
@@ -527,8 +528,8 @@ public List<ConsumerVO> getConsumers(final TopicVO topic)
    {
       validateInitialized();
       return getConsumerStream(topic)
-         .filter(consumer -> consumer.getTopic(topic.getName()) != null)
-         .collect(Collectors.toList());
+               .filter(consumer -> consumer.getTopic(topic.getName()) != null)
+               .collect(Collectors.toList());
    }
 
    @Override
@@ -540,10 +541,10 @@ public List<ConsumerVO> getConsumers(final String topic)
    private Stream<ConsumerVO> getConsumerStream(TopicVO topic)
    {
       return consumerTreeCache.getCurrentChildren(ZkUtils.ConsumersPath()).keySet().stream()
-         .map(g -> getConsumerByTopic(g, topic))
-         .filter(Optional::isPresent)
-         .map(Optional::get)
-         .sorted(Comparator.comparing(ConsumerVO::getGroupId));
+               .map(g -> getConsumerByTopic(g, topic))
+               .filter(Optional::isPresent)
+               .map(Optional::get)
+               .sorted(Comparator.comparing(ConsumerVO::getGroupId));
    }
 
    @Override
@@ -569,11 +570,11 @@ public Optional<ConsumerVO> getConsumerByTopic(String groupId, TopicVO topic)
 
       // todo: get number of threads in each instance (subscription -> topic -> # threads)
       Optional.ofNullable(consumerTreeCache.getCurrentChildren(groupDirs.consumerRegistryDir()))
-         .ifPresent(
-            children ->
-               children.keySet().stream()
-                  .map(id -> readConsumerRegistration(groupDirs, id))
-                  .forEach(consumer::addActiveInstance));
+        .ifPresent(
+          children ->
+            children.keySet().stream()
+              .map(id -> readConsumerRegistration(groupDirs, id))
+              .forEach(consumer::addActiveInstance));
 
       Stream<String> topicStream = null;
 
@@ -591,19 +592,19 @@ public Optional<ConsumerVO> getConsumerByTopic(String groupId, TopicVO topic)
       else
       {
          topicStream = Optional.ofNullable(
-            consumerTreeCache.getCurrentChildren(groupDirs.consumerGroupDir() + "/owners"))
-            .map(Map::keySet)
-            .map(Collection::stream)
-            .orElse(Stream.empty());
+           consumerTreeCache.getCurrentChildren(groupDirs.consumerGroupDir() + "/owners"))
+                         .map(Map::keySet)
+                         .map(Collection::stream)
+                         .orElse(Stream.empty());
       }
 
       topicStream
-         .map(ConsumerTopicVO::new)
-         .forEach(consumerTopic -> {
-            getConsumerPartitionStream(groupId, consumerTopic.getTopic(), topic)
-               .forEach(consumerTopic::addOffset);
-            consumer.addTopic(consumerTopic);
-         });
+        .map(ConsumerTopicVO::new)
+        .forEach(consumerTopic -> {
+           getConsumerPartitionStream(groupId, consumerTopic.getTopic(), topic)
+             .forEach(consumerTopic::addOffset);
+           consumer.addTopic(consumerTopic);
+        });
 
       return Optional.of(consumer);
    }
@@ -644,24 +645,24 @@ private Stream<ConsumerPartitionVO> getConsumerPartitionStream(String groupId,
          Map<Integer, Long> consumerOffsets = getConsumerOffsets(groupId, topic);
 
          return topic.getPartitions().stream()
-            .map(partition -> {
-               int partitionId = partition.getId();
+                  .map(partition -> {
+                     int partitionId = partition.getId();
 
-               final ConsumerPartitionVO consumerPartition = new ConsumerPartitionVO(groupId, topicName, partitionId);
-               consumerPartition.setOwner(
-                  Optional.ofNullable(
-                     consumerTreeCache.getCurrentData(groupTopicDirs.consumerOwnerDir() + "/" + partitionId))
-                     .map(data -> new String(data.getData()))
-                     .orElse(null));
+                     final ConsumerPartitionVO consumerPartition = new ConsumerPartitionVO(groupId, topicName, partitionId);
+                     consumerPartition.setOwner(
+                       Optional.ofNullable(
+                         consumerTreeCache.getCurrentData(groupTopicDirs.consumerOwnerDir() + "/" + partitionId))
+                         .map(data -> new String(data.getData()))
+                         .orElse(null));
 
-               consumerPartition.setOffset(consumerOffsets.getOrDefault(partitionId, -1L));
+                     consumerPartition.setOffset(consumerOffsets.getOrDefault(partitionId, -1L));
 
-               final Optional<TopicPartitionVO> topicPartition = topic.getPartition(partitionId);
-               consumerPartition.setSize(topicPartition.map(TopicPartitionVO::getSize).orElse(-1L));
-               consumerPartition.setFirstOffset(topicPartition.map(TopicPartitionVO::getFirstOffset).orElse(-1L));
+                     final Optional<TopicPartitionVO> topicPartition = topic.getPartition(partitionId);
+                     consumerPartition.setSize(topicPartition.map(TopicPartitionVO::getSize).orElse(-1L));
+                     consumerPartition.setFirstOffset(topicPartition.map(TopicPartitionVO::getFirstOffset).orElse(-1L));
 
-               return consumerPartition;
-            });
+                     return consumerPartition;
+                  });
       }
       else
       {
@@ -678,15 +679,15 @@ private Map<Integer, Long> getConsumerOffsets(String groupId, TopicVO topic)
          // for both and assume that the largest offset is the correct one.
 
          ForkJoinTask<Map<Integer, Long>> kafkaTask =
-            threadPool.submit(() -> getConsumerOffsets(groupId, topic, false));
+           threadPool.submit(() -> getConsumerOffsets(groupId, topic, false));
 
          ForkJoinTask<Map<Integer, Long>> zookeeperTask =
-            threadPool.submit(() -> getConsumerOffsets(groupId, topic, true));
+           threadPool.submit(() -> getConsumerOffsets(groupId, topic, true));
 
          Map<Integer, Long> zookeeperOffsets = zookeeperTask.get();
          Map<Integer, Long> kafkaOffsets = kafkaTask.get();
          zookeeperOffsets.entrySet()
-            .forEach(entry -> kafkaOffsets.merge(entry.getKey(), entry.getValue(), Math::max));
+           .forEach(entry -> kafkaOffsets.merge(entry.getKey(), entry.getValue(), Math::max));
          return kafkaOffsets;
       }
       catch (InterruptedException ex)
@@ -705,8 +706,8 @@ private Map<Integer, Long> getConsumerOffsets(String groupId,
                                                  boolean zookeeperOffsets)
    {
       return retryTemplate.execute(
-         context -> brokerChannel(zookeeperOffsets ? null : offsetManagerBroker(groupId))
-            .execute(channel -> getConsumerOffsets(channel, groupId, topic, zookeeperOffsets)));
+        context -> brokerChannel(zookeeperOffsets ? null : offsetManagerBroker(groupId))
+                     .execute(channel -> getConsumerOffsets(channel, groupId, topic, zookeeperOffsets)));
    }
 
    /**
@@ -729,27 +730,27 @@ private Map<Integer, Long> getConsumerOffsets(BlockingChannel channel,
    {
 
       final OffsetFetchRequest request = new OffsetFetchRequest(
-         groupId,
-         topic.getPartitions().stream()
-            .map(p -> new TopicAndPartition(topic.getName(), p.getId()))
-            .collect(Collectors.toList()),
-         (short) (zookeeperOffsets ? 0 : 1), 0, // version 0 = zookeeper offsets, 1 = kafka offsets
-         clientId());
+        groupId,
+        topic.getPartitions().stream()
+          .map(p -> new TopicAndPartition(topic.getName(), p.getId()))
+          .collect(Collectors.toList()),
+        (short) (zookeeperOffsets ? 0 : 1), 0, // version 0 = zookeeper offsets, 1 = kafka offsets
+        clientId());
 
       LOG.debug("Sending consumer offset request: {}", request);
 
       channel.send(request.underlying());
 
       final kafka.api.OffsetFetchResponse underlyingResponse =
-         kafka.api.OffsetFetchResponse.readFrom(channel.receive().buffer());
+        kafka.api.OffsetFetchResponse.readFrom(channel.receive().payload());
 
       LOG.debug("Received consumer offset response: {}", underlyingResponse);
 
       OffsetFetchResponse response = new OffsetFetchResponse(underlyingResponse);
 
       return response.offsets().entrySet().stream()
-         .filter(entry -> entry.getValue().error() == ErrorMapping.NoError())
-         .collect(Collectors.toMap(entry -> entry.getKey().partition(), entry -> entry.getValue().offset()));
+               .filter(entry -> entry.getValue().error().code() == ErrorMapping.NoError())
+               .collect(Collectors.toMap(entry -> entry.getKey().partition(), entry -> entry.getValue().offset()));
    }
 
    /**
@@ -758,22 +759,22 @@ private Map<Integer, Long> getConsumerOffsets(BlockingChannel channel,
    private Integer offsetManagerBroker(String groupId)
    {
       return retryTemplate.execute(
-         context ->
-            brokerChannel(null)
-               .execute(channel -> offsetManagerBroker(channel, groupId))
+        context ->
+          brokerChannel(null)
+            .execute(channel -> offsetManagerBroker(channel, groupId))
       );
    }
 
    private Integer offsetManagerBroker(BlockingChannel channel, String groupId)
    {
-      final ConsumerMetadataRequest request =
-         new ConsumerMetadataRequest(groupId, (short) 0, 0, clientId());
+      final GroupCoordinatorRequest request =
+        new GroupCoordinatorRequest(groupId, (short) 0, 0, clientId());
 
       LOG.debug("Sending consumer metadata request: {}", request);
 
       channel.send(request);
-      ConsumerMetadataResponse response =
-         ConsumerMetadataResponse.readFrom(channel.receive().buffer());
+      GroupCoordinatorResponse response =
+        GroupCoordinatorResponse.readFrom(channel.receive().payload());
 
       LOG.debug("Received consumer metadata response: {}", response);
 
@@ -792,48 +793,48 @@ private Map<Integer, Long> getTopicPartitionSizes(TopicVO topic, long time)
          PartitionOffsetRequestInfo requestInfo = new PartitionOffsetRequestInfo(time, 1);
 
          return threadPool.submit(() ->
-               topic.getPartitions().parallelStream()
-                  .filter(p -> p.getLeader() != null)
-                  .collect(Collectors.groupingBy(p -> p.getLeader().getId())) // Group partitions by leader broker id
-                  .entrySet().parallelStream()
-                  .map(entry -> {
-                     final Integer brokerId = entry.getKey();
-                     final List<TopicPartitionVO> brokerPartitions = entry.getValue();
-                     try
-                     {
-                        // Get the size of the partitions for a topic from the leader.
-                        final OffsetResponse offsetResponse =
-                           sendOffsetRequest(brokerId, topic, requestInfo, brokerPartitions);
-
-
-                        // Build a map of partitionId -> topic size from the response
-                        return brokerPartitions.stream()
-                           .collect(Collectors.toMap(TopicPartitionVO::getId,
-                                                     partition -> Optional.ofNullable(
-                                                        offsetResponse.offsets(topic.getName(), partition.getId()))
-                                                        .map(Arrays::stream)
-                                                        .orElse(LongStream.empty())
-                                                        .findFirst()
-                                                        .orElse(-1L)));
-                     }
-                     catch (Exception ex)
-                     {
-                        LOG.error("Unable to get partition log size for topic {} partitions ({})",
-                                  topic.getName(),
-                                  brokerPartitions.stream()
-                                     .map(TopicPartitionVO::getId)
-                                     .map(String::valueOf)
-                                     .collect(Collectors.joining(",")),
-                                  ex);
-
-                        // Map each partition to -1, indicating we got an error
-                        return brokerPartitions.stream().collect(Collectors.toMap(TopicPartitionVO::getId, tp -> -1L));
-                     }
-                  })
-                  .map(Map::entrySet)
-                  .flatMap(Collection::stream)
-                  .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)))
-            .get();
+                                    topic.getPartitions().parallelStream()
+                                      .filter(p -> p.getLeader() != null)
+                                      .collect(Collectors.groupingBy(p -> p.getLeader().getId())) // Group partitions by leader broker id
+                                      .entrySet().parallelStream()
+                                      .map(entry -> {
+                                         final Integer brokerId = entry.getKey();
+                                         final List<TopicPartitionVO> brokerPartitions = entry.getValue();
+                                         try
+                                         {
+                                            // Get the size of the partitions for a topic from the leader.
+                                            final OffsetResponse offsetResponse =
+                                              sendOffsetRequest(brokerId, topic, requestInfo, brokerPartitions);
+
+
+                                            // Build a map of partitionId -> topic size from the response
+                                            return brokerPartitions.stream()
+                                                     .collect(Collectors.toMap(TopicPartitionVO::getId,
+                                                       partition -> Optional.ofNullable(
+                                                         offsetResponse.offsets(topic.getName(), partition.getId()))
+                                                                      .map(Arrays::stream)
+                                                                      .orElse(LongStream.empty())
+                                                                      .findFirst()
+                                                                      .orElse(-1L)));
+                                         }
+                                         catch (Exception ex)
+                                         {
+                                            LOG.error("Unable to get partition log size for topic {} partitions ({})",
+                                              topic.getName(),
+                                              brokerPartitions.stream()
+                                                .map(TopicPartitionVO::getId)
+                                                .map(String::valueOf)
+                                                .collect(Collectors.joining(",")),
+                                              ex);
+
+                                            // Map each partition to -1, indicating we got an error
+                                            return brokerPartitions.stream().collect(Collectors.toMap(TopicPartitionVO::getId, tp -> -1L));
+                                         }
+                                      })
+                                      .map(Map::entrySet)
+                                      .flatMap(Collection::stream)
+                                      .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)))
+                  .get();
       }
       catch (InterruptedException e)
       {
@@ -851,27 +852,27 @@ private OffsetResponse sendOffsetRequest(Integer brokerId, TopicVO topic,
                                             List<TopicPartitionVO> brokerPartitions)
    {
       final OffsetRequest offsetRequest = new OffsetRequest(
-         brokerPartitions.stream()
-            .collect(Collectors.toMap(
-               partition -> new TopicAndPartition(topic.getName(), partition.getId()),
-               partition -> requestInfo)),
-         (short) 0, clientId());
+        brokerPartitions.stream()
+          .collect(Collectors.toMap(
+            partition -> new TopicAndPartition(topic.getName(), partition.getId()),
+            partition -> requestInfo)),
+        (short) 0, clientId());
 
       LOG.debug("Sending offset request: {}", offsetRequest);
 
       return retryTemplate.execute(
-         context ->
-            brokerChannel(brokerId)
-               .execute(channel ->
-                        {
-                           channel.send(offsetRequest.underlying());
-                           final kafka.api.OffsetResponse underlyingResponse =
-                              kafka.api.OffsetResponse.readFrom(channel.receive().buffer());
+        context ->
+          brokerChannel(brokerId)
+            .execute(channel ->
+            {
+               channel.send(offsetRequest.underlying());
+               final kafka.api.OffsetResponse underlyingResponse =
+                 kafka.api.OffsetResponse.readFrom(channel.receive().payload());
 
-                           LOG.debug("Received offset response: {}", underlyingResponse);
+               LOG.debug("Received offset response: {}", underlyingResponse);
 
-                           return new OffsetResponse(underlyingResponse);
-                        }));
+               return new OffsetResponse(underlyingResponse);
+            }));
    }
 
    private class BrokerListener implements PathChildrenCacheListener
@@ -897,8 +898,8 @@ public void childEvent(CuratorFramework framework, PathChildrenCacheEvent event)
             case INITIALIZED:
             {
                brokerPathCache.getCurrentData().stream()
-                  .map(BrokerListener.this::parseBroker)
-                  .forEach(CuratorKafkaMonitor.this::addBroker);
+                 .map(BrokerListener.this::parseBroker)
+                 .forEach(CuratorKafkaMonitor.this::addBroker);
                break;
             }
          }
@@ -926,4 +927,4 @@ private BrokerVO parseBroker(ChildData input)
       }
    }
 
-}
+}
\ No newline at end of file