Kafka 常见问题及方案总结

public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
    // 发送前执行处理器包装record
    ProducerRecord<K, V> interceptedRecord = this.interceptors.onSend(record);
    return doSend(interceptedRecord, callback);
}


private Future<RecordMetadata> doSend(ProducerRecord<K, V> record, Callback callback) {
    // 省略......
    try {
        // 省略......
        byte[] serializedKey;
        try {
            // 将 key 序列化成字节数组
            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 {
            // 将 value 序列化成字节数组
            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);
        }
        // 选择分区
        int partition = partition(record, serializedKey, serializedValue, cluster);
        // 省略......
        RecordAccumulator.RecordAppendResult result = accumulator.append(record.topic(), partition, timestamp, serializedKey,
                serializedValue, headers, appendCallbacks, remainingWaitMs, abortOnNewBatch, nowMs, cluster);
        // 省略......
    } catch (ApiException e) {
        // 省略......
    } 
}

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;
    }
    // 如果 key 不为空 且配置项 partitionerIgnoreKeys 为 false
    if (serializedKey != null && !partitionerIgnoreKeys) {
        // 通过计算 key 的 hash 值并与分区数量取模得出分区
        return BuiltInPartitioner.partitionForKey(serializedKey, cluster.partitionsForTopic(record.topic()).size());
    } else {
        // partition、partitioner、key 都没指定就返回未知分区，后续默认使用随机分区策略
        return RecordMetadata.UNKNOWN_PARTITION;
    }
}

try { 
    producer.send(record).get();
} catch (Throwable e) {
    // 重试或告警
    handle(record, e)
}

// 消费相关部分源码
public void maybeAutoCommitOffsetsAsync(long now) {
    // 是否开启offset自动提交
    if (autoCommitEnabled) {
        nextAutoCommitTimer.update(now);
        // 距离上次提交是否超过配置间隔
        if (nextAutoCommitTimer.isExpired()) {
            // 重置时间
            nextAutoCommitTimer.reset(autoCommitIntervalMs);
            // 异步提交
            autoCommitOffsetsAsync();
        }
    }
}

update sms_record set status = 1 where id = #{id} and status = 0;

update t set version = version + 1 where id = #{id} and version = #{version};