/**  
 * See {@link KafkaProducer#send(ProducerRecord)}  
 */Future<RecordMetadata> send(ProducerRecord<K, V> record);  
  
/**  支持回调方法
 * See {@link KafkaProducer#send(ProducerRecord, Callback)}  
 */Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback);

public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "172.28.203.172:9092");
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        Producer<String, String> producer = new KafkaProducer<>(props);
        for(int i = 0; i < 100; i++) {
            producer.send(new ProducerRecord<String, String>("test", Integer.toString(i), Integer.toString(i)));
        }

        producer.close();
    }

@Override  
public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {  
    // 拦截器前置send动作  
    ProducerRecord<K, V> interceptedRecord = this.interceptors.onSend(record);  
    return doSend(interceptedRecord, callback);  
}

private Future<RecordMetadata> doSend(ProducerRecord<K, V> record, Callback callback) {  
    // 1.1 创建callback对象  
    AppendCallbacks appendCallbacks = new AppendCallbacks(callback, this.interceptors, record);  
  
    try {  
        //1.2 检查producer是否被close  
        throwIfProducerClosed();  
        // first make sure the metadata for the topic is available  
        long nowMs = time.milliseconds();  
        ClusterAndWaitTime clusterAndWaitTime;  
  
        //1.3 拉取指定topic、分区的元数据，和等待时间  
        try {  
            clusterAndWaitTime = waitOnMetadata(record.topic(), record.partition(), nowMs, maxBlockTimeMs);  
        } catch (KafkaException e) {  
            if (metadata.isClosed())  
                throw new KafkaException("Producer closed while send in progress", e);  
            throw e;  
        }  
        nowMs += clusterAndWaitTime.waitedOnMetadataMs;  
        long remainingWaitMs = Math.max(0, maxBlockTimeMs - clusterAndWaitTime.waitedOnMetadataMs);  
        Cluster cluster = clusterAndWaitTime.cluster;  
  
        //1.4 key value进行序列化  
        byte[] serializedKey;  
        try {  
            serializedKey = keySerializer.serialize(record.topic(), record.headers(), record.key());  
        } catch (ClassCastException cce) {  
            throw new SerializationException("Can't convert key of class " + record.key().getClass().getName() +  
                    " to class " + producerConfig.getClass(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG).getName() +  
                    " specified in key.serializer", cce);  
        }  
        byte[] serializedValue;  
        try {  
            serializedValue = valueSerializer.serialize(record.topic(), record.headers(), record.value());  
        } catch (ClassCastException cce) {  
            throw new SerializationException("Can't convert value of class " + record.value().getClass().getName() +  
                    " to class " + producerConfig.getClass(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG).getName() +  
                    " specified in value.serializer", cce);  
        }  
  
        // 1.5 计算当前消息所属的partition  
        int partition = partition(record, serializedKey, serializedValue, cluster);  
  
        // 1.6 设置消息header为readOnly  
        setReadOnly(record.headers());  
        Header[] headers = record.headers().toArray();  
  
        //1.7 检查消息大小是否符合  
        int serializedSize = AbstractRecords.estimateSizeInBytesUpperBound(apiVersions.maxUsableProduceMagic(),  
                compression.type(), serializedKey, serializedValue, headers);  
        ensureValidRecordSize(serializedSize);  
        long timestamp = record.timestamp() == null ? nowMs : record.timestamp();  
  
        // 自定义partitioner  
        boolean abortOnNewBatch = partitioner != null;  
  
        // 1.8 将消息追加到accumulator中  
        RecordAccumulator.RecordAppendResult result = accumulator.append(record.topic(), partition, timestamp, serializedKey,  
                serializedValue, headers, appendCallbacks, remainingWaitMs, abortOnNewBatch, nowMs, cluster);  
        assert appendCallbacks.getPartition() != RecordMetadata.UNKNOWN_PARTITION;  
  
        // 1.9 消息入新batch的情况  
        if (result.abortForNewBatch) {  
            int prevPartition = partition;  
            onNewBatch(record.topic(), cluster, prevPartition);  
            partition = partition(record, serializedKey, serializedValue, cluster);  
            if (log.isTraceEnabled()) {  
                log.trace("Retrying append due to new batch creation for topic {} partition {}. The old partition was {}", record.topic(), partition, prevPartition);  
            }  
            result = accumulator.append(record.topic(), partition, timestamp, serializedKey,  
                serializedValue, headers, appendCallbacks, remainingWaitMs, false, nowMs, cluster);  
        }  
  
        // 2.1 开启事务的情况  
        if (transactionManager != null) {  
            transactionManager.maybeAddPartition(appendCallbacks.topicPartition());  
        }  
  
        // 2.2 如果batch满了，或者新batch被创建，唤醒后台sender线程  
        if (result.batchIsFull || result.newBatchCreated) {  
            log.trace("Waking up the sender since topic {} partition {} is either full or getting a new batch", record.topic(), appendCallbacks.getPartition());  
            this.sender.wakeup();  
        }  
        return result.future;  
        // handling exceptions and record the errors;  
        // for API exceptions return them in the future,        // for other exceptions throw directly    } catch (ApiException e) {  
        log.debug("Exception occurred during message send:", e);  
        if (callback != null) {  
            TopicPartition tp = appendCallbacks.topicPartition();  
            RecordMetadata nullMetadata = new RecordMetadata(tp, -1, -1, RecordBatch.NO_TIMESTAMP, -1, -1);  
            callback.onCompletion(nullMetadata, e);  
        }  
        this.errors.record();  
        this.interceptors.onSendError(record, appendCallbacks.topicPartition(), e);  
        if (transactionManager != null) {  
            transactionManager.maybeTransitionToErrorState(e);  
        }  
        return new FutureFailure(e);  
    } catch (InterruptedException e) {  
        this.errors.record();  
        this.interceptors.onSendError(record, appendCallbacks.topicPartition(), e);  
        throw new InterruptException(e);  
    } catch (KafkaException e) {  
        this.errors.record();  
        this.interceptors.onSendError(record, appendCallbacks.topicPartition(), e);  
        throw e;  
    } catch (Exception e) {  
        // we notify interceptor about all exceptions, since onSend is called before anything else in this method  
        this.interceptors.onSendError(record, appendCallbacks.topicPartition(), e);  
        throw e;  
    }  
}

this.sender = newSender(logContext, kafkaClient, this.metadata);  
String ioThreadName = NETWORK_THREAD_PREFIX + " | " + clientId;  
this.ioThread = new KafkaThread(ioThreadName, this.sender, true);  
this.ioThread.start();

while (running) {  
    try {  
        runOnce();  
    } catch (Exception e) {  
        log.error("Uncaught error in kafka producer I/O thread: ", e);  
    }  
}

void runOnce() {  
    ……
    //省略事务处理

	//创建发送给broker的请求并发送
    long currentTimeMs = time.milliseconds();
    long pollTimeout = sendProducerData(currentTimeMs);
    //处理实际网络IO socket的入口，负责发送请求、接收响应
    client.poll(pollTimeout, currentTimeMs);
}

@Override  
public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {  
    // 拦截器前置send动作  
    ProducerRecord<K, V> interceptedRecord = this.interceptors.onSend(record);  
    return doSend(interceptedRecord, callback);  
}

private int partition(ProducerRecord<K, V> record, byte[] serializedKey, byte[] serializedValue, Cluster cluster) {  
    if (record.partition() != null)  
        return record.partition();  
  
    if (partitioner != null) {  
        int customPartition = partitioner.partition(  
            record.topic(), record.key(), serializedKey, record.value(), serializedValue, cluster);  
        if (customPartition < 0) {  
            throw new IllegalArgumentException(String.format(  
                "The partitioner generated an invalid partition number: %d. Partition number should always be non-negative.", customPartition));  
        }  
        return customPartition;  
    }  
  
    if (serializedKey != null && !partitionerIgnoreKeys) {  
        // hash the keyBytes to choose a partition  
        return BuiltInPartitioner.partitionForKey(serializedKey, cluster.partitionsForTopic(record.topic()).size());  
    } else {  
        return RecordMetadata.UNKNOWN_PARTITION;  
    }  
}

public void run() {  
    try {  
        log.debug("Consumer network thread started");  
  
        // Wait until we're securely in the background network thread to initialize these objects...  
        initializeResources();  
  
        while (running) {  
            try {  
                runOnce();  
            } catch (final Throwable e) {  
                // Swallow the exception and continue  
                log.error("Unexpected error caught in consumer network thread", e);  
            }  
        }  
    } finally {  
        cleanup();  
    }  
}

@Test
    public void testConsumer() {
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "localhost:9092");
        props.setProperty("group.id", "test");
        props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.setProperty("client.id","consumer.client.id.demo");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList(topic));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records)
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
        }
    }

public void subscribe(Collection<String> topics, ConsumerRebalanceListener listener) {  
    if (listener == null)  
        throw new IllegalArgumentException("RebalanceListener cannot be null");  
  
    subscribeInternal(topics, Optional.of(listener));  
}

private void subscribeInternal(Collection<String> topics, Optional<ConsumerRebalanceListener> listener) {  
    //1.1 获取lock，并且判断是否已经close  
    acquireAndEnsureOpen();  
    try {  
        //1.2 判断group id是否有效  
        maybeThrowInvalidGroupIdException();  
  
        //1.3 校验参数  
        if (topics == null)  
            throw new IllegalArgumentException("Topic collection to subscribe to cannot be null");  
  
        //1.4 若为空，则unsubscribe  
        if (topics.isEmpty()) {  
            // treat subscribing to empty topic list as the same as unsubscribing  
            unsubscribe();  

		} else {  
            for (String topic : topics) {  
                if (isBlank(topic))  
                    throw new IllegalArgumentException("Topic collection to subscribe to cannot contain null or empty topic");  
            }  
  
            // 1.5 更新buffer中不再指定的partition  
            final Set<TopicPartition> currentTopicPartitions = new HashSet<>();  
  
            for (TopicPartition tp : subscriptions.assignedPartitions()) {  
                if (topics.contains(tp.topic()))  
                    currentTopicPartitions.add(tp);  
            }  
  
            fetchBuffer.retainAll(currentTopicPartitions);  
            log.info("Subscribed to topic(s): {}", String.join(", ", topics));  
  
            // 1.6 调用SubscriptionState.subscribe 更新订阅topic  
            if (subscriptions.subscribe(new HashSet<>(topics), listener))  
                //若请求成功，更新metadata  
                this.metadataVersionSnapshot = metadata.requestUpdateForNewTopics();  
  
            // 1.7 向handler添加event  
            applicationEventHandler.add(new SubscriptionChangeEvent());  
        }  
    } finally {  
        //1.8 释放lock  
        release();  
    }  
}

// 日志消息文件对象，FileRecords是文件关联对象
private final FileRecords log;  

//三个索引文件对象
private final LazyIndex<OffsetIndex> lazyOffsetIndex;  
private final LazyIndex<TimeIndex> lazyTimeIndex;  
private final TransactionIndex txnIndex;

//起始offset
private final long baseOffset;

//broker端参数log.index.interval.bytes值，用于控制LogSegment新增索引项的频率，默认写入4KB时新增一条索引项
private final int indexIntervalBytes;  

//LogSegment新增数据的扰动值，打散磁盘IO
private final long rollJitterMs; 

//写入日志的最新时间戳
private volatile OptionalLong rollingBasedTimestamp = OptionalLong.empty();
  
//最后一次更新索引至今，写入的字节数 
private int bytesSinceLastIndexEntry = 0;

/**  
 * Append a set of records to the file. This method is not thread-safe and must be * protected with a lock. * * @param records The records to append  
 * @return the number of bytes written to the underlying file  
 */public int append(MemoryRecords records) throws IOException {  
    if (records.sizeInBytes() > Integer.MAX_VALUE - size.get())  
        throw new IllegalArgumentException("Append of size " + records.sizeInBytes() +  
                " bytes is too large for segment with current file position at " + size.get());  
  
    int written = records.writeFullyTo(channel);  
    size.getAndAdd(written);  
    return written;  
}  
  
/**  
 * Commit all written data to the physical disk */public void flush() throws IOException {  
    channel.force(true);  
}  
  
/**  
 * Close this record set */public void close() throws IOException {  
    flush();  
    trim();  
    channel.close();  
}

public static FileRecords open(File file,  
                               boolean mutable,  
                               boolean fileAlreadyExists,  
                               int initFileSize,  
                               boolean preallocate) throws IOException {  
    FileChannel channel = openChannel(file, mutable, fileAlreadyExists, initFileSize, preallocate);  
    int end = (!fileAlreadyExists && preallocate) ? 0 : Integer.MAX_VALUE;  
    return new FileRecords(file, channel, 0, end, false);  
}

public class LogSegment implements Closeable {
	private final LazyIndex<OffsetIndex> lazyOffsetIndex;  
	private final LazyIndex<TimeIndex> lazyTimeIndex;  
	private final TransactionIndex txnIndex;
}	

public T get() throws IOException {  
    IndexWrapper wrapper = indexWrapper;  
    if (wrapper instanceof IndexValue<?>)  
        return ((IndexValue<T>) wrapper).index;  
    else {  
        lock.lock();  
        try {  
            if (indexWrapper instanceof IndexValue<?>)  
                return ((IndexValue<T>) indexWrapper).index;  
            else if (indexWrapper instanceof IndexFile) {  
                IndexFile indexFile = (IndexFile) indexWrapper;  
                IndexValue<T> indexValue = new IndexValue<>(loadIndex(indexFile.file));  
                indexWrapper = indexValue;  
                return indexValue.index;  
            } else  
                throw new IllegalStateException("Unexpected type for indexWrapper " + indexWrapper.getClass());  
        } finally {  
            lock.unlock();  
        }  
    }  
}

private T loadIndex(File file) throws IOException {  
    switch (indexType) {  
        case OFFSET:  
            return (T) new OffsetIndex(file, baseOffset, maxIndexSize, true);  
        case TIME:  
            return (T) new TimeIndex(file, baseOffset, maxIndexSize, true);  
        default:  
            throw new IllegalStateException("Unexpected indexType " + indexType);  
    }  
}

private int indexSlotRangeFor(ByteBuffer idx, long target, IndexSearchType searchEntity,  
                              SearchResultType searchResultType) {  
    // check if the index is empty  
    if (entries == 0)  
        return -1;  
  
  
    //1.1 获取当前第一个热区entry offset  
    int firstHotEntry = Math.max(0, entries - 1 - warmEntries());  
    // 1.2 检查目标offset，是否在热区范围内  
    if (compareIndexEntry(parseEntry(idx, firstHotEntry), target, searchEntity) < 0) {  
        //1.3 缩减二分查找范围在热区范围内  
        return binarySearch(idx, target, searchEntity,  
            searchResultType, firstHotEntry, entries - 1);  
    }  
  
    //1.4 全量查找前检查offset是否小于最小offset  
    // check if the target offset is smaller than the least offset    if (compareIndexEntry(parseEntry(idx, 0), target, searchEntity) > 0) {  
        switch (searchResultType) {  
            case LARGEST_LOWER_BOUND:  
                return -1;  
            case SMALLEST_UPPER_BOUND:  
                return 0;  
        }  
    }  
  
    return binarySearch(idx, target, searchEntity, searchResultType, 0, firstHotEntry);  
}

protected final int warmEntries() {  
    return 8192 / entrySize();  
}

// data-plane  
private[network] val dataPlaneAcceptors = new ConcurrentHashMap[EndPoint, DataPlaneAcceptor]()  
val dataPlaneRequestChannel = new RequestChannel(maxQueuedRequests, DataPlaneAcceptor.MetricPrefix, time, apiVersionManager.newRequestMetrics)  
// control-plane  
private[network] var controlPlaneAcceptorOpt: Option[ControlPlaneAcceptor] = None  
val controlPlaneRequestChannelOpt: Option[RequestChannel] = config.controlPlaneListenerName.map(_ =>  
  new RequestChannel(20, ControlPlaneAcceptor.MetricPrefix, time, apiVersionManager.newRequestMetrics))

private[network] val processors = new ArrayBuffer[Processor]()

@Override  
public void run() {  
    int retries = 0;  
    int processedRecords = 0;  
    long remainingRecords = Long.MAX_VALUE;  
  
    // it is recommended to have a relatively short txn timeout in order to clear pending offsets faster  
    int transactionTimeoutMs = 10_000;  
    // consumer must be in read_committed mode, which means it won't be able to read uncommitted data  
    boolean readCommitted = true;  
  
    try (KafkaProducer<Integer, String> producer = new Producer("processor-producer", bootstrapServers, outputTopic,  
            true, transactionalId, true, -1, transactionTimeoutMs, null).createKafkaProducer();  
         KafkaConsumer<Integer, String> consumer = new Consumer("processor-consumer", bootstrapServers, inputTopic,  
             "processor-group", Optional.of(groupInstanceId), readCommitted, -1, null).createKafkaConsumer()) {  
        // called first and once to fence zombies and abort any pending transaction  
        producer.initTransactions();  
        consumer.subscribe(singleton(inputTopic), this);  
  
        Utils.printOut("Processing new records");  
        while (!closed && remainingRecords > 0) {  
            try {  
                ConsumerRecords<Integer, String> records = consumer.poll(ofMillis(200));  
                if (!records.isEmpty()) {  
                    // begin a new transaction session  
                    producer.beginTransaction();  
  
                    for (ConsumerRecord<Integer, String> record : records) {  
                        // process the record and send downstream  
                        ProducerRecord<Integer, String> newRecord =  
                            new ProducerRecord<>(outputTopic, record.key(), record.value() + "-ok");  
                        producer.send(newRecord);  
                    }  
  
                    // checkpoint the progress by sending offsets to group coordinator broker  
                    // note that this API is only available for broker >= 2.5                    
	            producer.sendOffsetsToTransaction(getOffsetsToCommit(consumer), consumer.groupMetadata());  
  
                    // commit the transaction including offsets  
                    producer.commitTransaction();  
                    processedRecords += records.count();  
                    retries = 0;  
                }  
            } catch (AuthorizationException | UnsupportedVersionException | ProducerFencedException  
                     | FencedInstanceIdException | OutOfOrderSequenceException | SerializationException e) {  
                // we can't recover from these exceptions  
                Utils.printErr(e.getMessage());  
                shutdown();  
            } catch (OffsetOutOfRangeException | NoOffsetForPartitionException e) {  
                // invalid or no offset found without auto.reset.policy  
                Utils.printOut("Invalid or no offset found, using latest");  
                consumer.seekToEnd(emptyList());  
                consumer.commitSync();  
                retries = 0;  
            } catch (KafkaException e) {  
                // abort the transaction  
                Utils.printOut("Aborting transaction: %s", e.getMessage());  
                producer.abortTransaction();  
                retries = maybeRetry(retries, consumer);  
            }  
  
            remainingRecords = getRemainingRecords(consumer);  
            if (remainingRecords != Long.MAX_VALUE) {  
                Utils.printOut("Remaining records: %d", remainingRecords);  
            }  
        }  
    } catch (Throwable e) {  
        Utils.printErr("Unhandled exception");  
        e.printStackTrace();  
    }  
    Utils.printOut("Processed %d records", processedRecords);  
    shutdown();  
}