// Path: src/main/java/kafka/network/Receive.java
// public abstract class Receive extends Transmission {
//     Logger logger = LoggerFactory.getLogger(Receive.class);
// 
//     public abstract ByteBuffer buffer();
// 
//     public abstract int readFrom(ReadableByteChannel channel);
// 
// 
//     public int readCompletely(ReadableByteChannel channel) {
//         int totalRead = 0;
//         while (!complete()) {
//             int read = readFrom(channel);
//             logger.trace("{} bytes read.", read);
//             totalRead += read;
//         }
//         return totalRead;
//     }
// }


import kafka.api.*;
import kafka.metrics.KafkaTimer;
import kafka.network.BlockingChannel;
import kafka.network.BoundedByteBufferSend;
import kafka.network.Receive;
import kafka.utils.Function0;
import kafka.utils.ThreadSafe;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Random;

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

    private Object lock = new Object();
    volatile private boolean shutdown = false;
    private BlockingChannel blockingChannel;
    public String brokerInfo;
    public ProducerRequestStats producerRequestStats;


    private void verifyRequest(RequestOrResponse request) {
        /**
         * This seems a little convoluted, but the idea is to turn on verification simply changing log4j settings
         * Also, when verification is turned on, care should be taken to see that the logs don't fill up with unnecessary
         * data. So, leaving the rest of the logging at TRACE, while errors should be logged at ERROR level
         */
        if (logger.isDebugEnabled()) {
            ByteBuffer buffer = new BoundedByteBufferSend(request).buffer;
            logger.trace("verifying sendbuffer of size " + buffer.limit());
            short requestTypeId = buffer.getShort();
            if (requestTypeId == RequestKeys.ProduceKey) {
                RequestOrResponse request1 = ProducerRequestReader.instance.readFrom(buffer);
                logger.trace(request1.toString());
            }
        }
    }

    /**
     * Common functionality for the public send methods
     */
// Path: src/main/java/kafka/network/Receive.java
// public abstract class Receive extends Transmission {
//     Logger logger = LoggerFactory.getLogger(Receive.class);
// 
//     public abstract ByteBuffer buffer();
// 
//     public abstract int readFrom(ReadableByteChannel channel);
// 
// 
//     public int readCompletely(ReadableByteChannel channel) {
//         int totalRead = 0;
//         while (!complete()) {
//             int read = readFrom(channel);
//             logger.trace("{} bytes read.", read);
//             totalRead += read;
//         }
//         return totalRead;
//     }
// }


// Path: src/main/java/kafka/producer/SyncProducer.java
import kafka.api.*;
import kafka.metrics.KafkaTimer;
import kafka.network.BlockingChannel;
import kafka.network.BoundedByteBufferSend;
import kafka.network.Receive;
import kafka.utils.Function0;
import kafka.utils.ThreadSafe;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Random;


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

    private Object lock = new Object();
    volatile private boolean shutdown = false;
    private BlockingChannel blockingChannel;
    public String brokerInfo;
    public ProducerRequestStats producerRequestStats;


    private void verifyRequest(RequestOrResponse request) {
        /**
         * This seems a little convoluted, but the idea is to turn on verification simply changing log4j settings
         * Also, when verification is turned on, care should be taken to see that the logs don't fill up with unnecessary
         * data. So, leaving the rest of the logging at TRACE, while errors should be logged at ERROR level
         */
        if (logger.isDebugEnabled()) {
            ByteBuffer buffer = new BoundedByteBufferSend(request).buffer;
            logger.trace("verifying sendbuffer of size " + buffer.limit());
            short requestTypeId = buffer.getShort();
            if (requestTypeId == RequestKeys.ProduceKey) {
                RequestOrResponse request1 = ProducerRequestReader.instance.readFrom(buffer);
                logger.trace(request1.toString());
            }
        }
    }

    /**
     * Common functionality for the public send methods
     */
    private Receive doSend(RequestOrResponse request) {