// 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/utils/Function2.java
// public interface Function2<A1,A2,B> {
//     B apply(A1 arg1, A2 arg2);
// }
// 
// Path: src/main/java/kafka/utils/Function3.java
// public interface Function3<A1,A2, A3,B> {
//     B apply(A1 arg1, A2 arg2, A3 arg3);
// }


import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.google.common.collect.Table;
import com.yammer.metrics.core.Gauge;
import kafka.cluster.Broker;
import kafka.common.TopicAndPartition;
import kafka.metrics.KafkaMetricsGroup;
import kafka.utils.Callable1;
import kafka.utils.Callable2;
import kafka.utils.Function2;
import kafka.utils.Function3;
import kafka.utils.Tuple2;
import kafka.utils.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Map;
import java.util.Set;
        logger.info("Removed fetcher for partitions {}", partitions);
    }

    public void shutdownIdleFetcherThreads() {
        synchronized (mapLock) {
            final Set<BrokerAndFetcherId> keysToBeRemoved = Sets.newHashSet();
            Utils.foreach(fetcherThreadMap, new Callable2<BrokerAndFetcherId, AbstractFetcherThread>() {
                @Override
                public void apply(BrokerAndFetcherId key, AbstractFetcherThread fetcher) {
                    if (fetcher.partitionCount() <= 0) {
                        fetcher.shutdown();
                        keysToBeRemoved.add(key);
                    }
                }
            });

            Utils.foreach(keysToBeRemoved, new Callable1<BrokerAndFetcherId>() {
                @Override
                public void apply(BrokerAndFetcherId brokerAndFetcherId) {
                    fetcherThreadMap.remove(brokerAndFetcherId);
                }
            });
        }
    }

    public void closeAllFetchers() {
        synchronized (mapLock) {
            Utils.foreach(fetcherThreadMap, new Callable2<BrokerAndFetcherId, AbstractFetcherThread>() {
                @Override
                public void apply(BrokerAndFetcherId _, AbstractFetcherThread fetcher) {
// 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/utils/Function2.java
// public interface Function2<A1,A2,B> {
//     B apply(A1 arg1, A2 arg2);
// }
// 
// Path: src/main/java/kafka/utils/Function3.java
// public interface Function3<A1,A2, A3,B> {
//     B apply(A1 arg1, A2 arg2, A3 arg3);
// }


// Path: src/main/java/kafka/server/AbstractFetcherManager.java
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.google.common.collect.Table;
import com.yammer.metrics.core.Gauge;
import kafka.cluster.Broker;
import kafka.common.TopicAndPartition;
import kafka.metrics.KafkaMetricsGroup;
import kafka.utils.Callable1;
import kafka.utils.Callable2;
import kafka.utils.Function2;
import kafka.utils.Function3;
import kafka.utils.Tuple2;
import kafka.utils.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Map;
import java.util.Set;

        logger.info("Removed fetcher for partitions {}", partitions);
    }

    public void shutdownIdleFetcherThreads() {
        synchronized (mapLock) {
            final Set<BrokerAndFetcherId> keysToBeRemoved = Sets.newHashSet();
            Utils.foreach(fetcherThreadMap, new Callable2<BrokerAndFetcherId, AbstractFetcherThread>() {
                @Override
                public void apply(BrokerAndFetcherId key, AbstractFetcherThread fetcher) {
                    if (fetcher.partitionCount() <= 0) {
                        fetcher.shutdown();
                        keysToBeRemoved.add(key);
                    }
                }
            });

            Utils.foreach(keysToBeRemoved, new Callable1<BrokerAndFetcherId>() {
                @Override
                public void apply(BrokerAndFetcherId brokerAndFetcherId) {
                    fetcherThreadMap.remove(brokerAndFetcherId);
                }
            });
        }
    }

    public void closeAllFetchers() {
        synchronized (mapLock) {
            Utils.foreach(fetcherThreadMap, new Callable2<BrokerAndFetcherId, AbstractFetcherThread>() {
                @Override
                public void apply(BrokerAndFetcherId _, AbstractFetcherThread fetcher) {
                    fetcher.shutdown();