public GroupInfo getGroupInfo(String groupId) throws IOException {
        ZookeeperUtils zkUtils = createZookeeperUtils();
        ZkClient zkClient = null;
        GroupInfo info = new GroupInfo(groupId);
        try {
            zkClient = zkUtils.getZkClient(zkServers);

            List<Broker> brokers = zkUtils.getBrokers(zkClient);
            List<String> topicsForGroup = zkUtils.getTopicsForGroup(zkClient, groupId);

            for (Broker broker : brokers) {
                SimpleConsumer consumer = null;
                try {
                    consumer = createSimpleConsumerFor(broker);
                    TopicMetadataRequest req = new TopicMetadataRequest(topicsForGroup);
                    TopicMetadataResponse resp = consumer.send(req);
                    List<TopicMetadata> metaData = resp.topicsMetadata();
                    for (TopicMetadata item : metaData) {
                        for (PartitionMetadata part : item.partitionsMetadata()) {
                            if (part.leader().getConnectionString().equalsIgnoreCase(broker.getConnectionString())) {
                                long logSize = getLastOffset(consumer, item.topic(), part.partitionId(), LatestTime(), CONSUMER_OFFSET_CHECKER);
                                if (logSize == -1) {
                                    throw new IllegalStateException("Log size cannot be -1");
                                }
                                long offset = zkUtils.getLastConsumedOffset(zkClient, groupId, item.topic(), part.partitionId());
                                if (offset > logSize) {
                                    offset = getLastOffset(consumer, item.topic(), part.partitionId(), EarliestTime(), CONSUMER_OFFSET_CHECKER);
                                }
                                if (offset == -1) {
                                    throw new IllegalStateException("Last offset cannot be -1");
                                }
                                info.addInfoFor(item.topic(), part.partitionId(), logSize, offset);
                            }
                        }
                    }
                } finally {
                    if (consumer != null)
                        consumer.close();
                }
            }
        } finally {
            zkUtils.close(zkClient);
        }

        return info;
    }