// Path: src/main/java/kafka/common/TopicAndPartition.java
// public class TopicAndPartition {
//     public final String topic;
//     public final int partition;
// 
//     public TopicAndPartition(String topic, int partition) {
//         this.topic = topic;
//         this.partition = partition;
//     }
// 
//     public TopicAndPartition(Tuple2<String, Integer> tuple) {
//         this(tuple._1, tuple._2);
//     }
// 
//     public TopicAndPartition(Partition partition) {
//         this(partition.topic, partition.partitionId);
//     }
// 
//     public TopicAndPartition(Replica replica) {
//         this(replica.topic, replica.partitionId);
//     }
// 
//     public Tuple2<String, Integer> asTuple() {
//         return Tuple2.make(topic, partition);
//     }
// 
//     @Override
//     public String toString() {
//         return String.format("[%s,%d]", topic, partition);
//     }
// 
//     @Override
//     public boolean equals(Object o) {
//         if (this == o) return true;
//         if (o == null || getClass() != o.getClass()) return false;
// 
//         TopicAndPartition that = (TopicAndPartition) o;
// 
//         if (partition != that.partition) return false;
//         if (topic != null ? !topic.equals(that.topic) : that.topic != null) return false;
// 
//         return true;
//     }
// 
//     @Override
//     public int hashCode() {
//         int result = topic != null ? topic.hashCode() : 0;
//         result = 31 * result + partition;
//         return result;
//     }
// }
// 
// Path: src/main/java/kafka/metrics/KafkaMetricsGroup.java
// public abstract class KafkaMetricsGroup {
//     /**
//      * Creates a new MetricName object for gauges, meters, etc. created for this
//      * metrics group.
//      *
//      * @param name Descriptive name of the metric.
//      * @return Sanitized metric name object.
//      */
//     private MetricName metricName(String name) {
//         Class<?> klass = this.getClass();
//         String pkg = (klass.getPackage() == null) ? "" : klass.getPackage().getName();
//         String simpleName = klass.getSimpleName().replaceAll("\\$$", "");
//         return new MetricName(pkg, simpleName, name);
//     }
// 
//     public <T> Gauge<T> newGauge(String name, Gauge<T> metric) {
//         return Metrics.defaultRegistry().newGauge(metricName(name), metric);
//     }
// 
//     public Meter newMeter(String name, String eventType, TimeUnit timeUnit) {
//         return Metrics.defaultRegistry().newMeter(metricName(name), eventType, timeUnit);
//     }
// 
//     public Histogram newHistogram(String name) {
//         return newHistogram(name, true);
//     }
//     public Histogram newHistogram(String name, boolean biased/* = true*/) {
//         return Metrics.defaultRegistry().newHistogram(metricName(name), biased);
//     }
// 
//     public Timer newTimer(String name, TimeUnit durationUnit, TimeUnit rateUnit) {
//         return Metrics.defaultRegistry().newTimer(metricName(name), durationUnit, rateUnit);
//     }
// }


import com.google.common.base.Predicate;
import com.google.common.collect.HashMultimap;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Multimap;
import com.yammer.metrics.core.Gauge;
import kafka.cluster.Broker;
import kafka.cluster.Cluster;
import kafka.common.ConsumerRebalanceFailedException;
import kafka.common.TopicAndPartition;
import kafka.metrics.KafkaMetricsGroup;
import kafka.metrics.KafkaMetricsReporter;
import kafka.serializer.Decoder;
import kafka.serializer.DefaultDecoder;
import kafka.utils.*;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.IZkStateListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkNodeExistsException;
import org.apache.zookeeper.Watcher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import static com.google.common.base.Preconditions.checkState;
import static kafka.utils.ZkUtils.*;
        consumerIdString = config.groupId + "_" + consumerUuid;

        logger = LoggerFactory.getLogger(ZookeeperConsumerConnector.class + "[" + consumerIdString + "]");

        connectZk();
        createFetcher();
        if (config.autoCommitEnable) {
            scheduler.startup();
            logger.info("starting auto committer every {} ms", config.autoCommitIntervalMs);
            scheduler.schedule("kafka-consumer-autocommit", new Runnable() {
                @Override
                public void run() {
                    autoCommit();
                }
            },
                    /*delay =*/ config.autoCommitIntervalMs,
                    /*period =*/ config.autoCommitIntervalMs,
                    /*unit =*/ TimeUnit.MILLISECONDS);
        }

        KafkaMetricsReporter.startReporters(config.props);
    }

    Logger logger = LoggerFactory.getLogger(ZookeeperConsumerConnector.class);

    private AtomicBoolean isShuttingDown = new AtomicBoolean(false);
    private Object rebalanceLock = new Object();
    private ConsumerFetcherManager fetcher = null;
    private ZkClient zkClient = null;
    private Pool<String, Pool<Integer, PartitionTopicInfo>> topicRegistry = new Pool<String, Pool<Integer, PartitionTopicInfo>>();
// Path: src/main/java/kafka/common/TopicAndPartition.java
// public class TopicAndPartition {
//     public final String topic;
//     public final int partition;
// 
//     public TopicAndPartition(String topic, int partition) {
//         this.topic = topic;
//         this.partition = partition;
//     }
// 
//     public TopicAndPartition(Tuple2<String, Integer> tuple) {
//         this(tuple._1, tuple._2);
//     }
// 
//     public TopicAndPartition(Partition partition) {
//         this(partition.topic, partition.partitionId);
//     }
// 
//     public TopicAndPartition(Replica replica) {
//         this(replica.topic, replica.partitionId);
//     }
// 
//     public Tuple2<String, Integer> asTuple() {
//         return Tuple2.make(topic, partition);
//     }
// 
//     @Override
//     public String toString() {
//         return String.format("[%s,%d]", topic, partition);
//     }
// 
//     @Override
//     public boolean equals(Object o) {
//         if (this == o) return true;
//         if (o == null || getClass() != o.getClass()) return false;
// 
//         TopicAndPartition that = (TopicAndPartition) o;
// 
//         if (partition != that.partition) return false;
//         if (topic != null ? !topic.equals(that.topic) : that.topic != null) return false;
// 
//         return true;
//     }
// 
//     @Override
//     public int hashCode() {
//         int result = topic != null ? topic.hashCode() : 0;
//         result = 31 * result + partition;
//         return result;
//     }
// }
// 
// Path: src/main/java/kafka/metrics/KafkaMetricsGroup.java
// public abstract class KafkaMetricsGroup {
//     /**
//      * Creates a new MetricName object for gauges, meters, etc. created for this
//      * metrics group.
//      *
//      * @param name Descriptive name of the metric.
//      * @return Sanitized metric name object.
//      */
//     private MetricName metricName(String name) {
//         Class<?> klass = this.getClass();
//         String pkg = (klass.getPackage() == null) ? "" : klass.getPackage().getName();
//         String simpleName = klass.getSimpleName().replaceAll("\\$$", "");
//         return new MetricName(pkg, simpleName, name);
//     }
// 
//     public <T> Gauge<T> newGauge(String name, Gauge<T> metric) {
//         return Metrics.defaultRegistry().newGauge(metricName(name), metric);
//     }
// 
//     public Meter newMeter(String name, String eventType, TimeUnit timeUnit) {
//         return Metrics.defaultRegistry().newMeter(metricName(name), eventType, timeUnit);
//     }
// 
//     public Histogram newHistogram(String name) {
//         return newHistogram(name, true);
//     }
//     public Histogram newHistogram(String name, boolean biased/* = true*/) {
//         return Metrics.defaultRegistry().newHistogram(metricName(name), biased);
//     }
// 
//     public Timer newTimer(String name, TimeUnit durationUnit, TimeUnit rateUnit) {
//         return Metrics.defaultRegistry().newTimer(metricName(name), durationUnit, rateUnit);
//     }
// }


// Path: src/main/java/kafka/consumer/ZookeeperConsumerConnector.java
import com.google.common.base.Predicate;
import com.google.common.collect.HashMultimap;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Multimap;
import com.yammer.metrics.core.Gauge;
import kafka.cluster.Broker;
import kafka.cluster.Cluster;
import kafka.common.ConsumerRebalanceFailedException;
import kafka.common.TopicAndPartition;
import kafka.metrics.KafkaMetricsGroup;
import kafka.metrics.KafkaMetricsReporter;
import kafka.serializer.Decoder;
import kafka.serializer.DefaultDecoder;
import kafka.utils.*;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.IZkStateListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkNodeExistsException;
import org.apache.zookeeper.Watcher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import static com.google.common.base.Preconditions.checkState;
import static kafka.utils.ZkUtils.*;

        consumerIdString = config.groupId + "_" + consumerUuid;

        logger = LoggerFactory.getLogger(ZookeeperConsumerConnector.class + "[" + consumerIdString + "]");

        connectZk();
        createFetcher();
        if (config.autoCommitEnable) {
            scheduler.startup();
            logger.info("starting auto committer every {} ms", config.autoCommitIntervalMs);
            scheduler.schedule("kafka-consumer-autocommit", new Runnable() {
                @Override
                public void run() {
                    autoCommit();
                }
            },
                    /*delay =*/ config.autoCommitIntervalMs,
                    /*period =*/ config.autoCommitIntervalMs,
                    /*unit =*/ TimeUnit.MILLISECONDS);
        }

        KafkaMetricsReporter.startReporters(config.props);
    }

    Logger logger = LoggerFactory.getLogger(ZookeeperConsumerConnector.class);

    private AtomicBoolean isShuttingDown = new AtomicBoolean(false);
    private Object rebalanceLock = new Object();
    private ConsumerFetcherManager fetcher = null;
    private ZkClient zkClient = null;
    private Pool<String, Pool<Integer, PartitionTopicInfo>> topicRegistry = new Pool<String, Pool<Integer, PartitionTopicInfo>>();
    private Pool<TopicAndPartition, Long> checkpointedOffsets = new Pool<TopicAndPartition, Long>();