Heaven.Blog

// 配置集群地址

props.put(“bootstrap.servers”, “linux1:9092”);

// ACK应答

props.put(“acks”, “all”);

// 重试次数

props.put(“retries”, 1);

// 批次大小 16K

props.put(“batch.size”, 16384);

props.put(“linger.ms”, 1);//等待时间

// 缓冲区内存大小 32M

props.put(“buffer.memory”, 33554432);//RecordAccumulator缓冲区大小

// Kafka发送数据时以k-v键值对方式发送

props.put(“key.serializer”, “org.apache.kafka.common.serialization.StringSerializer”);

props.put(“value.serializer”, “org.apache.kafka.common.serialization.StringSerializer”);

props.put(“interceptor.classes”, “com.atguigu.itdachang.kafka.MyProducerInterceptor”);

try {

this.producerConfig = config;

this.time = time;

String transactionalId = config.getString(“transactional.id”);

this.clientId = config.getString(“client.id”);

LogContext logContext;

if (transactionalId == null) {

logContext = new LogContext(String.format(“[Producer clientId=%s] “, this.clientId));

} else {

logContext = new LogContext(String.format(“[Producer clientId=%s, transactionalId=%s] “, this.clientId, transactionalId));

}

this.log = logContext.logger(KafkaProducer.class);

this.log.trace(“Starting the Kafka producer”);

Map<String, String> metricTags = Collections.singletonMap(“client-id”, this.clientId);

MetricConfig metricConfig = (new MetricConfig()).samples(config.getInt(“metrics.num.samples”)).timeWindow(config.getLong(“metrics.sample.window.ms”), TimeUnit.MILLISECONDS).recordLevel(RecordingLevel.forName(config.getString(“metrics.recording.level”))).tags(metricTags);

List<MetricsReporter> reporters = config.getConfiguredInstances(“metric.reporters”, MetricsReporter.class, Collections.singletonMap(“client.id”, this.clientId));

JmxReporter jmxReporter = new JmxReporter();

jmxReporter.configure(config.originals(Collections.singletonMap(“client.id”, this.clientId)));

reporters.add(jmxReporter);

MetricsContext metricsContext = new KafkaMetricsContext(“kafka.producer”, config.originalsWithPrefix(“metrics.context.”));

this.metrics = new Metrics(metricConfig, reporters, time, metricsContext);

this.partitioner = (Partitioner)config.getConfiguredInstance(“partitioner.class”, Partitioner.class, Collections.singletonMap(“client.id”, this.clientId));

long retryBackoffMs = config.getLong(“retry.backoff.ms”);

if (keySerializer == null) {

this.keySerializer = (Serializer)config.getConfiguredInstance(“key.serializer”, Serializer.class);

this.keySerializer.configure(config.originals(Collections.singletonMap(“client.id”, this.clientId)), true);

} else {

config.ignore(“key.serializer”);

this.keySerializer = keySerializer;

}

if (valueSerializer == null) {

this.valueSerializer = (Serializer)config.getConfiguredInstance(“value.serializer”, Serializer.class);

this.valueSerializer.configure(config.originals(Collections.singletonMap(“client.id”, this.clientId)), false);

} else {

config.ignore(“value.serializer”);

this.valueSerializer = valueSerializer;

}

List<ProducerInterceptor<K, V>> interceptorList = config.getConfiguredInstances(“interceptor.classes”, ProducerInterceptor.class, Collections.singletonMap(“client.id”, this.clientId));

if (interceptors != null) {

this.interceptors = interceptors;

} else {

this.interceptors = new ProducerInterceptors(interceptorList);

}

ClusterResourceListeners clusterResourceListeners = this.configureClusterResourceListeners(keySerializer, valueSerializer, interceptorList, reporters);

this.maxRequestSize = config.getInt(“max.request.size”);

this.totalMemorySize = config.getLong(“buffer.memory”);

this.compressionType = CompressionType.forName(config.getString(“compression.type”));

this.maxBlockTimeMs = config.getLong(“max.block.ms”);

int deliveryTimeoutMs = configureDeliveryTimeout(config, this.log);

this.apiVersions = new ApiVersions();

this.transactionManager = this.configureTransactionState(config, logContext);

this.accumulator = new RecordAccumulator(logContext, config.getInt(“batch.size”), this.compressionType, lingerMs(config), retryBackoffMs, deliveryTimeoutMs, this.metrics, “producer-metrics”, time, this.apiVersions, this.transactionManager, new BufferPool(this.totalMemorySize, config.getInt(“batch.size”), this.metrics, time, “producer-metrics”));

List<InetSocketAddress> addresses = ClientUtils.parseAndValidateAddresses(config.getList(“bootstrap.servers”), config.getString(“client.dns.lookup”));

if (metadata != null) {

this.metadata = metadata;

} else {

this.metadata = new ProducerMetadata(retryBackoffMs, config.getLong(“metadata.max.age.ms”), config.getLong(“metadata.max.idle.ms”), logContext, clusterResourceListeners, Time.SYSTEM);

this.metadata.bootstrap(addresses);

}

this.errors = this.metrics.sensor(“errors”);

this.sender = this.newSender(logContext, kafkaClient, this.metadata);

String ioThreadName = “kafka-producer-network-thread | ” + this.clientId;

this.ioThread = new KafkaThread(ioThreadName, this.sender, true);

this.ioThread.start();

config.logUnused();

AppInfoParser.registerAppInfo(“kafka.producer”, this.clientId, this.metrics, time.milliseconds());

this.log.debug(“Kafka producer started”);

} catch (Throwable var22) {

this.close(Duration.ofMillis(0L), true);

throw new KafkaException(“Failed to construct kafka producer”, var22);

}

}

private final Integer partition;

private final Headers headers;

private final K key;

private final V value;

private final Long timestamp;

public class MyProducerInterceptor implements ProducerInterceptor {

@Override

public ProducerRecord onSend(ProducerRecord oldRecord) {

ProducerRecord newRecord = new ProducerRecord(oldRecord.topic(), oldRecord.key(), “test:” + oldRecord.value());

return newRecord;

}

}

clusterAndWaitTime = this.waitOnMetadata(record.topic(), record.partition(), nowMs, this.maxBlockTimeMs);

} catch (KafkaException var22) {

if (this.metadata.isClosed()) {

throw new KafkaException(“Producer closed while send in progress”, var22);

}

throw var22;

}

try {

serializedKey = this.keySerializer.serialize(record.topic(), record.headers(), record.key());

} catch (ClassCastException var21) {

throw new SerializationException(“Can’t convert key of class ” + record.key().getClass().getName() + ” to class ” + this.producerConfig.getClass(“key.serializer”).getName() + ” specified in key.serializer”, var21);

}

byte[] serializedValue;

try {

serializedValue = this.valueSerializer.serialize(record.topic(), record.headers(), record.value());

} catch (ClassCastException var20) {

throw new SerializationException(“Can’t convert value of class ” + record.value().getClass().getName() + ” to class ” + this.producerConfig.getClass(“value.serializer”).getName() + ” specified in value.serializer”, var20);

}

tp = new TopicPartition(record.topic(), partition);

this.ensureValidRecordSize(serializedSize);

private final List<Node> nodes;

private final Set<String> unauthorizedTopics;

private final Set<String> invalidTopics;

private final Set<String> internalTopics;

private final Node controller;

private final Map<TopicPartition, PartitionInfo> partitionsByTopicPartition;

private final Map<String, List<PartitionInfo>> partitionsByTopic;

private final Map<String, List<PartitionInfo>> availablePartitionsByTopic;

private final Map<Integer, List<PartitionInfo>> partitionsByNode;

private final Map<Integer, Node> nodesById;

private final ClusterResource clusterResource;

private final Map<String, Uuid> topicIds;

private final long refreshBackoffMs;

private final long metadataExpireMs;

private int updateVersion;

private int requestVersion;

private long lastRefreshMs;

private long lastSuccessfulRefreshMs;

private KafkaException fatalException;

private Set<String> invalidTopics;

private Set<String> unauthorizedTopics;

private MetadataCache cache = MetadataCache.empty();

private boolean needFullUpdate;

private boolean needPartialUpdate;

private final ClusterResourceListeners clusterResourceListeners;

private boolean isClosed;

private final Map<TopicPartition, Integer> lastSeenLeaderEpochs;

private final long metadataIdleMs;

private final Map<String, Long> topics = new HashMap();

private final Set<String> newTopics = new HashSet();

private final Logger log;

private final Time time;

throw new InvalidTopicException(topic);

}

Objects.requireNonNull(topic, “topic cannot be null”);

if (this.topics.put(topic, nowMs + this.metadataIdleMs) == null) {

this.newTopics.add(topic);

this.requestUpdateForNewTopics();

}

}

public synchronized int requestUpdateForNewTopics() {

this.lastRefreshMs = 0L;

this.needPartialUpdate = true;

++this.requestVersion;

return this.updateVersion;

}

return new KafkaProducer.ClusterAndWaitTime(cluster, 0L);

}

do {

} while(partitionsCount == null);

} while(partition != null && partition >= partitionsCount);

this.handleCompletedSends(responses, updatedNow);

this.handleCompletedReceives(responses, updatedNow);

this.handleDisconnections(responses, updatedNow);

this.handleConnections();

this.handleInitiateApiVersionRequests(updatedNow);

this.handleTimedOutConnections(responses, updatedNow);

this.handleTimedOutRequests(responses, updatedNow);

this.completeResponses(responses);

return responses;

long timeToExpire = this.updateRequested() ? 0L : Math.max(this.lastSuccessfulRefreshMs + this.metadataExpireMs – nowMs, 0L);

return Math.max(timeToExpire, this.timeToAllowUpdate(nowMs));

}

MetadataRequestAndVersion requestAndVersion = this.metadata.newMetadataRequestAndVersion(now);

org.apache.kafka.common.requests.MetadataRequest.Builder metadataRequest = requestAndVersion.requestBuilder;

NetworkClient.this.log.debug(“Sending metadata request {} to node {}”, metadataRequest, node);

NetworkClient.this.sendInternalMetadataRequest(metadataRequest, nodeConnectionId, now);

this.inProgress = new NetworkClient.DefaultMetadataUpdater.InProgressData(requestAndVersion.requestVersion, requestAndVersion.isPartialUpdate, (SyntheticClass_1)null);

return (long)NetworkClient.this.defaultRequestTimeoutMs;

return NetworkClient.this.reconnectBackoffMs;

} else if (NetworkClient.this.connectionStates.canConnect(nodeConnectionId, now)) {

NetworkClient.this.log.debug(“Initialize connection to node {} for sending metadata request”, node);

NetworkClient.this.initiateConnect(node, now);

return NetworkClient.this.reconnectBackoffMs;

}

while(var4.hasNext()) {

NetworkReceive receive = (NetworkReceive)var4.next();

String source = receive.source();

NetworkClient.InFlightRequest req = this.inFlightRequests.completeNext(source);

AbstractResponse response = parseResponse(receive.payload(), req.header);

if (this.throttleTimeSensor != null) {

this.throttleTimeSensor.record((double)response.throttleTimeMs(), now);

}

if (this.log.isDebugEnabled()) {

this.log.debug(“Received {} response from node {} for request with header {}: {}”, new Object[]{req.header.apiKey(), req.destination, req.header, response});

}

this.maybeThrottle(response, req.header.apiVersion(), req.destination, now);

if (req.isInternalRequest && response instanceof MetadataResponse) {

this.metadataUpdater.handleSuccessfulResponse(req.header, now, (MetadataResponse)response);

} else if (req.isInternalRequest && response instanceof ApiVersionsResponse) {

this.handleApiVersionsResponse(responses, req, now, (ApiVersionsResponse)response);

} else {

responses.add(req.completed(response, now));

}

}

return;

}

this.metadataUpdater.handleSuccessfulResponse(req.header, now, (MetadataResponse)response);

} else if (req.isInternalRequest && response instanceof ApiVersionsResponse) {

this.handleApiVersionsResponse(responses, req, now, (ApiVersionsResponse)response);

} else {

responses.add(req.completed(response, now));

}

return keyBytes == null ? this.stickyPartitionCache.partition(topic, cluster) : Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;

}

ProducerBatch last = (ProducerBatch)deque.peekLast();

if (last != null) {

FutureRecordMetadata future = last.tryAppend(timestamp, key, value, headers, callback, nowMs);

if (future != null) {

return new RecordAccumulator.RecordAppendResult(future, deque.size() > 1 || last.isFull(), false, false);

}

last.closeForRecordAppends();

}

return null;

}

if (!this.recordsBuilder.hasRoomFor(timestamp, key, value, headers)) {

return null;

} else {

Long checksum = this.recordsBuilder.append(timestamp, key, value, headers);

this.maxRecordSize = Math.max(this.maxRecordSize, AbstractRecords.estimateSizeInBytesUpperBound(this.magic(), this.recordsBuilder.compressionType(), key, value, headers));

this.lastAppendTime = now;

FutureRecordMetadata future = new FutureRecordMetadata(this.produceFuture, (long)this.recordCount, timestamp, checksum, key == null ? -1 : key.length, value == null ? -1 : value.length, Time.SYSTEM);

this.thunks.add(new ProducerBatch.Thunk(callback, future));

++this.recordCount;

return future;

}

}

RecordAccumulator.RecordAppendResult var31 = new RecordAccumulator.RecordAppendResult((FutureRecordMetadata)null, false, false, true);

return var31;

}

int prevPartition = partition;

this.partitioner.onNewBatch(record.topic(), cluster, partition);

partition = this.partition(record, serializedKey, serializedValue, cluster);

tp = new TopicPartition(record.topic(), partition);

if (this.log.isTraceEnabled()) {

this.log.trace(“Retrying append due to new batch creation for topic {} partition {}. The old partition was {}”, new Object[]{record.topic(), partition, prevPartition});

}

interceptCallback = new KafkaProducer.InterceptorCallback(callback, this.interceptors, tp);

result = this.accumulator.append(tp, timestamp, serializedKey, serializedValue, headers, interceptCallback, remainingWaitMs, false, nowMs);

}

this.log.trace(“Waking up the sender since topic {} partition {} is either full or getting a new batch”, record.topic(), partition);

this.sender.wakeup();

}

if (nodes.isEmpty()) {

return Collections.emptyMap();

} else {

Map<Integer, List<ProducerBatch>> batches = new HashMap();

Iterator var7 = nodes.iterator();

while(var7.hasNext()) {

Node node = (Node)var7.next();

List<ProducerBatch> ready = this.drainBatchesForOneNode(cluster, node, maxSize, now);

batches.put(node.id(), ready);

}

return batches;

}

}

Iterator var4 = collated.entrySet().iterator();

while(var4.hasNext()) {

Entry<Integer, List<ProducerBatch>> entry = (Entry)var4.next();

this.sendProduceRequest(now, (Integer)entry.getKey(), this.acks, this.requestTimeoutMs, (List)entry.getValue());

}

}

if (!batches.isEmpty()) {

Map<TopicPartition, ProducerBatch> recordsByPartition = new HashMap(batches.size());

byte minUsedMagic = this.apiVersions.maxUsableProduceMagic();

Iterator var9 = batches.iterator();

while(var9.hasNext()) {

ProducerBatch batch = (ProducerBatch)var9.next();

if (batch.magic() < minUsedMagic) {

minUsedMagic = batch.magic();

}

}

TopicProduceDataCollection tpd = new TopicProduceDataCollection();

Iterator var16 = batches.iterator();

while(var16.hasNext()) {

ProducerBatch batch = (ProducerBatch)var16.next();

TopicPartition tp = batch.topicPartition;

MemoryRecords records = batch.records();

if (!records.hasMatchingMagic(minUsedMagic)) {

records = (MemoryRecords)batch.records().downConvert(minUsedMagic, 0L, this.time).records();

}

TopicProduceData tpData = tpd.find(tp.topic());

if (tpData == null) {

tpData = (new TopicProduceData()).setName(tp.topic());

tpd.add(tpData);

}

tpData.partitionData().add((new PartitionProduceData()).setIndex(tp.partition()).setRecords(records));

recordsByPartition.put(tp, batch);

}

String transactionalId = null;

if (this.transactionManager != null && this.transactionManager.isTransactional()) {

transactionalId = this.transactionManager.transactionalId();

}

org.apache.kafka.common.requests.ProduceRequest.Builder requestBuilder = ProduceRequest.forMagic(minUsedMagic, (new ProduceRequestData()).setAcks(acks).setTimeoutMs(timeout).setTransactionalId(transactionalId).setTopicData(tpd));

RequestCompletionHandler callback = (response) -> {

this.handleProduceResponse(response, recordsByPartition, this.time.milliseconds());

};

String nodeId = Integer.toString(destination);

ClientRequest clientRequest = this.client.newClientRequest(nodeId, requestBuilder, now, acks != 0, this.requestTimeoutMs, callback);

this.client.send(clientRequest, now);

this.log.trace(“Sent produce request to {}: {}”, nodeId, requestBuilder);

}

}

this.log.trace(“Cancelled request with header {} due to node {} being disconnected”, requestHeader, response.destination());

var7 = batches.values().iterator();

while(var7.hasNext()) {

batch = (ProducerBatch)var7.next();

this.completeBatch(batch, new PartitionResponse(Errors.NETWORK_EXCEPTION, String.format(“Disconnected from node %s”, response.destination())), (long)correlationId, now);

}

} else if (response.versionMismatch() != null) {

this.log.warn(“Cancelled request {} due to a version mismatch with node {}”, new Object[]{response, response.destination(), response.versionMismatch()});

var7 = batches.values().iterator();

while(var7.hasNext()) {

batch = (ProducerBatch)var7.next();

this.completeBatch(batch, new PartitionResponse(Errors.UNSUPPORTED_VERSION), (long)correlationId, now);

}

}

if (this.transactionManager != null) {

this.transactionManager.removeInFlightBatch(batch);

}

this.accumulator.splitAndReenqueue(batch);

this.maybeRemoveAndDeallocateBatch(batch);

this.sensors.recordBatchSplit();

}

if (this.canRetry(batch, response, now)) {

this.log.warn(“Got error produce response with correlation id {} on topic-partition {}, retrying ({} attempts left). Error: {}”, new Object[]{correlationId, batch.topicPartition, this.retries – batch.attempts() – 1, this.formatErrMsg(response)});

this.reenqueueBatch(batch, now);

} else if (error == Errors.DUPLICATE_SEQUENCE_NUMBER) {

this.completeBatch(batch, response);

} else {

Object exception;

if (error == Errors.TOPIC_AUTHORIZATION_FAILED) {

exception = new TopicAuthorizationException(Collections.singleton(batch.topicPartition.topic()));

} else if (error == Errors.CLUSTER_AUTHORIZATION_FAILED) {

exception = new ClusterAuthorizationException(“The producer is not authorized to do idempotent sends”);

} else {

exception = error.exception(response.errorMessage);

}

this.failBatch(batch, response, (RuntimeException)exception, batch.attempts() < this.retries);

}

if (error.exception() instanceof InvalidMetadataException) {

if (error.exception() instanceof UnknownTopicOrPartitionException) {

this.log.warn(“Received unknown topic or partition error in produce request on partition {}. The topic-partition may not exist or the user may not have Describe access to it”, batch.topicPartition);

} else {

this.log.warn(“Received invalid metadata error in produce request on partition {} due to {}. Going to request metadata update now”, batch.topicPartition, error.exception(response.errorMessage).toString());

}

this.metadata.requestUpdate();

}

val info = partition.appendRecordsToLeader(records, origin, requiredAcks)