// 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/SystemTime.java
// public class SystemTime implements Time {
//     public static final SystemTime instance = new SystemTime();
// 
//     @Override
//     public long milliseconds() {
//         return System.currentTimeMillis();
//     }
// 
//     @Override
//     public long nanoseconds() {
//         return System.nanoTime();
//     }
// 
//     @Override
//     public void sleep(long ms) {
//         try {
//             Thread.sleep(ms);
//         } catch (InterruptedException e) {
// 
//         }
//     }
// }


import com.google.common.collect.Lists;
import com.yammer.metrics.core.Gauge;
import kafka.common.KafkaException;
import kafka.metrics.KafkaMetricsGroup;
import kafka.utils.SystemTime;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
    /**
     * Close the connection for the request
     */
    public void closeConnection(int processor, Request request) {
        try {
            responseQueues[processor].put(new Response(processor, request, null, ResponseAction.CloseConnectionAction));
        } catch (InterruptedException e) {
            throw new KafkaException(e);
        }
        for (ResponseListener onResponse : responseListeners)
            onResponse.onResponse(processor);
    }

    /**
     * Get the next request or block until there is one
     */
    public Request receiveRequest() {
        try {
            return requestQueue.take();
        } catch (InterruptedException e) {
            throw new KafkaException(e);
        }
    }

    /**
     * Get a response for the given processor if there is one
     */
    public Response receiveResponse(int processor) {
        Response response = responseQueues[processor].poll();
        if (response != null)
// 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/SystemTime.java
// public class SystemTime implements Time {
//     public static final SystemTime instance = new SystemTime();
// 
//     @Override
//     public long milliseconds() {
//         return System.currentTimeMillis();
//     }
// 
//     @Override
//     public long nanoseconds() {
//         return System.nanoTime();
//     }
// 
//     @Override
//     public void sleep(long ms) {
//         try {
//             Thread.sleep(ms);
//         } catch (InterruptedException e) {
// 
//         }
//     }
// }


// Path: src/main/java/kafka/network/RequestChannel.java
import com.google.common.collect.Lists;
import com.yammer.metrics.core.Gauge;
import kafka.common.KafkaException;
import kafka.metrics.KafkaMetricsGroup;
import kafka.utils.SystemTime;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

    /**
     * Close the connection for the request
     */
    public void closeConnection(int processor, Request request) {
        try {
            responseQueues[processor].put(new Response(processor, request, null, ResponseAction.CloseConnectionAction));
        } catch (InterruptedException e) {
            throw new KafkaException(e);
        }
        for (ResponseListener onResponse : responseListeners)
            onResponse.onResponse(processor);
    }

    /**
     * Get the next request or block until there is one
     */
    public Request receiveRequest() {
        try {
            return requestQueue.take();
        } catch (InterruptedException e) {
            throw new KafkaException(e);
        }
    }

    /**
     * Get a response for the given processor if there is one
     */
    public Response receiveResponse(int processor) {
        Response response = responseQueues[processor].poll();
        if (response != null)
            response.request.responseDequeueTimeMs = SystemTime.instance.milliseconds();